RDD、DataFrame和DataSet的區別是什么

si19891002 8年前發布 | 17K 次閱讀 Spark SQL 分布式系統 數據挖掘

來自: http://developer.51cto.com/art/201603/507668.htm

RDD、DataFrame和DataSet的區別是什么

RDD、DataFrame和DataSet是容易產生混淆的概念,必須對其相互之間對比,才可以知道其中異同:DataFrame多了數據的結構信息,即schema。RDD是分布式的 Java對象的集合。DataFrame是分布式的Row對象的集合。

<dl> 
 <dt>        作者:jacksu 來源: 
  <a href="/misc/goto?guid=4959669737051459023" target="_blank" rel="nofollow,noindex">簡書</a> | 
  <em>2016-03-21 10:40</em> 
 </dt> 
</dl> 

</div>

51CTO技術沙龍 | 賦予APP不同凡響的交互和體驗>>

RDD、DataFrame和DataSet是容易產生混淆的概念,必須對其相互之間對比,才可以知道其中異同。

RDD和DataFrame

RDD、DataFrame和DataSet的區別是什么

RDD-DataFrame

上圖直觀地體現了DataFrame和RDD的區別。左側的RDD[Person]雖然以Person為類型參數,但Spark框架本身不了解 Person類的內部結構。而右側的DataFrame卻提供了詳細的結構信息,使得Spark SQL可以清楚地知道該數據集中包含哪些列,每列的名稱和類型各是什么。DataFrame多了數據的結構信息,即schema。RDD是分布式的 Java對象的集合。DataFrame是分布式的Row對象的集合。DataFrame除了提供了比RDD更豐富的算子以外,更重要的特點是提升執行效率、減少數據讀取以及執行計劃的優化,比如filter下推、裁剪等。

提升執行效率

RDD API是函數式的,強調不變性,在大部分場景下傾向于創建新對象而不是修改老對象。這一特點雖然帶來了干凈整潔的API,卻也使得Spark應用程序在運行期傾向于創建大量臨時對象,對GC造成壓力。在現有RDD API的基礎之上,我們固然可以利用mapPartitions方法來重載RDD單個分片內的數據創建方式,用復用可變對象的方式來減小對象分配和GC的開銷,但這犧牲了代碼的可讀性,而且要求開發者對Spark運行時機制有一定的了解,門檻較高。另一方面,Spark SQL在框架內部已經在各種可能的情況下盡量重用對象,這樣做雖然在內部會打破了不變性,但在將數據返回給用戶時,還會重新轉為不可變數據。利用 DataFrame API進行開發,可以免費地享受到這些優化效果。

減少數據讀取

分析大數據,最快的方法就是 ——忽略它。這里的“忽略”并不是熟視無睹,而是根據查詢條件進行恰當的剪枝。

上文討論分區表時提到的分區剪 枝便是其中一種——當查詢的過濾條件中涉及到分區列時,我們可以根據查詢條件剪掉肯定不包含目標數據的分區目錄,從而減少IO。

對于一些“智能”數據格 式,Spark SQL還可以根據數據文件中附帶的統計信息來進行剪枝。簡單來說,在這類數據格式中,數據是分段保存的,每段數據都帶有最大值、最小值、null值數量等 一些基本的統計信息。當統計信息表名某一數據段肯定不包括符合查詢條件的目標數據時,該數據段就可以直接跳過(例如某整數列a某段的最大值為100,而查詢條件要求a > 200)。

此外,Spark SQL也可以充分利用RCFile、ORC、Parquet等列式存儲格式的優勢,僅掃描查詢真正涉及的列,忽略其余列的數據。

執行優化

RDD、DataFrame和DataSet的區別是什么

人口數據分析示例

為了說明查詢優化,我們來看上圖展示的人口數據分析的示例。圖中構造了兩個DataFrame,將它們join之后又做了一次filter操作。如果原封不動地執行這個執行計劃,最終的執行效率是不高的。因為join是一個代價較大的操作,也可能會產生一個較大的數據集。如果我們能將filter 下推到 join下方,先對DataFrame進行過濾,再join過濾后的較小的結果集,便可以有效縮短執行時間。而Spark SQL的查詢優化器正是這樣做的。簡而言之,邏輯查詢計劃優化就是一個利用基于關系代數的等價變換,將高成本的操作替換為低成本操作的過程。

得到的優化執行計劃在轉換成物 理執行計劃的過程中,還可以根據具體的數據源的特性將過濾條件下推至數據源內。最右側的物理執行計劃中Filter之所以消失不見,就是因為溶入了用于執行最終的讀取操作的表掃描節點內。

對于普通開發者而言,查詢優化 器的意義在于,即便是經驗并不豐富的程序員寫出的次優的查詢,也可以被盡量轉換為高效的形式予以執行。

RDD和DataSet

  • DataSet以Catalyst邏輯執行計劃表示,并且數據以編碼的二進制形式被存儲,不需要反序列化就可以執行sorting、shuffle等操作。
  • DataSet創立需要一個顯式的Encoder,把對象序列化為二進制,可以把對象的scheme映射為SparkSQl類型,然而RDD依賴于運行時反射機制。

通過上面兩點,DataSet的性能比RDD的要好很多。

DataFrame和DataSet

Dataset可以認為是DataFrame的一個特例,主要區別是Dataset每一個record存儲的是一個強類型值而不是一個Row。因此具有如下三個特點:

DataSet可以在編譯時檢查類型

并且是面向對象的編程接口。用wordcount舉例:

//DataFrame

// Load a text file and interpret each line as a java.lang.String
val ds = sqlContext.read.text("/home/spark/1.6/lines").as[String]
val result = ds
  .flatMap(_.split(" "))               // Split on whitespace
  .filter(_ != "")                     // Filter empty words
  .toDF()                              // Convert to DataFrame to perform aggregation / sorting
  .groupBy($"value")                   // Count number of occurences of each word
  .agg(count("*") as "numOccurances")
  .orderBy($"numOccurances" desc)      // Show most common words first

后面版本DataFrame會繼承DataSet,DataFrame是面向Spark SQL的接口。

//DataSet,完全使用scala編程,不要切換到DataFrame

val wordCount = 
  ds.flatMap(_.split(" "))
    .filter(_ != "")
    .groupBy(_.toLowerCase()) // Instead of grouping on a column expression (i.e. $"value") we pass a lambda function
    .count()

DataFrame和DataSet可以相互轉化, df.as[ElementType] 這樣可以把DataFrame轉化為DataSet, ds.toDF() 這樣可以把DataSet轉化為DataFrame。

【編輯推薦】

  1. 使用 PHP 直接在共享內存中存儲數據集
  2. 組合鏈接數據集與Drupal 7和SPARQL Views
  3. 使用 SPSS 文本挖掘工具構建社交媒體數據集市
  4. 重磅!8大策略讓你對抗機器學習數據集里的不均衡數據
  5. 選擇 Parquet for Spark SQL 的 5 大原因

【責任編輯:Ophira TEL:(010)68476606】

</div>

 本文由用戶 si19891002 自行上傳分享,僅供網友學習交流。所有權歸原作者,若您的權利被侵害,請聯系管理員。
 轉載本站原創文章,請注明出處,并保留原始鏈接、圖片水印。
 本站是一個以用戶分享為主的開源技術平臺,歡迎各類分享!