大資料學習之spark

Spark簡介

Apache Spark 是專為大規模資料處理而設計的快速通用的計算引擎， Spark 擁有 Hadoop MapReduce 所具有的優點；但不同于 MapReduce 的是 Spark的Job 中間輸出結果可以快取在記憶體中，從而不再需要讀寫 HDFS ，減少了磁盤資料互動，因此 Spark 能更好地適用于資料挖掘與機器學習等需要迭代的演算法，

Spark的組件可以分為SparkCore快速執行引擎，支持多種語言；SparkStreaming流式計算框架；SparkGraph圖形計算框架；MLbase機器學習；SparkSql使用sql處理業務，本文談談SparkCore，SparkSql，SparkStreaming和Spark優化相關知識，

SparkCore

Spark系統架構

1.Master ( standalone 模式)：資源管理的主節點（行程），

2.Cluster Manager ：在集群上獲取資源的外部服務（例如： standalone ； yarn ； mesos ），

3.Worker ( standalone 模式）：資源管理的從節點（行程）或者說是是管理本機資源的行程，

4.Application ：基于 Spark 的用戶程式，包含 driver 程式和運行在集群上的 executor 程式，即一個完整的 spark 應用 ，

5.Dirver （ program ）：用來連接作業行程（ worker ）的程式 ，

6.Executor ：是在一個 worker 行程所管理的節點上為某 Application 啟動的一個個行程，這個行程負責運行任務，并且負責將資料存在記憶體或者磁盤上，每個應用之間都有各自獨立的executors ，

7.Task ：被發送到 executor 上的作業單元，

8.Job ：包含很多任務（ Task ）的并行計算，和 action 算子對應，

9.Stage ：一個 job 會被拆分成很多組任務，每組任務被稱為 Stage （就像 MapReduce 分為MapTask 和ReduceTask 一樣），

Spark相關名詞概念

Partition

由于Spark RDD 是一種分布式的資料集，資料量很大，因此要它被切分并存盤在各個結點的磁區當中，

Spark包含兩種資料磁區方式：HashPartitioner（哈希磁區）和RangePartitioner（范圍磁區），

哈希磁區是根據hash值進行磁區，這樣如果資料hash值的數目相差太大時會導致資料傾斜，為了避免造成這種情況我們可以通過范圍磁區（基于蓄水池初樣演算法）使得磁區中的資料均勻，

RDD

RDD(Resilient Distributed Dataset) 彈性分布式資料集，

RDD的5大屬性：

1.RDD 是由多個 partition 組成的，

2.RDD中每一個task運行在自己的partition上，

3.RDD需要依賴其他的RDD

4.磁區器必需作用在 (K,V) 格式的 RDD 上，

5.RDD默認會尋找最優的計算位置（計算向資料靠攏，盡量減少資料的拉取操作）

Lineage血統：RDD 的最重要的特性之一就是血緣關系（Lineage )，它描述了一個 RDD 是如何從父 RDD 計算得來的，如果某個 RDD 丟失了，則可以根據血緣關系，從父 RDD 計算得來，

寬窄依賴和stage

窄依賴：父 RDD 和子 RDD 的 partition 之間的關系是一對一的，或者父 RDD 和子 RDD 的 partition 關系是
多對一的，不會有 shuffle 的產生，

寬依賴：父 RDD 與子 RDD 的 partition 之間的關系是一對多，會有 shuffle 的產生，

stage階段切割規則：從后往前，遇到寬依賴就切割 stage （有n個寬依賴就有n+1個stage），

spark算子

單檔案算子：

轉換算子（本質就是函式）， 轉換 算子是延遲執行，也叫懶加載執行（需要行動算子執行才會執行）

常見 的轉換算子有filter ，map ，flatMap，sample，reduceByKey ，sortByKey / sortBy 等

行動算子， 行動算子是觸發執行，一個 application 應用程式中有幾個 行動算子執行，就有幾個 job 運行；

count ，take(n) ，first ，foreach ，collect等；

控制算子：cache，persist，checkpoint等；

這里要聊下cache和checkpoint的異同點：他們都是做 RDD 持久化的，cache是在觸發action之后，把資料寫入到記憶體或者磁盤中，不會截斷血緣關系 ，checkpoint 也是在觸發action之后，執行任務，單獨再啟動一個job，負責寫入資料到hdfs中，會截斷血緣關系，

多檔案算子：

轉換算子join,union,intersection,subtract,mapPartitions,mapPartition,distinct,cogroup等

行動算子foreachPartition等

任務執行流程

