2. 程式人生 > >Spark-SQL之DataFrame操作大全

Spark-SQL之DataFrame操作大全

阿新 • • 發佈：2018-12-21

　　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方法，待後續補充。

相關推薦

Spark-SQL之DataFrame操作大全

　　Spark SQL中的DataFrame類似於一張關係型資料表。在關係型資料庫中對單表或進行的查詢操作，在DataFrame中都可以通過呼叫其API介面來實現。可以參考，Scala提供的DataFrame AP

Spark-Sql之DataFrame實戰詳解

集合 case 編程方式 優化 所表 register 操作數 print ava 1、DataFrame簡介： 在Spark中，DataFrame是一種以RDD為基礎的分布式數據據集，類似於傳統數據庫聽二維表格，DataFrame帶有Schema元信息，即DataFram

Spark 系列（八）—— Spark SQL 之 DataFrame 和 Dataset

## 一、Spark SQL簡介 Spark SQL 是 Spark 中的一個子模組，主要用於操作結構化資料。它具有以下特點： + 能夠將 SQL 查詢與 Spark 程式無縫混合，允許您使用 SQL 或 DataFrame API 對結構化資料進行查詢； + 支援多種開發語言； + 支援

Spark SQL and DataFrame Guide(1.4.1)——之DataFrames

ati been -m displays txt -a 版本 ava form Spark SQL是處理結構化數據的Spark模塊。它提供了DataFrames這樣的編程抽象。同一時候也能夠作為分布式SQL查詢引擎使用。 DataFrames D

Spark SQL中Dataframe join操作含null值的列

dataframe util pre table log n-n dram blog between 當在Spark SQL中對兩個Dataframe使用join時，當作為連接的字段的值含有null值。由於null表示的含義是未知，既不知道有沒有，在SQL中null值與任何

Spark修煉之道（進階篇）——Spark入門到精通：第十三節 Spark Streaming—— Spark SQL、DataFrame與Spark Streaming

主要內容 Spark SQL、DataFrame與Spark Streaming 1. Spark SQL、DataFrame與Spark Streaming import org.apache.spark.SparkConf import org

Spark SQL 之 Join 實現

結構 很多 找到 過濾 sql查詢優化 ade read 轉換成 分析 原文地址：Spark SQL 之 Join 實現 Spark SQL 之 Join 實現 塗小剛 2017-07-19 217標簽： spark ， 數據庫 Join作為SQL中

11.spark sql之RDD轉換DataSet

Once lds nco ldd 方法 att context gin statement 簡介 ??Spark SQL提供了兩種方式用於將RDD轉換為Dataset。 使用反射機制推斷RDD的數據結構 ??當spark應用可以推斷RDD數據結構時，可使用這種方式。這種

12.spark sql之讀寫數據

rcfile serializa fig jdbc連接 nco .sh nat 字段 jdb 簡介 ??Spark SQL支持多種結構化數據源，輕松從各種數據源中讀取Row對象。這些數據源包括Parquet、JSON、Hive表及關系型數據庫等。 ??當只使用一部分字段時，

10.spark sql之快速入門

繼續 ssi org 特性 splay ssa select 開始 pytho 前世今生 Hive&Shark ??隨著大數據時代的來臨，Hadoop風靡一時。為了使熟悉RDBMS但又不理解MapReduce的技術人員快速進行大數據開發，Hive應運而生。Hive是

Spark Streaming整合Spark SQL之wordcount案例

完整原始碼地址：  https://github.com/apache/spark/blob/v2.3.2/examples/src/main/scala/org/apache/spark/examples/streaming/SqlNetworkWordCount.scala 案例原

二、spark SQL互動scala操作示例

一、安裝spark spark SQL是spark的一個功能模組，所以我們事先要安裝配置spark，參考： https://www.cnblogs.com/lay2017/p/10006935.html 二、資料準備 演示操作將從一個類似json檔案裡面讀取資料作為資料來源，並初始化為dat

十三.Spark SQL之通過Zeppelin進行統計資料的展示

    Spark SQL學習有一段時間了,因此花了一些時間寫了一個日誌清洗的專案,專案已經上傳到github上了,    專案地址感興趣的可以拉下來看看。      在這裡我不講關於專案的實現過程,清洗之後的結果進行資料展示的時候,除了echarts框架,還發

十六.Spark SQL之讀取複雜的json資料

第一步.準備json資料 test.json {"name":"liguohui","nums":[1,2,3,4,5]} {"name":"zhangsan","nums":[6,7,8,9,10]} test2.json {"name":"Yin","ad

第68課：Spark SQL通過JDBC操作MySQL

內容：     1.SparkSQL操作關係資料庫意義     2.SparkSQL操作關係資料庫 一、通過SparkSQL操作關係資料庫意義     1.SparkSQL可以通過jdbc從傳統關係型資料庫中讀寫資料，讀取資料後直接生成DataFrame，然後在加上藉助

Spark修煉之道（高階篇）——Spark原始碼閱讀：第十三節 Spark SQL之SQLContext（一)

作者：周志湖 1. SQLContext的建立 SQLContext是Spark SQL進行結構化資料處理的入口，可以通過它進行DataFrame的建立及SQL的執行，其建立方式如下： //sc為SparkContext val sqlContext

Spark SQL之外部資料來源

概述   從Spark 1.2版本開始，Spark SQL正式支援外部資料來源。它可以通過DataFrame介面對各種資料來源進行操作，例如orc,parquet,json,hive,jdbc,avro等。它既可以通過轉換成RDD進行操作，也可以被建立為一個臨時檢視。將外部資料讀入後建立

Spark SQL之Join優化

SparkSQL總體流程介紹 在闡述Join實現之前，我們首先簡單介紹SparkSQL的總體流程，一般地，我們有兩種方式使用SparkSQL，一種是直接寫sql語句，這個需要有元資料庫支援，例如Hive等，另一種是通過Dataset/DataFrame編寫Spark應用

spark-SQL的DataFrame和DataSet

       方法一：                   sqlContext.read() 返回DataFrameReader物件 sqlContext.read().json("student.json")讀取一個json檔案（這個json檔案中的內容不能是巢狀的）讀進來變成DataFrame,

六.SparkSQL之DataFrame操作例項實戰API

前面已經說過DataFrame的基本API操作,接下來就講下實際開發過程中的應用,首先需要準備一個數據原始檔,以下面這種格式的資料(這種資料是可以清洗日誌得到的) 第一步: 例項實戰演示 import org.apache.spark.sql.SparkSes