Hello Spark! | Spark，從入門到精通

歡迎閱讀美圖數據技術團隊的「Spark，從入門到精通」系列文章，本系列文章將由淺入深爲你們介紹 Spark，從框架入門到底層架構的實現，相信總有一種姿式適合你，歡迎你們持續關注：）

什麼是 Spark？

Spark 是 UC Berkeley AMP lab 所開源的類 Hadoop MapReduce 的通用並行框架，是專爲大規模數據處理而設計的快速通用的大數據處理引擎及輕量級的大數據處理統一平臺。

當咱們在談 Spark 的時候多是指一個 Spark 應用程序，替代 MapReduce 運行在 Yarn上，存儲在 HDFS 上的一個大數據批處理程序；也多是指使用包含 Spark sql、Spark streaming 等子項目；甚至 Tachyon、Mesos 等大數據處理的統一平臺，或者稱爲 Spark 生態。

發展至今，Spark 已不只僅是 MapReduce 的替換方案，它已經發出成了一個包含衆多子項目的 Spark 生態。如上圖所示，Spark 生態可分爲四層：

• 數據存儲層，以 HDFS 、Tachyon 爲表明的一些分佈式文件存儲系統或各類數據庫；
• 資源管理層，Yarn、Mesos 等資源管理器；
• 數據處理引擎；
• 應用層，以 Spark 爲基礎產生的衆多項目;

Spark SQL 提供 HiveQL（經過 Apache Hive 的 SQL 變體 Hive 查詢語言）與Spark 進行交互的 API。每一個數據庫表被當作一個 RDD，Spark SQL 查詢被轉換爲 Spark 操做。Spark Streaming 對實時數據流進行處理和控制，它容許程序可以像普通 RDD 同樣處理實時數據。

接下來的系列文章將會詳細介紹 Spark 生態中的其餘模塊與各個子項目，接下來將經過與 MapReduce 的對比來介紹數據處理引擎Spark的特色及其原理。

Spark 的特色

根據谷歌和百度的搜索結果顯示，Spark 的搜索趨勢已與 Hadoop 持平甚至趕超，標誌着 Spark 已經成爲計算部分的事實標準，也就是說大數據技術繞不開 Spark 了。

在大數據的存儲、計算、資源調度中，Spark 主要解決計算問題，即主要替代 Mapreduce 的功能，底層存儲和資源調度不少公司仍然選擇使用 HDFS、Yarn 來承載。爲何衆多企業在 Hadoop 生態框架裏都選擇用 Spark 做爲處理引擎？讓咱們仔細看看它有什麼特色。

1.速度快。Spark 基於內存進行計算（ 也有部分計算基於磁盤） ；

2.容易上手開發。 Spark 基於 RDD 的計算模型， 比 Hadoop 基於 Map-Reduce 的計算模型要更易於理解、易於上手開發實現各類複雜功能，如二次排序、 topN 等複雜操做時更加便捷。；

3.超強的通用性。 Spark 提供了 Spark RDD、 Spark SQL、 Spark Streaming、 Spark MLlib、 Spark GraphX 等技術組件， 能夠一站式地完成大數據領域的離線批處理、 交互式查詢、 流式計算、 機器學習、圖計算等常見的任務；

4.集成 Hadoop。 Spark 能夠完美集成 Hadoop。 Hadoop 的 HDFS、 Hive、HBase 負責存儲， Yarn 負責資源調度， Spark 負責大數據計算是比較流行的大數據解決方案。

4.極高的活躍度。 Spark 目前是 Apache 基金會的頂級項目， 全世界有大量的優秀工程師是 Spark 的 committer， 而且世界上不少頂級的 IT 公司都在大規模地使用Spark。

看看一樣是負責計算問題的 MapReduce，以下圖所示是 MapReduce 計算 WordCount。

MapReduce 解決了大數據處理中多種場景問題，可是它的侷限性也很明顯：