啟動集群后， Worker 節點會向 Master 節點匯報資源情況， Master 掌握了集群資源情況，此時當 我們 提交一個 Application 后，集群會根據 RDD 之間的依賴關系將 Application 形成一個 DAG 有向無環圖，每個Application中有一個或多個行動算子，一個行動算子就是一個job任務，每個任務Spark 會在 Driver 端創建兩個物件DAGScheduler 和 TaskScheduler ，DAGScheduler 是任務調度的高層調度器，主要作用就是將 DAG 根據 RDD 之間的寬窄依賴關系劃分為一個個的 Stage ，然后將這些 Stage 以 TaskSet 的形式提交給 TaskScheduler （ TaskScheduler 是任務調度的低層調度器，這里 TaskSet 其實就是一個封裝多個 task 任務的集合，也就是 stage 中的并行的 task 任務），TaskSchedule 會遍歷 TaskSet 集合，拿到每個 task 后會將 task 發送到 Executor 中去執行（其實就是發送到 Executor 中的執行緒池 ThreadPool 去執行），task 在 Executor 執行緒池中的運行情況會向 TaskScheduler 反饋，當 task 執行失敗時，則由TaskScheduler 負責重試，將 task 重新發送給 Executor 去執行，默認重試3次，如果重試3次依然失敗，那么這個 task 所在的 stage 就失敗了，stage 失敗了則由 DAGScheduler 來負責重試，重新發送 TaskSet 到 TaskScheduler ， Stage 默認重試4次，如果重試4次以后依然失敗，那么這個 job 就失敗了， job 失敗了， Application 就失敗了，

這里面的關系可以這樣理解：

每個Application有一個或多個job，每個job中有多個stage，每個stage中有多個RDD，每個RDD中有多個磁區，每個stage最后一個RDD有多少磁區就有多少task，

在窄依賴stage中第一個磁區到最后一個磁區的計算程序稱之為pipeline 管道計算模式，

Driver端所在節點的位置由任務提交方式決定，在clint模式下，在哪個節點啟動行程，哪個節點就是Driver端，生產環境下不建議使用，會使Driver端在同一節點導致客戶端網卡流量暴增的問題，在cluster模式下，會隨機在集群一臺節點啟動 Driver 行程，

SparkSQL

SparkSQL的由來

SparkSQL的前身是shark，Shark 是基于 Spark 計算框架之上且兼容 Hive 語法的 SQL 執行引擎，由于底層的計算采用了Spark ，性能比 MapReduce 的 Hive 普遍快2倍以上，當資料全部加載在記憶體的話，將快10倍以上，因此 Shark 可以作為互動式查詢應用服務來使用， SparkSQL 產生的根本原因是其完全脫離了 Hive 的限制，能夠在 scala / java 中寫 SQL 陳述句，支持簡單的 SQL 語法檢查，能夠在 SQL 中寫 Hive 陳述句訪問Hive 資料，并將結果取回作為 RDD 使用，

Dataset與DataFrame

可以簡單的把DataFrame理解成RDD+表元資料資訊，把資料文本通過構建case class樣例類或StructType來實作DataFrame，進而可以通過sql陳述句進行RDD的操作，代替了使用scala撰寫代碼的操作，

Dataset可以理解為DataFrame的加強版，

這里具體的DSL操作和hive中的hsql操作差不多，也就基本是以前的sql操作，這里就不展開細說了，

SparkStreaming

SparkStreaming簡介

SparkStreaming是準實時資料流處理框架（默認是5秒計算一次，也可自己設定），實時資料的來源可以是Kafka, Flume, Twitter, ZeroMQ或者TCP sockets，在接受資料同時可以使用高級功能的復雜算子來處理流資料，最終處理后的資料可以存放在檔案系統，資料庫等，方便實時展現，

編程模型DStream

DStream（Discretized Stream）作為Spark Streaming的基礎抽象，它代表持續性的資料流，

DStream你可以理解為對RDD的增強，Spark Streaming 計算還是基于Spark Core的，Spark Core 的核心又是RDD. 所以Spark Streaming 肯定也要和RDD扯上關系，Spark Streaming 并沒有直接讓用戶使用RDD而是自己抽象了一套DStream的概念，DStream 和 RDD 是包含的關系，你可以理解為Java里的裝飾模式，也就是DStream 是對RDD的增強，但是行為表現和RDD是基本上差不多的，

DSeram的操作可以類比RDD的操作，這里要注意的是RDD是一批一批資料進行操作，而Dstrean是每隔一端時間處理資料，這里要關注一下Dstream的視窗操作，它的一個視窗會包括多個執行時間段，

SparkStreaming集成Kafka

我們把生產者的資料放入kafka中，由kafka作為中間件，SparkStreaming來消費并計算處理資料，

Kafka與Spark Streaming集成時有兩種方法：舊的基于receiver的方法，新的基于direct stream的方法，

基于receiver的方法：

這里就是kafka向SparkStreaming不停地推送資料，這樣的話當SparkStreaming處理不了這些資料的話，多余的資料會存盤在WAL檔案中，這樣的話kafka的削峰就失去了意義，當大量的資料積壓處理不了時，會導致SparkStreaming集群down掉，所以這種方法不適于處理大量快速產生的資料，

