DataFrame 是 Spark 在 RDD 之后新推出的一個(gè)數(shù)據(jù)集，從屬于 Spark SQL 模塊劫瞳，適用于結(jié)構(gòu)化數(shù)據(jù)和媳。對(duì)于我們這些用慣了數(shù)據(jù)庫(kù)表和散列/字典結(jié)構(gòu)的人來(lái)說(shuō)闰渔，尤為親切暮现。
可以直接讀取關(guān)系型數(shù)據(jù)庫(kù)產(chǎn)生 DataFrame：
from pyspark.sql import SparkSession
spark = 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)說(shuō)明一下：
- url - 數(shù)據(jù)庫(kù)的 JDBC 連接串
- dbtable - 可以是表名还绘，也可以是一個(gè)子查詢。如果是子查詢的話栖袋，必須用括號(hào)括起來(lái)，并加別名抚太，參見(jiàn)上面代碼示例塘幅。
- user - 數(shù)據(jù)庫(kù)用戶名
- password - 數(shù)據(jù)庫(kù)密碼
- partitionColumn - 用于并發(fā)分區(qū)的表字段昔案。下面幾個(gè)選項(xiàng)都是圍繞這個(gè)字段來(lái)的。Spark 會(huì)根據(jù)分區(qū)數(shù)量按這個(gè)字段的上下限把取出來(lái)的數(shù)據(jù)等分成幾份电媳，并行處理踏揣。
- lowerBound - 字段下限
- upperBound - 字段上限
- numPartitions - 分區(qū)數(shù)量
得到的 DataFrame，可以通過(guò)字段名來(lái)引用列或者某行的值匾乓，如：
username = df.name
age = 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ù)變化瘟檩、聚合操作了抹缕。不過(guò)遍查文檔，你也找不到 DataFrame 的 map 方法墨辛，你需要用 select 和 withColumn 這兩個(gè)方法來(lái)替代歉嗓。其實(shí)回想 DataFrame 在 Saprk SQL 這個(gè)模塊里，所以它的很多行為都會(huì)非常像關(guān)系數(shù)據(jù)庫(kù)的 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 來(lái)寫(xiě)伦糯，就是這樣：
import pyspark.sql.functions as F
df3 = df2.select("*", F.abs(df2.age).alias("abs_age"))
這里有三個(gè)新知識(shí)點(diǎn)：一個(gè)是在 select 中使用 "*"，這樣可以避免我們把 df2 的所有字段重新敲一遍嗽元。然后我們新增了一個(gè)字段敛纲，是 age 的絕對(duì)值。我們這里用到了 abs 函數(shù)剂癌，它來(lái)自 pyspark.sql.functions淤翔，是 spark 內(nèi)置的眾多變換函數(shù)之一。pyspark.sql.functions 里提供了相當(dāng)多的變換函數(shù)佩谷，可以在文檔里查到旁壮。這些變換函數(shù)結(jié)合加減乘除這些運(yùn)算监嗜，基本上可以做完成任何變換了。實(shí)在搞不定的抡谐，可以用我們后面講的 UDF（用戶自定義函數(shù)）裁奇。第三點(diǎn)是用到了 alias 函數(shù)，來(lái)為新變換出的字段命名麦撵，類似 SQL 語(yǔ)句中的 "as"刽肠。下圖是內(nèi)置函數(shù)文檔目錄：
上面這段代碼使用 withColumn 會(huì)更加簡(jiǎn)潔，withColumn 就是在一個(gè) DataFrame 的基礎(chǔ)上增加或替換一個(gè)字段：
import pyspark.sql.functions as F
df3 = df2.withColumn("abs_age", F.abs(df2.age))
然后有些字段變換成其他字段以后就沒(méi)用了免胃，可以 drop 掉它：
df4 = df.drop("name")
df5 = df.select("*", df.age+1).drop(df.age)
我們到哪了音五？我們還是在說(shuō) RDD 的 map 方法，可以用 DataFrame 的 select杜秸、withColumn放仗、alias、drop撬碟、加減乘除诞挨、pyspark.sql.functions 組合來(lái)替代∧馗颍看似是用很多東西才替代了 RDD 的 map惶傻，但是實(shí)際開(kāi)發(fā)的時(shí)候會(huì)很省事，省去了在 map 函數(shù)中每次都要把需要的字段逐一返回了其障，特別是有時(shí)候只是把某些字段在 pair 的 key 和 value 間移動(dòng)银室。后面還會(huì)看到，DataFrame 可以直接對(duì)任意字段做 groupBy励翼，而不用先移動(dòng)擺好 pair 的 key 和 value蜈敢。
如果你需要做的變換實(shí)在太復(fù)雜，無(wú)法用加減乘除和 pyspark.sql.functions 來(lái)搞定汽抚，那么 DataFrame 也支持自定義函數(shù)：udf抓狭。udf 本身是一個(gè)普通的內(nèi)置函數(shù)，可以用它來(lái)包裝普通的 python 函數(shù)造烁，生成 select 和 withColumn 支持的變換函數(shù)：
import pyspark.sql.functions as F
import pyspark.sql.types as T
def do_something(col1, col2, col3):
return col1*col1 + col2 / col3
udf_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 等操作告组，就像在寫(xiě) 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 注冊(cè)成臨時(shí)表煤伟，然后真的寫(xiě) SQL：
df.createOrReplaceTempView("people")
sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
# +----+-------+
# | age| name|
# +----+-------+
# |null|Michael|
# | 30| Andy|
# | 19| Justin|
# +----+-------+
agg 這個(gè)方法是用來(lái)對(duì) groupBy 之后的數(shù)據(jù)集做聚合操作的（對(duì)應(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ì)說(shuō)沥阱。
好了，本文就到這里伊群，有任何疑問(wèn)歡迎留言探討考杉。
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