• MapReduce 只提供 Map 和 Reduce 兩個操做，欠缺表達力，複雜的計算須要大量的 Job 才能完成。
• 中間結果也放在 HDFS 文件系統中，迭代計算的話效率很低。
• 適用 Batch 數據處理，對於交互式數據處理而言實時數據處理的支持不夠。
• 須要寫不少底層代碼，難上手。如上所示的 WordCount 程序至少須要三個 java 類：Map 類、Reduce 類、Job 類，這裏不詳細列出。

許多項目針對它的侷限性進行了改進（如 Tez 等），接着看 Spark 的具體操做流程：

首先咱們能夠看到 Spark 提供了豐富的算子（textFile、FlatMap、Map、ReduceByKey 等），在計算的中間結果也沒有存儲到 HDFS 的操做。而後，對於上圖的 WordCount 程序，Spark 只須要以下一行代碼：

sc.textFile(s"${path}").flatMap(_.split(" ")).map(word => (word, 1)).reduceByKey(_ + _).saveAsTextFile("hdfs://xxx")

下表列舉了 Spark 和 MapReduce 做爲數據處理引擎的一些對比。值得一提的是關於數據處理的規模，Spark 在誕生後，社區裏有不少質疑 Spark 處理數據規模的聲音，隨後官方給出了對於一 PB 數據排序的實驗，而且處理時間打破了當時的記錄。但咱們也不能忽視，在實際生產過程當中，咱們面對的不是一個程序或者一個任務，在同一個集羣，若是有不少的 Spark 程序沒有獲得很好的優化，會浪費大量的內存，從而讓一些程序須要排隊等待，在這種狀況下，Spark 處理的數據規模可能會小於 MapReduce 處理的數據規模。（以後的系列文章也會介紹關於 Spark 內存調優的相關內容）

關於最後一點容錯性，MapReduce 中每一步操做的結果都會被存入磁盤，在計算出現錯誤時能夠很好的從磁盤進行恢復；Spark 則須要根據 RDD 中的信息進行數據的從新計算，會耗費必定的資源。Spark 提供兩種方式進行故障恢復：經過數據的血緣關係再執行一遍前面的處理；Checkpoint 將數據集存儲到持久存儲中。理論上若是選擇在每一個完成的小步驟上加 CheckPoint，Spark 的容錯性能能夠和 MR 達到同樣的穩健。固然，不多有人會這麼作。

咱們經過 Spark 與 MapReduce 對比。看到了 Spark 對 MapReduce 侷限性的改進，還有它快速、通用的特色。接下來將經過 Spark 的設計思想和執行過程來講明它爲何能夠作到這些特色。

Spark 的基本原理

如上圖所示，在 Spark 集羣中由一個節點做爲 driver 端建立 SparkContext。Spark 應用程序的入口負責調度各個運算資源，協調各個 Worker Node上 的 Executor。根據用戶輸入的參數會產生若干個 workr，workr 節點運行若干個 executor，一個 executor 是一個進程，運行各自的 task，每一個 task 執行相同的代碼段處理不一樣的數據。

如上圖所示是 Spark 的具體執行過程，client 提交做業，經過反射 invoke 執行用戶代碼 main 函數，以後開始啓動 CoarseGrainedExecutorBackend 和初始化 SparkContext。

SparkContext 初始化包括初始化監控頁面 SparkUI、執行環境 SparkEnv、安全管理器 SecurityManager、stage 劃分及調度器 DAGScheduler、task 做業調度器 TaskSchedulerImpl、與 Executor 通訊的調度端 CoarseGrainedSchedulerBackend。

DAG Scheduler 將做業劃分後，依次提交 stage 對應的 taskSet 給 TaskSchedulerImpl，TaskSchedulerImpl 會 submit taskset 給 driver 端的 CoarseGrainedSchedulerBackend 後端，接着 CoarseGrainedSchedulerBackend 會一個一個的 LaunchTask。在遠端的 CoarseGrainedExecutorBackend 接收到 task 提交 event 後，會調用 Executor 執行 task，最終 task 是在 TaskRunner 的 run 方法內運行。

