Spark 系列（三）—— 彈性式數據集RDDs

1、RDD簡介

RDD 全稱爲 Resilient Distributed Datasets，是 Spark 最基本的數據抽象，它是隻讀的、分區記錄的集合，支持並行操做，能夠由外部數據集或其餘 RDD 轉換而來，它具備如下特性：

• 一個 RDD 由一個或者多個分區（Partitions）組成。對於 RDD 來講，每一個分區會被一個計算任務所處理，用戶能夠在建立 RDD 時指定其分區個數，若是沒有指定，則默認採用程序所分配到的 CPU 的核心數；
• RDD 擁有一個用於計算分區的函數 compute；
• RDD 會保存彼此間的依賴關係，RDD 的每次轉換都會生成一個新的依賴關係，這種 RDD 之間的依賴關係就像流水線同樣。在部分分區數據丟失後，能夠經過這種依賴關係從新計算丟失的分區數據，而不是對 RDD 的全部分區進行從新計算；
• Key-Value 型的 RDD 還擁有 Partitioner(分區器)，用於決定數據被存儲在哪一個分區中，目前 Spark 中支持 HashPartitioner(按照哈希分區) 和 RangeParationer(按照範圍進行分區)；
• 一個優先位置列表 (可選)，用於存儲每一個分區的優先位置 (prefered location)。對於一個 HDFS 文件來講，這個列表保存的就是每一個分區所在的塊的位置，按照「移動數據不如移動計算「的理念，Spark 在進行任務調度的時候，會盡量的將計算任務分配到其所要處理數據塊的存儲位置。

RDD[T] 抽象類的部分相關代碼以下：

// 由子類實現以計算給定分區
def compute(split: Partition, context: TaskContext): Iterator[T]

// 獲取全部分區
protected def getPartitions: Array[Partition]

// 獲取全部依賴關係
protected def getDependencies: Seq[Dependency[_]] = deps

// 獲取優先位置列表
protected def getPreferredLocations(split: Partition): Seq[String] = Nil

// 分區器 由子類重寫以指定它們的分區方式
@transient val partitioner: Option[Partitioner] = None

2、建立RDD

RDD 有兩種建立方式，分別介紹以下：

2.1 由現有集合建立

這裏使用 spark-shell 進行測試，啓動命令以下：

spark-shell --master local[4]

啓動 spark-shell 後，程序會自動建立應用上下文，至關於執行了下面的 Scala 語句：

val conf = new SparkConf().setAppName("Spark shell").setMaster("local[4]")
val sc = new SparkContext(conf)

由現有集合建立 RDD，你能夠在建立時指定其分區個數，若是沒有指定，則採用程序所分配到的 CPU 的核心數：

val data = Array(1, 2, 3, 4, 5)
// 由現有集合建立 RDD,默認分區數爲程序所分配到的 CPU 的核心數
val dataRDD = sc.parallelize(data) 
// 查看分區數
dataRDD.getNumPartitions
// 明確指定分區數
val dataRDD = sc.parallelize(data,2)

執行結果以下：

2.2 引用外部存儲系統中的數據集

引用外部存儲系統中的數據集，例如本地文件系統，HDFS，HBase 或支持 Hadoop InputFormat 的任何數據源。

val fileRDD = sc.textFile("/usr/file/emp.txt")
// 獲取第一行文本
fileRDD.take(1)

使用外部存儲系統時須要注意如下兩點：

• 若是在集羣環境下從本地文件系統讀取數據，則要求該文件必須在集羣中全部機器上都存在，且路徑相同；
• 支持目錄路徑，支持壓縮文件，支持使用通配符。

2.3 textFile & wholeTextFiles

二者均可以用來讀取外部文件，可是返回格式是不一樣的：

• textFile：其返回格式是 RDD[String] ，返回的是就是文件內容，RDD 中每個元素對應一行數據；
• wholeTextFiles：其返回格式是 RDD[(String, String)]，元組中第一個參數是文件路徑，第二個參數是文件內容；
• 二者都提供第二個參數來控制最小分區數；
• 從 HDFS 上讀取文件時，Spark 會爲每一個塊建立一個分區。

def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {...}
def wholeTextFiles(path: String,minPartitions: Int = defaultMinPartitions): RDD[(String, String)]={..}

3、操做RDD

RDD 支持兩種類型的操做：transformations（轉換，從現有數據集建立新數據集）和 actions（在數據集上運行計算後將值返回到驅動程序）。RDD 中的全部轉換操做都是惰性的，它們只是記住這些轉換操做，但不會當即執行，只有遇到 action 操做後纔會真正的進行計算，這相似於函數式編程中的惰性求值。

val list = List(1, 2, 3)
// map 是一個 transformations 操做，而 foreach 是一個 actions 操做
sc.parallelize(list).map(_ * 10).foreach(println)
// 輸出： 10 20 30

4、緩存RDD

4.1 緩存級別

Spark 速度很是快的一個緣由是 RDD 支持緩存。成功緩存後，若是以後的操做使用到了該數據集，則直接從緩存中獲取。雖然緩存也有丟失的風險，可是因爲 RDD 之間的依賴關係，若是某個分區的緩存數據丟失，只須要從新計算該分區便可。

Spark 支持多種緩存級別 ：

