spark計算模型RDD

RDD介紹

1.RDD概念以及特性

RDD（Resilient Distributed Dataset）叫做彈性分散式數據集，是Spark中最基本的數據抽象，它代表一個不可變、可分區、裡面的元素可並行計算的集合。RDD具有數據流模型的特點：自動容錯、位置感知性調度和可伸縮性。RDD允許用戶在執行多個查詢時顯式地將數據緩存在記憶體中，後續的查詢能夠重用這些數據，這極大地提升了查詢速度。（A Resilient Distributed Dataset）彈性分散式數據集合。並且是spark最基本的編程抽象，而且RDD是只讀、可分區的、可以進行並行計算的一個對象。

• 數據集：一個數據集合，用於存放數據的。RDD是一個數據容器，用來組織管理數據的。跟Array和List類似，並且都能夠進行map、flatMap、filter等等

• 分散式：RDD中的數據是分散式存儲的，可用於分散式計算。RDD的數據是分佈存儲的，也就是Spark集群中每個節點上只存儲了RDD的部分數據。計算同樣也是分散式並行計算的

• 彈性：

  • 存儲的彈性：RDD的數據可以在記憶體和磁碟之間進行自由切換

  • 可靠性的彈性：RDD的在丟失數據的時候能夠自動恢復。RDD在計算過程中會出現失敗的情況，失敗以後會進行一定次數的重試（4次）

  • 並行度的彈性：RDD的數據分區可以改變，進而增加並行計算的粒度

• RDD其他特點：

  • RDD的數據是只讀，每次操作都會產生新的RDD。安全。

  • RDD中數據可以緩存在記憶體、磁碟、HDFS之上

1.1RDD彈性

1) 自動進行記憶體和磁碟數據存儲的切換

​ Spark優先把數據放到記憶體中，如果記憶體放不下，就會放到磁碟裡面，程式進行自動的存儲切換

2) 基於血統的高效容錯機制

​ 在RDD進行轉換和動作的時候，會形成RDD的Lineage依賴鏈，當某一個RDD失效的時候，可以通過重新計算上游的RDD來重新生成丟失的RDD數據。

3) Task如果失敗會自動進行特定次數的重試

​ RDD的計算任務如果運行失敗，會自動進行任務的重新計算，預設次數是4次。

4) Stage如果失敗會自動進行特定次數的重試

​ 如果Job的某個Stage階段計算失敗，框架也會自動進行任務的重新計算，預設次數也是4次。

5) Checkpoint和Persist可主動或被動觸發

​ RDD可以通過Persist持久化將RDD緩存到記憶體或者磁碟，當再次用到該RDD時直接讀取就行。也可以將RDD進行檢查點，檢查點會將數據存儲在HDFS中，該RDD的所有父RDD依賴都會被移除。

6) 數據調度彈性

​ Spark把這個JOB執行模型抽象為通用的有向無環圖DAG，可以將多Stage的任務串聯或並行執行，調度引擎自動處理Stage的失敗以及Task的失敗。

7) 數據分片的高度彈性

​ 可以根據業務的特征，動態調整數據分片的個數，提升整體的應用執行效率。

​ RDD是一種分散式的記憶體抽象，表示一個只讀的記錄分區的集合，它只能通過其他RDD轉換而創建，為此，RDD支持豐富的轉換操作(如map, join, filter, groupBy等)，通過這種轉換操作，新的RDD則包含瞭如何從其他RDDs衍生所必需的信息，所以說RDDs之間是有依賴關係的。基於RDDs之間的依賴，RDDs會形成一個有向無環圖DAG，該DAG描述了整個流式計算的流程，實際執行的時候，RDD是通過血緣關係(Lineage)一氣呵成的，即使出現數據分區丟失，也可以通過血緣關係重建分區，總結起來，基於RDD的流式計算任務可描述為：從穩定的物理存儲(如分散式文件系統)中載入記錄，記錄被傳入由一組確定性操作構成的DAG，然後寫回穩定存儲（HDFS或磁碟）。另外RDD還可以將數據集緩存到記憶體中，使得在多個操作之間可以重用數據集，基於這個特點可以很方便地構建迭代型應用(圖計算、機器學習等)或者互動式數據分析應用。可以說Spark最初也就是實現RDD的一個分散式系統，後面通過不斷發展壯大成為現在較為完善的大數據生態系統，簡單來講，Spark-RDD的關係類似於Hadoop-MapReduce關係。

1.2RDD的五大屬性

1) 一組分片（Partition），即數據集的基本組成單位。對於RDD來說，每個分片都會被一個計算任務處理，並決定並行計算的粒度。

