Spark-SQL之DataFrame操作大全
Spark SQL中的DataFrame類似於一張關係型資料表。在關係型資料庫中對單表或進行的查詢操作,在DataFrame中都可以通過呼叫其API介面來實現。可以參考,Scala提供的DataFrame API。
本文中的程式碼基於Spark-1.6.2的文件實現。
一、DataFrame物件的生成
Spark-SQL可以以其他RDD物件、parquet檔案、json檔案、hive表,以及通過JDBC連線到其他關係型資料庫作為資料來源來生成DataFrame物件。本文將以MySQL資料庫為資料來源,生成DataFrame物件後進行相關的DataFame之上的操作。
文中生成DataFrame的程式碼如下:
object DataFrameOperations {
def main (args: Array[String ]) {
val sparkConf = new SparkConf().setAppName( "Spark SQL DataFrame Operations").setMaster( "local[2]" )
val sparkContext = new SparkContext(sparkConf)
val sqlContext = new SQLContext(sparkContext)
val url = "jdbc:mysql://m000:3306/test"
val jdbcDF = sqlContext.read.format( "jdbc" ).options(
Map( "url" -> url,
"user" -> "root",
"password" -> "root",
"dbtable" -> "spark_sql_test" )).load()
val joinDF1 = sqlContext.read.format( "jdbc" ).options(
Map("url" -> url ,
"user" -> "root",
"password" -> "root",
"dbtable" -> "spark_sql_join1" )).load()
val joinDF2 = sqlContext.read.format( "jdbc" ).options(
Map ( "url" -> url ,
"user" -> "root",
"password" -> "root",
"dbtable" -> "spark_sql_join2" )).load()
... ...
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
後續程式碼都在上面... ...
處。
二、DataFrame物件上Action操作
1、show
:展示資料
以表格的形式在輸出中展示jdbcDF
中的資料,類似於select * from spark_sql_test
的功能。
show
方法有四種呼叫方式,分別為,
(1)show
只顯示前20條記錄。
示例:
jdbcDF.show
- 1
結果:
(2)show(numRows: Int)
顯示numRows
條
示例:
jdbcDF.show(3)
- 1
結果:
(3)show(truncate: Boolean)
是否最多隻顯示20個字元,預設為true
。
示例:
jdbcDF.show(true)
jdbcDF.show(false)
- 1
- 2
結果:
(4)show(numRows: Int, truncate: Boolean)
綜合前面的顯示記錄條數,以及對過長字串的顯示格式。
示例:
jdbcDF.show(3, false)
- 1
結果:
2、collect
:獲取所有資料到陣列
不同於前面的show
方法,這裡的collect
方法會將jdbcDF
中的所有資料都獲取到,並返回一個Array
物件。
jdbcDF.collect()
- 1
結果如下,結果陣列包含了jdbcDF
的每一條記錄,每一條記錄由一個GenericRowWithSchema
物件來表示,可以儲存欄位名及欄位值。
3、collectAsList
:獲取所有資料到List
功能和collect
類似,只不過將返回結構變成了List
物件,使用方法如下
jdbcDF.collectAsList()
- 1
結果如下,
4、describe(cols: String*)
:獲取指定欄位的統計資訊
這個方法可以動態的傳入一個或多個String
型別的欄位名,結果仍然為DataFrame
物件,用於統計數值型別欄位的統計值,比如count, mean, stddev, min, max
等。
使用方法如下,其中c1
欄位為字元型別,c2
欄位為整型,c4
欄位為浮點型
jdbcDF .describe("c1" , "c2", "c4" ).show()
- 1
結果如下,
5、first, head, take, takeAsList
:獲取若干行記錄
這裡列出的四個方法比較類似,其中
(1)first
獲取第一行記錄
(2)head
獲取第一行記錄,head(n: Int)
獲取前n行記錄
(3)take(n: Int)
獲取前n行資料
(4)takeAsList(n: Int)
獲取前n行資料,並以List
的形式展現
以Row
或者Array[Row]
的形式返回一行或多行資料。first
和head
功能相同。
take
和takeAsList
方法會將獲得到的資料返回到Driver端,所以,使用這兩個方法時需要注意資料量,以免Driver發生OutOfMemoryError
使用和結果略。
二、DataFrame物件上的條件查詢和join等操作
以下返回為DataFrame型別的方法,可以連續呼叫。
1、where條件相關
(1)where(conditionExpr: String)
:SQL語言中where關鍵字後的條件
傳入篩選條件表示式,可以用and
和or
。得到DataFrame型別的返回結果,
示例:
jdbcDF .where("id = 1 or c1 = 'b'" ).show()
- 1
結果,
(2)filter
:根據欄位進行篩選
傳入篩選條件表示式,得到DataFrame型別的返回結果。和where
使用條件相同
示例:
jdbcDF .filter("id = 1 or c1 = 'b'" ).show()
- 1
結果,
2、查詢指定欄位
(1)select
:獲取指定欄位值
根據傳入的String
型別欄位名,獲取指定欄位的值,以DataFrame型別返回
示例:
jdbcDF.select( "id" , "c3" ).show( false)
- 1
結果:
還有一個過載的select
方法,不是傳入String
型別引數,而是傳入Column
型別引數。可以實現select id, id+1 from test
這種邏輯。
jdbcDF.select(jdbcDF( "id" ), jdbcDF( "id") + 1 ).show( false)
- 1
結果:
能得到Column
型別的方法是apply
以及col
方法,一般用apply
方法更簡便。
(2)selectExpr
:可以對指定欄位進行特殊處理
可以直接對指定欄位呼叫UDF函式,或者指定別名等。傳入String
型別引數,得到DataFrame物件。
示例,查詢id
欄位,c3
欄位取別名time
,c4
欄位四捨五入:
jdbcDF .selectExpr("id" , "c3 as time" , "round(c4)" ).show(false)
- 1
結果,
(3)col
:獲取指定欄位
只能獲取一個欄位,返回物件為Column型別。
val idCol = jdbcDF.col(“id”)果略。
(4)apply
:獲取指定欄位
只能獲取一個欄位,返回物件為Column型別
示例:
val idCol1 = jdbcDF.apply("id")
val idCol2 = jdbcDF("id")
- 1
- 2
結果略。
(5)drop
:去除指定欄位,保留其他欄位
返回一個新的DataFrame物件,其中不包含去除的欄位,一次只能去除一個欄位。
示例:
jdbcDF.drop("id")
jdbcDF.drop(jdbcDF("id"))
- 1
- 2
結果:
3、limit
limit
方法獲取指定DataFrame的前n行記錄,得到一個新的DataFrame物件。和take
與head
不同的是,limit
方法不是Action操作。
jdbcDF.limit(3).show( false)
- 1
結果,
4、order by
(1)orderBy
和sort
:按指定欄位排序,預設為升序
示例1,按指定欄位排序。加個-
表示降序排序。sort
和orderBy
使用方法相同
jdbcDF.orderBy(- jdbcDF("c4")).show(false)
// 或者
jdbcDF.orderBy(jdbcDF("c4").desc).show(false)
- 1
- 2
- 3
結果,
示例2,按欄位字串升序排序
jdbcDF.orderBy("c4").show(false)
- 1
結果,
(2)sortWithinPartitions
和上面的sort
方法功能類似,區別在於sortWithinPartitions
方法返回的是按Partition排好序的DataFrame物件。
5、group by
(1)groupBy
:根據欄位進行group by
操作
groupBy
方法有兩種呼叫方式,可以傳入String
型別的欄位名,也可傳入Column
型別的物件。
使用方法如下,
jdbcDF .groupBy("c1" )
jdbcDF.groupBy( jdbcDF( "c1"))
- 1
- 2
(2)cube
和rollup
:group by的擴充套件
功能類似於SQL
中的group by cube/rollup
,略。
(3)GroupedData物件
該方法得到的是GroupedData
型別物件,在GroupedData
的API中提供了group by
之後的操作,比如,
max(colNames: String*)
方法,獲取分組中指定欄位或者所有的數字型別欄位的最大值,只能作用於數字型欄位min(colNames: String*)
方法,獲取分組中指定欄位或者所有的數字型別欄位的最小值,只能作用於數字型欄位mean(colNames: String*)
方法,獲取分組中指定欄位或者所有的數字型別欄位的平均值,只能作用於數字型欄位sum(colNames: String*)
方法,獲取分組中指定欄位或者所有的數字型別欄位的和值,只能作用於數字型欄位count()
方法,獲取分組中的元素個數
執行結果示例:
count
max
這裡面比較複雜的是以下兩個方法,
agg
,該方法和下面介紹的類似,可以用於對指定欄位進行聚合操作。
pivot
6、distinct
(1)distinct
:返回一個不包含重複記錄的DataFrame
返回當前DataFrame中不重複的Row記錄。該方法和接下來的dropDuplicates()
方法不傳入指定欄位時的結果相同。
示例:
jdbcDF.distinct()
- 1
結果,
(2)dropDuplicates
:根據指定欄位去重
根據指定欄位去重。類似於select distinct a, b
操作
示例:
jdbcDF.dropDuplicates(Seq("c1"))
- 1
結果:
7、聚合
聚合操作呼叫的是agg
方法,該方法有多種呼叫方式。一般與groupBy
方法配合使用。
以下示例其中最簡單直觀的一種用法,對id
欄位求最大值,對c4
欄位求和。
jdbcDF.agg("id" -> "max", "c4" -> "sum")
- 1
結果:
8、union
unionAll
方法:對兩個DataFrame進行組合
類似於SQL
中的UNION ALL
操作。
示例:
jdbcDF.unionALL(jdbcDF.limit(1))
- 1
結果:
9、join
重點來了。在SQL
語言中用得很多的就是join
操作,DataFrame中同樣也提供了join
的功能。
接下來隆重介紹join
方法。在DataFrame中提供了六個過載的join
方法。
(1)、笛卡爾積
joinDF1.join(joinDF2)
- 1
(2)、using
一個欄位形式
下面這種join類似於a join b using column1
的形式,需要兩個DataFrame中有相同的一個列名,
joinDF1.join(joinDF2, "id")
- 1
joinDF1
和joinDF2
根據欄位id
進行join
操作,結果如下,using
欄位只顯示一次。
(3)、using
多個欄位形式
除了上面這種using
一個欄位的情況外,還可以using
多個欄位,如下
joinDF1.join(joinDF2, Seq("id", "name"))
- 1
(4)、指定join
型別
兩個DataFrame的join
操作有inner, outer, left_outer, right_outer, leftsemi
型別。在上面的using
多個欄位的join情況下,可以寫第三個String
型別引數,指定join
的型別,如下所示
joinDF1.join(joinDF2, Seq("id", "name"), "inner")
- 1
(5)、使用Column
型別來join
如果不用using
模式,靈活指定join
欄位的話,可以使用如下形式
joinDF1.join(joinDF2 , joinDF1("id" ) === joinDF2( "t1_id"))
- 1
結果如下,
(6)、在指定join
欄位同時指定join
型別
如下所示
joinDF1.join(joinDF2 , joinDF1("id" ) === joinDF2( "t1_id"), "inner")
- 1
10、獲取指定欄位統計資訊
stat
方法可以用於計算指定欄位或指定欄位之間的統計資訊,比如方差,協方差等。這個方法返回一個DataFramesStatFunctions
型別物件。
下面程式碼演示根據c4
欄位,統計該欄位值出現頻率在30%
以上的內容。在jdbcDF
中欄位c1
的內容為"a, b, a, c, d, b"
。其中a
和b
出現的頻率為2 / 6
,大於0.3
jdbcDF.stat.freqItems(Seq ("c1") , 0.3).show()
- 1
結果如下:
11、獲取兩個DataFrame中共有的記錄
intersect
方法可以計算出兩個DataFrame中相同的記錄,
jdbcDF.intersect(jdbcDF.limit(1)).show(false)
- 1
結果如下:
12、獲取一個DataFrame中有另一個DataFrame中沒有的記錄
示例:
jdbcDF.except(jdbcDF.limit(1)).show(false)
- 1
結果如下,
13、操作欄位名
(1)withColumnRenamed
:重新命名DataFrame中的指定欄位名
如果指定的欄位名不存在,不進行任何操作。下面示例中將jdbcDF
中的id
欄位重新命名為idx
。
jdbcDF.withColumnRenamed( "id" , "idx" )
- 1
結果如下:
(2)withColumn
:往當前DataFrame中新增一列
whtiColumn(colName: String , col: Column)
方法根據指定colName
往DataFrame中新增一列,如果colName
已存在,則會覆蓋當前列。
以下程式碼往jdbcDF
中新增一個名為id2
的列,
jdbcDF.withColumn("id2", jdbcDF("id")).show( false)
- 1
結果如下,
14、行轉列
有時候需要根據某個欄位內容進行分割,然後生成多行,這時可以使用explode
方法
下面程式碼中,根據c3
欄位中的空格將欄位內容進行分割,分割的內容儲存在新的欄位c3_
中,如下所示
jdbcDF.explode( "c3" , "c3_" ){time: String => time.split( " " )}
- 1
結果如下,
15、其他操作
API中還有na, randomSplit, repartition, alias, as
方法,待後續補充。