Storage Level （存儲級別）	Meaning（含義）
MEMORY_ONLY	默認的緩存級別，將 RDD 以反序列化的 Java 對象的形式存儲在 JVM 中。若是內存空間不夠，則部分分區數據將再也不緩存。
MEMORY_AND_DISK	將 RDD 以反序列化的 Java 對象的形式存儲 JVM 中。若是內存空間不夠，將未緩存的分區數據存儲到磁盤，在須要使用這些分區時從磁盤讀取。
MEMORY_ONLY_SER	將 RDD 以序列化的 Java 對象的形式進行存儲（每一個分區爲一個 byte 數組）。這種方式比反序列化對象節省存儲空間，但在讀取時會增長 CPU 的計算負擔。僅支持 Java 和 Scala 。
MEMORY_AND_DISK_SER	相似於 MEMORY_ONLY_SER，可是溢出的分區數據會存儲到磁盤，而不是在用到它們時從新計算。僅支持 Java 和 Scala。
DISK_ONLY	只在磁盤上緩存 RDD
MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc	與上面的對應級別功能相同，可是會爲每一個分區在集羣中的兩個節點上創建副本。
OFF_HEAP	與 MEMORY_ONLY_SER 相似，但將數據存儲在堆外內存中。這須要啓用堆外內存。

啓動堆外內存須要配置兩個參數：

• spark.memory.offHeap.enabled ：是否開啓堆外內存，默認值爲 false，須要設置爲 true；
• spark.memory.offHeap.size : 堆外內存空間的大小，默認值爲 0，須要設置爲正值。

4.2 使用緩存

緩存數據的方法有兩個：persist 和 cache 。cache 內部調用的也是 persist，它是 persist 的特殊化形式，等價於 persist(StorageLevel.MEMORY_ONLY)。示例以下：

// 全部存儲級別均定義在 StorageLevel 對象中
fileRDD.persist(StorageLevel.MEMORY_AND_DISK)
fileRDD.cache()

4.3 移除緩存

Spark 會自動監視每一個節點上的緩存使用狀況，並按照最近最少使用（LRU）的規則刪除舊數據分區。固然，你也可使用 RDD.unpersist() 方法進行手動刪除。

5、理解shuffle

5.1 shuffle介紹

在 Spark 中，一個任務對應一個分區，一般不會跨分區操做數據。但若是遇到 reduceByKey 等操做，Spark 必須從全部分區讀取數據，並查找全部鍵的全部值，而後彙總在一塊兒以計算每一個鍵的最終結果 ，這稱爲 Shuffle。

5.2 Shuffle的影響

Shuffle 是一項昂貴的操做，由於它一般會跨節點操做數據，這會涉及磁盤 I/O，網絡 I/O，和數據序列化。某些 Shuffle 操做還會消耗大量的堆內存，由於它們使用堆內存來臨時存儲須要網絡傳輸的數據。Shuffle 還會在磁盤上生成大量中間文件，從 Spark 1.3 開始，這些文件將被保留，直到相應的 RDD 再也不使用並進行垃圾回收，這樣作是爲了不在計算時重複建立 Shuffle 文件。若是應用程序長期保留對這些 RDD 的引用，則垃圾回收可能在很長一段時間後纔會發生，這意味着長時間運行的 Spark 做業可能會佔用大量磁盤空間，一般可使用 spark.local.dir 參數來指定這些臨時文件的存儲目錄。

5.3 致使Shuffle的操做

因爲 Shuffle 操做對性能的影響比較大，因此須要特別注意使用，如下操做都會致使 Shuffle：

• 涉及到從新分區操做： 如 repartition 和 coalesce；
• 全部涉及到 ByKey 的操做：如 groupByKey 和 reduceByKey，但 countByKey 除外；
• 聯結操做：如 cogroup 和 join。

5、寬依賴和窄依賴

RDD 和它的父 RDD(s) 之間的依賴關係分爲兩種不一樣的類型：

• 窄依賴 (narrow dependency)：父 RDDs 的一個分區最多被子 RDDs 一個分區所依賴；
• 寬依賴 (wide dependency)：父 RDDs 的一個分區能夠被子 RDDs 的多個子分區所依賴。

以下圖，每個方框表示一個 RDD，帶有顏色的矩形表示分區：

區分這兩種依賴是很是有用的：

• 首先，窄依賴容許在一個集羣節點上以流水線的方式（pipeline）對父分區數據進行計算，例如先執行 map 操做，而後執行 filter 操做。而寬依賴則須要計算好全部父分區的數據，而後再在節點之間進行 Shuffle，這與 MapReduce 相似。
• 窄依賴可以更有效地進行數據恢復，由於只需從新對丟失分區的父分區進行計算，且不一樣節點之間能夠並行計算；而對於寬依賴而言，若是數據丟失，則須要對全部父分區數據進行計算並再次 Shuffle。

6、DAG的生成

RDD(s) 及其之間的依賴關係組成了 DAG(有向無環圖)，DAG 定義了這些 RDD(s) 之間的 Lineage(血統) 關係，經過血統關係，若是一個 RDD 的部分或者所有計算結果丟失了，也能夠從新進行計算。那麼 Spark 是如何根據 DAG 來生成計算任務呢？主要是根據依賴關係的不一樣將 DAG 劃分爲不一樣的計算階段 (Stage)：

• 對於窄依賴，因爲分區的依賴關係是肯定的，其轉換操做能夠在同一個線程執行，因此能夠劃分到同一個執行階段；
• 對於寬依賴，因爲 Shuffle 的存在，只能在父 RDD(s) 被 Shuffle 處理完成後，才能開始接下來的計算，所以遇到寬依賴就須要從新劃分階段。

參考資料

1. 張安站 . Spark 技術內幕：深刻解析 Spark 內核架構設計與實現原理[M] . 機械工業出版社 . 2015-09-01
2. RDD Programming Guide
3. RDD：基於內存的集羣計算容錯抽象

更多大數據系列文章能夠參見 GitHub 開源項目： 大數據入門指南