一篇文章看懂 Spark RDD
1 簡介
Apache Spark 是專為大規模數據處理而設計的快速通用的計算引擎。它產生于 UC Berkeley AMP Lab，繼承了 MapReduce 的優點，但是不同于 MapReduce 的是，Spark 可以將結果保存在內存中，一直迭代計算下去，除非遇到 shuffle 。因此 Spark 能更好的適用于數據挖掘與機器學習等要迭代的算法。值得注意的是，官網說的 Spark 是 MR 計算速度的 100 倍。僅僅適用于邏輯回歸等這樣的迭代計算。
2 Spark 的運行模式
- Local 模式：多用于本機編寫、測試代碼。
- Standalone 模式：這是 Spark 自帶的資源調度框架，它支持完全分布式。
- Yarn 模式：這是 hadoop 里面的一個資源調度框架，Spark 同樣也可以使用。
- Mesos 模式：為應用程序(如Hadoop、Spark、Kafka、ElasticSearch)提供API的整個數據中心和云環境中的資源管理和調度。
下面分別介紹一下 Standalone 和 Yarn 模式下任務流程。
Standalone-client 提交方式
提交命令如下：以官方給的計算 PI 的代碼為例。
- ./spark-submit
- --master spark://node1:7077
- --class org.apache.spark.example.SaprkPi
- ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
- 1000
執行流程圖以及原理：
Standalone-cluster 提交方式
提交命令如下：以官方給的計算 PI 的代碼為例。
- ./spark-submit
- --master spark://node1:7077
- --deploy-mode cluster
- --class org.apache.spark.example.SaprkPi
- ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
- 1000
執行流程圖以及原理：
Yarn-cluster 提交方式
提交命令如下：以官方給的計算 PI 的代碼為例。
- ./spark-submit
- --master yarn
- --deploy-mode client
- --class org.apache.spark.example.SaprkPi
- ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
- 1000
執行流程圖以及原理：
Yarn-cluster 提交方式
提交命令如下：以官方給的計算 PI 的代碼為例。
- ./spark-submit
- --master yarn
- --deploy-mode cluster
- --class org.apache.spark.example.SaprkPi
- ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
- 1000
執行流程圖以及原理：
3 RDD
Spark core 最核心的就是 Resilient Distributed Dataset (RDD) 了，RDD 比較抽象了。源碼中 RDD.scala 中對 RDD 進行了一段描述。最主要的是下面的五個方面;
- /**
- * Internally, each RDD is characterized by five main properties:
- *
- * - A list of partitions
- * - A function for computing each split
- * - A list of dependencies on other RDDs
- * - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
- * - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
- * an HDFS file)
- *
- * All of the scheduling and execution in Spark is done based on these methods, allowing each RDD
- * to implement its own way of computing itself. Indeed, users can implement custom RDDs (e.g. for
- * reading data from a new storage system) by overriding these functions.
- */
RDD 的五大特性：
- RDD 是由一系列的 Partition 組成的。
- 函數作用在每一個 split 上。
- RDD 之間有一系列依賴關系。
- 分區器是作用在 K,V 格式的 RDD 上。
- RDD 提供一系列***的位置
先記住這五個特性，之后的學習會慢慢體會到這樣設計的好處。下面是理解 RDD 的邏輯圖;
看這個圖再回頭理解一下上面的五個 RDD 的特性。
RDD 的彈性表現在 Partition 的數量上，并且大小沒有限制。RDD 的依賴關系，可以基于上一個 RDD 計算出下一個 RDD。RDD 的每個 partition 是分布在不同數據節點上的，所有 RDD 的分布式的。RDD 提供了一些列的***的計算位置，體現了數據的本地化，我之前的這篇文章寫過：一文搞懂數據本地化級別
RDD 還有一個 Lineage 的東西，叫做血統。
Lineage 簡介：利用內存加快數據加載，在其它的In-Memory類數據庫或Cache類系統中也有實現。Spark的主要區別在于它采用血統來實現分布式運算環境下的數據容錯性(節點失效、數據丟失)問題。
RDD Lineage被稱為RDD運算圖或RDD依賴關系圖，是RDD所有父RDD的圖。它是在RDD上執行transformations函數并創建邏輯執行計劃(logical execution plan)的結果，是RDD的邏輯執行計劃。
相比其它系統的細顆粒度的內存數據更新級別的備份或者LOG機制，RDD 的 Lineage 記錄的是粗顆粒度的特定數據轉換(Transformation)操作(filter, map, join etc.)行為。當這個 RDD 的部分分區數據丟失時，它可以通過Lineage找到丟失的父RDD的分區進行局部計算來恢復丟失的數據，這樣可以節省資源提高運行效率。這種粗顆粒的數據模型，限制了Spark的運用場合，但同時相比細顆粒度的數據模型，也帶來了性能的提升。
4 控制算子
控制算子有三種：cache, persist, checkpoint, 以上算子都可以將 RDD 持久化、持久化的單位是 Partition。
cache 和 persist 都是懶執行的，必須有一個 action 算子來觸發他們執行。checkpoint 不僅可以將 RDD 持久化到磁盤，還能切斷 RDD 之間的依賴關系。
說幾點區別：
- cache 的持久化級別是 Memory_Only，就這一個。
- persist 的持久化級別：常用的有Memory_Only 和Memory_and_Disk_2， 數字 2 表示副本數。
- checkpoint 主要是用來做容錯的。
checkpoint 的執行原理是：當 RDD 的 job 執行完畢之后，會從 finalRDD 進行回溯。當回溯到某一個 RDD 調用了 checkpoint 方法，會對當前的 RDD 做一個標記。Spark 框架會自動啟動一個新的 Job ，重新計算這個 RDD 的數據，將數據持久化到 HDFS 上。根據這個原理，我們可以進行優化，對 RDD 進行 checkpoint 之前，***先對這個 RDD 進行 cache, 這樣啟動新的 job 只需要將內存中的數據拷貝到 HDFS 上就可以了，節省了重新計算這一步。
5 RDD 的依賴關系
窄依賴：指父RDD的每一個分區最多被一個子RDD的分區所用，表現為一個父RDD的分區對應于一個子RDD的分區，和兩個父RDD的分區對應于一個子RDD 的分區。圖中，map/filter/union屬于***類，對輸入進行協同劃分(co-partitioned)的join屬于第二類。窄依賴不會產生 shuffle。
寬依賴：指子RDD的分區依賴于父RDD的所有分區，這是因為 shuffle 類操作，如圖中的 groupByKey 和未經協同劃分的 join。 遇到寬依賴會產生 shuffle 。
上面我們說到了 RDD 之間的依賴關系，這些依賴關系形成了一個人 DAG 有向無環圖。DAG 創建完成之后，會被提交給 DAGScheduler, 它負責把 DAG 劃分相互依賴的多個 stage ，劃分依據就是 RDD 之間的窄寬依賴。換句話說就是，遇到一個寬依賴就劃分一個 stage，每一個 stage 包含一個或多個 stask 任務。然后將這些 task 以 taskset 的方式提交給 TaskScheduler 運行。也可以說 stage 是由一組并行的 task 組成。下圖很清楚的描述了 stage 的劃分。
6 Stage劃分思路
接上圖，Spark 劃分 stage 的整體思路是：從后往前推，遇到寬依賴就斷開，劃分為一個stage;遇到窄依賴就將這個 RDD 加入該 stage 中。
因此在圖中 RDD C, RDD D, RDD E, RDD F 被構建在一個 stage 中, RDD A被構建在一個單獨的Stage中,而 RDD B 和 RDD G 又被構建在同一個 stage中。
另一個角度
一個 Job 會被拆分為多組 Task，每組任務被稱為一個Stage就像 Map Stage，Reduce Stage。
Stage 的劃分簡單的說是以 shuffle 和 result 這兩種類型來劃分。在 Spark中有兩類 task，一類是 shuffleMapTask，一類是 resultTask，***類 task的輸出是 shuffle 所需數據，第二類 task 的輸出是 result，stage的劃分也以此為依據，shuffle 之前的所有變換是一個 stage，shuffle之后的操作是另一個stage。
如果 job 中有多次 shuffle，那么每個 shuffle 之前都是一個 stage. 會根據 RDD 之間的依賴關系將 DAG圖劃分為不同的階段，對于窄依賴，由于 partition 依賴關系的確定性，partition 的轉換處理就可以在同一個線程里完成，窄依賴就被 spark 劃分到同一個 stage 中，而對于寬依賴，只能等父 RDD shuffle 處理完成后，下一個 stage 才能開始接下來的計算。之所以稱之為 ShuffleMapTask 是因為它需要將自己的計算結果通過 shuffle 到下一個 stage 中。