那麼在過程 4 中 DAG Scheduler 如何劃分做業？若是產生 stage、task 等給 Executor 執行呢？接着咱們看做業劃分執行的示例。

上圖描述了一個 Spark 程序，從 HDFS 上讀取數據產生 RDD-A 而後 flatmap 操做到 RDD-B，讀取另外一部分數據的到RDD-C，而後 map 操做的到 RDD-D，RDD-D 聚合操做 RDD-E，RDD-B 和 RDD-E 加入後獲得 RDD-F，而後再將結果存儲到 HDFS 上。

Spark 根據 RDD 之間的不一樣點依賴關係切分紅不一樣的階段（Stage），途中有四個階段，其中 Stage0 和 Stage2 因爲沒有依賴關係是能夠並行執行的。但Stage2須要等待Stage1執行完畢。RDD-D 到 RDD- F 的聚合操做以及 Stage0 和 Stage2 獲得的 RDD- B 和 RDD-E join在一塊兒的到 RDD-F，這個過程會產生 shaffle。沒有依賴關係的Stage是能夠並行執行的，可是對於job，Spark是串行執行的，若是想要並行執行Job，能夠在Spark程序中進行多線程編程。

在這個 DAG 圖中，Spark 可以充分了解數據之間的血緣關係，這樣某些任務失敗後能夠根據血緣關係從新執行計算獲取失敗了的 RDD。

寬依賴和窄依賴
窄依賴是指父RDD的每一個分區只被子RDD的一個分區所使用，子RDD分區一般對應常數個父RDD分區；
寬依賴是指父RDD的每一個分區均可能被多個子RDD分區所使用，子RDD分區一般對應全部的父RDD分區。這個概念在下面的例子中會涉及。

Spark 提供了豐富的算子，操做也更加通用。那麼這種劃分做業、執行並行計算的方案如何使 Spark 產生基於內存計算的快速效果呢？都說 Spark 擅長迭代計算，那麼咱們經過一個經典的迭代問題 PageRank 算法來與 MapReduce 比較一下。

via http://www.jos.org.cn/jos/ch/...

上圖是 MapReduce 進行 pagerank 算法的一次迭代過程，須要注意的是灰色的部分都是須要存儲到磁盤的數據。

via http://www.jos.org.cn/jos/ch/...

上圖所示是 Spark 執行 pageRank 算法的一次迭代過程，相較於 MapReduce 作了不少改進。首先在內存足夠的狀況下 Spark 容許用戶將經常使用的數據緩存到內存中,加快了系統的運行速度；其次 Spark 對數據之間的依賴關係有了明確的劃分，根據寬依賴與窄依賴關係進行任務的調度，能夠實現管道化操做，使系統靈活性得以提升。

MapReduce 進行 pagerank 算法的二次迭代，via http://www.jos.org.cn/jos/ch/...

Spark 進行 pagerank 算法的二次迭代，via http://www.jos.org.cn/jos/ch/...

如圖所示 Spark 能夠將具備窄依賴關係的 RDD 分區分配到一個任務中,進行管道化操做，任務內部數據無需經過網絡傳輸且任務之間互不干擾，所以 Spark 兩次迭代只有三次 shuffle。

在一次迭代過程當中，MapReduce 與 Spark 在性能上可能並無很大的差異，可是隨着迭代次數的增長，二者的差距逐漸顯現出來。Spark 根據依賴關係採用的任務調度策略使得 shuffle 次數相較於 MapReduce 有了顯著下降，所以 Spark 的設計十分適合進行迭代運算。

回顧本篇文章，咱們依次從概念、特色及原理三個角度初步介紹了 Spark，下一篇咱們將具體介紹 Spark on Yarn 的運做流程與機制，敬請期待。

附：Spark 相關術語表