DataFrame 是 Spark 在 RDD 之后新推出的一個(gè)數(shù)據(jù)集扫茅，從屬于 Spark SQL 模塊于微，適用于結(jié)構(gòu)化數(shù)據(jù)逗嫡。對于我們這些用慣了數(shù)據(jù)庫表和散列/字典結(jié)構(gòu)的人來說青自，尤為親切。
可以直接讀取關(guān)系型數(shù)據(jù)庫產(chǎn)生 DataFrame：
frompyspark.sqlimportSparkSessionspark = SparkSession \? ? ? ? .builder \? ? ? ? .appName("myapp") \? ? ? ? .config("spark.sql.shuffle.partitions",10) \? ? ? ? .getOrCreate()table ="(select * from users where province='jiangxi') as myusers"df = spark.read \? ? ? ? ? ? ? ? .format("jdbc") \? ? ? ? ? ? ? ? .option("driver","com.mysql.jdbc.Driver") \? ? ? ? ? ? ? ? .option("url","jdbc:mysql://"+dbhost+"/"+dbname) \? ? ? ? ? ? ? ? .option("dbtable", table) \? ? ? ? ? ? ? ? .option("user", dbuser) \? ? ? ? ? ? ? ? .option("password", dbpass) \? ? ? ? ? ? ? ? .option("partitionColumn", partitionColumn) \? ? ? ? ? ? ? ? .option("lowerBound", lowerBound) \? ? ? ? ? ? ? ? .option("upperBound", upperBound) \? ? ? ? ? ? ? ? .option("numPartitions","4") \? ? ? ? ? ? ? ? .load()
幾個(gè)選項(xiàng)說明一下：
url - 數(shù)據(jù)庫的 JDBC 連接串
dbtable - 可以是表名驱证，也可以是一個(gè)子查詢延窜。如果是子查詢的話，必須用括號括起來抹锄，并加別名逆瑞，參見上面代碼示例。
user - 數(shù)據(jù)庫用戶名
password - 數(shù)據(jù)庫密碼
partitionColumn - 用于并發(fā)分區(qū)的表字段伙单。下面幾個(gè)選項(xiàng)都是圍繞這個(gè)字段來的获高。Spark 會(huì)根據(jù)分區(qū)數(shù)量按這個(gè)字段的上下限把取出來的數(shù)據(jù)等分成幾份，并行處理车份。
lowerBound - 字段下限
upperBound - 字段上限
numPartitions - 分區(qū)數(shù)量
得到的 DataFrame谋减，可以通過字段名來引用列或者某行的值牡彻，如：
username = df.nameage = df["age"]
而不像以前用 RDD扫沼，老是 split 成元組，然后 0庄吼、1缎除、2、3... 地引用总寻，很容易搞錯(cuò)位置器罐。
調(diào)用 show 方法可以查看 DataFrame 里的數(shù)據(jù)：
df.show()
如果數(shù)據(jù)當(dāng)中有中文的話，show 方法會(huì)報(bào)編碼錯(cuò)誤渐行，需要提前設(shè)置一下環(huán)境變量：
export PYTHONIOENCODING=utf8
然后轰坊，就可以像 RDD 一樣，進(jìn)行各種數(shù)據(jù)變化祟印、聚合操作了肴沫。不過遍查文檔，你也找不到 DataFrame 的 map 方法蕴忆，你需要用 select 和 withColumn 這兩個(gè)方法來替代颤芬。其實(shí)回想 DataFrame 在 Saprk SQL 這個(gè)模塊里，所以它的很多行為都會(huì)非常像關(guān)系數(shù)據(jù)庫的 SQL 查詢套鹅。
可以把每個(gè) DataFrame 都想象成一個(gè)臨時(shí)表站蝠，select 方法，就是從這個(gè)表里選擇出一些字段卓鹿，做一點(diǎn)變換菱魔，變成另外一個(gè) DataFrame（另外一個(gè)臨時(shí)表）：
df2 = df.select(df['name'], df['age'] +1)df2.show()# +-------+---------+# |? name|(age + 1)|# +-------+---------+# |Michael|? ? null|# |? Andy|? ? ? 31|# | Justin|? ? ? 20|# +-------+---------+
在實(shí)際使用中，我們經(jīng)常會(huì)在某個(gè) DataFrame 基礎(chǔ)上加上一個(gè)變換后的字段吟孙，如果用 select 來寫豌习，就是這樣：
importpyspark.sql.functionsasFdf3 = df2.select("*", F.abs(df2.age).alias("abs_age"))
這里有三個(gè)新知識點(diǎn)：一個(gè)是在 select 中使用 "*"存谎，這樣可以避免我們把 df2 的所有字段重新敲一遍。然后我們新增了一個(gè)字段肥隆，是 age 的絕對值既荚。我們這里用到了 abs 函數(shù)，它來自 pyspark.sql.functions栋艳，是 spark 內(nèi)置的眾多變換函數(shù)之一恰聘。pyspark.sql.functions 里提供了相當(dāng)多的變換函數(shù)，可以在文檔里查到吸占。這些變換函數(shù)結(jié)合加減乘除這些運(yùn)算晴叨，基本上可以做完成任何變換了。實(shí)在搞不定的矾屯，可以用我們后面講的 UDF（用戶自定義函數(shù)）兼蕊。第三點(diǎn)是用到了 alias 函數(shù)，來為新變換出的字段命名件蚕，類似 SQL 語句中的 "as"孙技。下圖是內(nèi)置函數(shù)文檔目錄：
內(nèi)置函數(shù)文檔目錄
上面這段代碼使用 withColumn 會(huì)更加簡潔，withColumn 就是在一個(gè) DataFrame 的基礎(chǔ)上增加或替換一個(gè)字段：
importpyspark.sql.functionsasFdf3 = df2.withColumn("abs_age", F.abs(df2.age))
然后有些字段變換成其他字段以后就沒用了排作，可以 drop 掉它：
df4 = df.drop("name")df5 = df.select("*", df.age+1).drop(df.age)
我們到哪了牵啦？我們還是在說 RDD 的 map 方法，可以用 DataFrame 的 select妄痪、withColumn哈雏、alias、drop衫生、加減乘除裳瘪、pyspark.sql.functions 組合來替代∽镎耄看似是用很多東西才替代了 RDD 的 map彭羹，但是實(shí)際開發(fā)的時(shí)候會(huì)很省事，省去了在 map 函數(shù)中每次都要把需要的字段逐一返回了站故，特別是有時(shí)候只是把某些字段在 pair 的 key 和 value 間移動(dòng)皆怕。后面還會(huì)看到，DataFrame 可以直接對任意字段做 groupBy西篓，而不用先移動(dòng)擺好 pair 的 key 和 value愈腾。
如果你需要做的變換實(shí)在太復(fù)雜，無法用加減乘除和 pyspark.sql.functions 來搞定岂津，那么 DataFrame 也支持自定義函數(shù)：udf虱黄。udf 本身是一個(gè)普通的內(nèi)置函數(shù)，可以用它來包裝普通的 python 函數(shù)吮成，生成 select 和 withColumn 支持的變換函數(shù)：
importpyspark.sql.functionsasFimportpyspark.sql.typesasTdefdo_something(col1, col2, col3):returncol1*col1 + col2 / col3udf_dosth = F.udf(do_something, T.IntegerType())df6 = df.select(df.name, udf_dosth(df.age, df.born_year, df.children_count))
好了橱乱，到此可以先松口氣辜梳，休息一下了。filter 方法 DataFrame 直接支持泳叠，不需要尋找替代品：
df.filter(df['age'] >21).show()# +---+----+# |age|name|# +---+----+# | 30|Andy|# +---+----+
DataFrame 可以很方便地做 join作瞄、groupBy 等操作，就像在寫 SQL：
people = sqlContext.read.parquet("...")department = sqlContext.read.parquet("...")people.filter(people.age >30).join(department, people.deptId == department.id) \? .groupBy(department.name,"gender").agg({"salary":"avg","age":"max"})
其實(shí)可以真的把 DataFrame 注冊成臨時(shí)表危纫，然后真的寫 SQL：
df.createOrReplaceTempView("people")sqlDF = spark.sql("SELECT * FROM people")sqlDF.show()# +----+-------+# | age|? name|# +----+-------+# |null|Michael|# |? 30|? Andy|# |? 19| Justin|# +----+-------+
agg 這個(gè)方法是用來對 groupBy 之后的數(shù)據(jù)集做聚合操作的（對應(yīng) RDD 的 reduceByKey 方法）宗挥。它支持 avg、max种蝶、min契耿、sum、count 等操作搪桂。但是這幾個(gè)操作在實(shí)際使用中是遠(yuǎn)遠(yuǎn)不夠用的，這時(shí)候我們還是需要自定義函數(shù)的盯滚。這種自定義函數(shù)叫做 UDAF( User Defined Aggregate Function)踢械。UDAF 只在 Spark 的 scala 和 Java 中支持顾瞻，pyspark并不支持会油。在 Scala 中翻翩，你需要重載 UserDefinedAggregateFunction 這個(gè)類即可。本文就不具體展示了稻薇，留待我稍后一篇專門介紹 Scala Spark 的文章里細(xì)說嫂冻。
DataFrame API 參考：http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame
pyspark.sql.functions 參考：
http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#module-pyspark.sql.functions
作者：許倫
鏈接：http://www.reibang.com/p/b1398f9f5a06