基于direct stream的方法：

這里就是SparkStreaming根據自己處理資料的能力去拉取Kafka中的資料，這樣流程大量簡化了，而且還有一個改進就是Kafka磁區與RDD磁區是一一對應的，更可控，

這里我們就需要自己來管理offset：

1.CheckPoint

checkpoint會記錄每個批次的狀態持久化到HDFS中，如果機器發生故障，或者程式故障停止，下次啟動時候，仍然可以從checkpoint的目錄中讀取故障時候rdd的狀態，便能接著上次處理的資料繼續處理

但這有個很大的弊端就是更新版本時checkpoint不能保持，這樣的話會出現新舊2個checkpoint，這時無論選擇哪一個checkpoint要么丟資料，要么資料重復，可以說非常的雞肋，這只適用于版本不更新的專案，

2.Kafka Itself

這就是將kafka的offset全部交給kafka管理，

但遇到資料峰值很高的時候會導致kafka集群的磁盤io特別高，這樣的話是非常不安全的，

3.Your own data store

這就是我們自己寫代碼管理offset，把每批次offset存盤到一個外部的存盤系統，

當流式專案停止后再次啟動，會首先從外部存盤系統讀取是否記錄的有偏移量，如果有的話，就讀取這個偏移量，然后把偏移量集合傳入到KafkaUtils.createDirectStream中進行構建InputSteam，這樣的話就可以接著上次停止后的偏移量繼續處理，然后每個批次中仍然的不斷更新外部存盤系統的偏移量，這樣以來就能夠無縫銜接了，無論是故障停止還是升級應用，都是透明的處理，除了相對麻煩，這無疑是可迭代和安全的方法了，

SparkStreaming反壓機制

舊版本：

計算程序中出現batch processing time > batch interval的情況時，可以通過設定引數spark.streaming.receiver.maxRate來限制Receiver的資料接收速率，

這樣的弊端是當集群處理能力高于maxRate，會造成資源浪費，

新版本：

通過反壓機制（back-pressure）來動態控制資料接收速率來適配集群資料處理能力，

反壓機制中的重要組件：RateController，RateEstimator，RateLimiter，令牌桶機制，

Spark 優化

資源調優

動態分配資源，給正在運行的Application分配更多的資源，

并行度調優

合理調整partition（task）的數量，減少資源的浪費進而提高spark任務的運行效率，

代碼調優

RDD

避免創建重復的RDD，應該不會有人一直重復創建吧，

對多次使用的RDD持久化到記憶體，基于記憶體中的資料的操作不需要從磁盤檔案中讀取資料，進而提高了性能，

算子

持久化算子推薦使用checkpoint，持久化到HDFS中，砍斷RDD間聯系，

盡量避免使用shuffle類的算子，shuffle類算子會導致磁盤間大量的io操作，

如果一定要使用shuffle類算子盡量使用map-sid預聚合操作，combiner區域聚合，這樣)可以降低shuffle write寫磁盤的資料量，shuffle read拉取資料量的大小，reduce端聚合的次數，

盡量使用高性能的算子，比如使用reduceByKey替代groupByKey，mapPartition替代map，foreachPartition替代foreach，多使用repartition和coalesce算子來操作磁區數量，磁區數量太多會造成磁盤間大量的io操作，磁區數量太少會造成task任務并行度不夠，造成資源浪費，

使用廣播變數

尤其是當變數大時，沒必要每個Executor中都存放這個變數，我們可以把變數放在Driver中，每個Executor都可以使用到，這樣大量減少了記憶體的使用，

使用Kryo優化序列化性能

Kryo序列化機制，比默認的Java序列化機制，速度要快，序列化后的資料要更小，所以Kryo序列化優化以后，可以讓網路傳輸的資料變少；在集群中耗費的記憶體資源大大減少，

使用高性能的庫fastutil

fastutil提供的集合類比我們平時使用的JDK的原生的Map、List、Set記憶體占用更小，并且在進行集合的遍歷、根據索引（或者key）獲取元素的值和設定元素的值的時候，提供更快的存取速度，

資料本地化

task要計算的資料所處位置的級別依次是:

PROCESS_LOCAL：在本行程的記憶體中，

NODE_LOCAL：在本節點所在的磁盤上或在本節點其他行程的記憶體中，

NO_PREFtask:在關系型資料庫中，如mysql，

RACK_LOCAL:在同機架的不同節點的磁盤或者行程的記憶體中.

ANY：跨機架，

我們要盡量使要計算的資料在前面的級別中（計算向資料靠攏，可以節省大量的計算資源），因此我們可以適當地增加每次發送task的等待時間，

記憶體調優

提高Executor總體記憶體的大小， 降低儲存記憶體比例或者降低聚合記憶體比例，