如果文件的block個數 <=2 那麼 sc.textFile(“file:///wordcount.txt”)分區個數為2

如果文件的block塊個數 >2 那麼 sc.textFile(“file:///wordcount.txt”)分區的個數等於block塊的個數

2) 一個計算每個分區的函數。Spark中RDD的計算是以分片為單位的，每個RDD都會實現compute函數以達到這個目的。compute函數會對迭代器進行複合，不需要保存每次計算的結果。RDD的每一個運算元操作比如map 都會通過compute方法作用在每個分區之上

3) RDD之間的依賴關係。RDD的每次轉換都會生成一個新的RDD，所以RDD之間就會形成類似於流水線一樣的前後依賴關係。在部分分區數據丟失時，Spark可以通過這個依賴關係重新計算丟失的分區數據，而不是對RDD的所有分區進行重新計算。每一個RDD都有其依賴列表RDD的依賴關係 都是存在一個序列集合中,作用：容錯 以及構建起血統機制

4) 一個Partitioner，即RDD的分片函數。當前Spark中實現了兩種類型的分片函數，一個是基於哈希的HashPartitioner，另外一個是基於範圍的RangePartitioner。只有對於於key-value的RDD，才會有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數不但決定了RDD本身的分片數量，也決定了parent RDD Shuffle輸出時的分片數量。

5) 一個列表，存取每個Partition的優先位置（preferred location）。對於一個HDFS文件來說，這個列表保存的就是每個Partition所在的塊的位置。按照“移動數據不如移動計算”的理念，Spark在進行任務調度的時候，會儘可能地將計算任務分配到其所要處理數據塊的存儲位置。

 a list of preferred locations to compute each split on (e.g. block locations for
  *    an HDFS file)
  Spark在讀取hdfs文件的是，hdfs文件每一個block預設有多個備份，spark會獲取每一個block塊以及其備份的位置信息構建成列表，在進行計算的時候，spark會在位置列表中選取一個最佳位置進行任務分配。 移動數據不如移動計算的原則。
       移動數據不如移動計算的原則最高境界：數據在當前運行程式的進程之中
  RDD是如何確定優先位置？
      getPreferredLocations(split: Partition): Seq[String] 
  通過以上方法確定計算的最佳位置。
  RDD的數據本地化:
      5種

2.RDD的構建方式

3種構建方式

• 根據以後數據集合構建RDD

  • val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8))

  • val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8))

• 根據外部文件 可以是本地文件也可是HDFS上文件

  • sc.textFile(filePath)

• 根據以後RDD創建新的RDD 需要經過運算元操作

  • val newRDD=lineRDD.flatMap(function)

3.RDD的運算元操作

RDD的運算元分為兩類

• 轉換運算元（Transform運算元）

  • 將一個RDD通過轉換運算元操作以後會構建新的RDD，比如map 、flatMap、reduceByKey

  • 轉換運算元操作都是直接new新的RDD，此時RDD並沒有進行真正的計算。轉換運算元只是對數據如何計算做了標記。轉換運算元都是懶載入。

• 重要運算元操作

  • mapPartitions ：作用於每個分區之上的

    • mapPartitions 和map區別：

      • mapPartitions 相當於partition批量操作

      • map作用於每一條數據

      • 重要區別：mapPartitions 這個在大量task運行的時候可能會出現記憶體溢出的情況。小數據量的操作 mapPartitions 要優於map操作

  • groupByKey運算元和ReduceByKey運算元的區別

    • 1.groupByKey 返回值：key->集合 ReduceByKey返回值： key-》值

    • 2.ReduceByKey操作會在本地進行初步merge操作，能夠減少網路數據的傳輸

  • coalesce 減少分區數據的運算元

    • 該運算元可以進行shuffle也可以不進shuffle操作， coalesce(numPartitions: Int, shuffle: Boolean = false）

  • repartition 實際上是調用了 coalesce 運算元 ，而且 repartition一定會進行shuffle操作，既可以增加也可以減少分區

• Action運算元

　　　　action運算元內部都會有一個runJob方法進行提交一個Job任務

　　廣播變數：

• 廣播變數需要數據傳遞

  • HTTP協議：基於HTTP協議將數據傳遞到Executor。Executor會Driver端申請下載（已經被廢棄）

  • torrent協議：預設的方式。 Driver下載到Executor上，然後Executor會再次數據源，將數據傳遞到下一個需要數據Executor之上。參考 （TorrentBroadcast類）

4.RDD的依賴關係

RDD和它依賴的父RDD的關係有兩種不同的類型，即窄依賴（narrow dependency）和寬依賴（wide dependency）。

• RDD的依賴類型

  • 窄依賴：父RDD中一個partition最多被子RDD中的一個partition所依賴，這種依賴關係就是窄依賴

  • 窄依賴運算元：map 、filter 、union 、flatMap等

  • 寬依賴：父RDD中一個partition被子RDD中的多個partition所依賴，這種依賴關係就是寬依賴

  • 寬依賴運算元：groupByKey、reduceByKey。凡是By基本上都是寬依賴

  一對一或者多對==一：窄依賴==

  一對多或者多對==多：寬依賴==

• 寬窄依賴運算元的判斷依據是轉換運算元是否會產生shuffle操作，如果有shuffle操作則是寬依賴，否則是窄依賴

• join既是寬依賴運算元也是窄依賴運算元 （在一個shuffle操作之後，在使用Join的時候，此時join就是窄依賴）