前言

這里只向大家介紹幾個關鍵的關鍵字知識點，詳細系統學習還是建議去B站

關鍵字

DAG（有向無環圖） cache（快取） checkpoint（檢查點） RDD（彈性資料集）算子

其中算子可以理解為Java里方法的呼叫，每點. 一次就是呼叫一次方法，Spark的算子與Scala的算子有很多共同之處，在學習Spark之前可以參考我前面 Scala常用算子 博客，里面對算子進行了分類

DAG（有向無環圖）

概念

DAG 是一組頂點和邊的組合，頂點代表了 RDD，邊代表了對 RDD 的一系列操作，

DAG Scheduler 會根據 RDD 的 transformation 動作，將 DAG 分為不同的 stage，每個 stage 中分為多個 task，這些 task 可以并行運行，

DAG 資料結構不是鏈表，而是 Graph 圖，Graph 上有很多節點，也叫做 Vertices 頂點，連接兩個節點的叫做 edges 邊，沒錯，就是咱們中小學數學課上學到的頂點和邊的概念，鏈表，Tree ，圖是三個復雜度遞進的資料結構，鏈表就是一條有方向的線，Tree 是有分叉的，但是任意兩個節點間只有一條路徑能到達另外一點，也就是不能形成閉合的圖形，

DAG 解決了什么問題

DAG 的出現主要是為了解決 Hadoop MapReduce 框架的局限性，那么 MapReduce 有什么局限性呢？

主要有兩個：

每個 MapReduce 操作都是相互獨立的，HADOOP不知道接下來會有哪些Map Reduce，
每一步的輸出結果，都會持久化到硬碟或者 HDFS 上，

當以上兩個特點結合之后，我們就可以想象，如果在某些迭代的場景下，MapReduce 框架會對硬碟和 HDFS 的讀寫造成大量浪費，

而且每一步都是堵塞在上一步中，所以當我們處理復雜計算時，會需要很長時間，但是資料量卻不大，

所以 Spark 中引入了 DAG，它可以優化計算計劃，比如減少 shuffle 資料，

作業流程

解釋器是第一層，Spark 通過使用Scala解釋器，來解釋代碼，并會對代碼做一些修改，
在Spark控制臺中輸入代碼時，Spark會創建一個 operator graph，來記錄各個操作，
當一個 RDD 的 Action 動作被呼叫時， Spark 就會把這個 operator graph 提交到 DAG scheduler 上，
DAG Scheduler 會把 operator graph 分為各個 stage，一個 stage 包含基于輸入資料磁區的task，DAG scheduler 會把各個操作連接在一起，
這些 Stage 將傳遞給 Task Scheduler，Task Scheduler 通過 cluster manager 啟動任務，Stage 任務的依賴關系， task scheduler 是不知道的，
在 slave 機器上的 Worker 們執行 task，

Cache 快取

為什么要用cache

spark中一個job是由RDD的一連串transformation操作和一個action操作組成，只有當執行到action操作代碼時才會觸發生成真正的job，從而根據action操作需要的RDD及其依賴的所有RDD轉換操作形成實際的任務，也就是會從源頭輸入資料開始執行整個計算程序，并沒有如我們想的單機程式那樣達到RDD復用的目的，

為了達到RDD復用的目的，就需要對想要復用的RDD進行cache，RDD的快取與釋放都是需要我們顯示操作的，

cache的使用

spark的cache使用簡單，只需要呼叫cache或persist方法即可，而且可以看到兩個方法實際都是呼叫的都是persist方法

def cache(): this.type = persist()
def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

注意點

cache之后一定不能直接去接算子，因為cache后有算子的話，它每次都會重新觸發這個計算程序，從而導致cache失效，

cache操作需要當第一個使用到它的job執行后才會生效，而不是cache后馬上可用，這是spark框架的延遲計算導致的，可能粗想起來也不會有什么問題，但是不正確的使用unpersist操作，也可能會導致cache失效，如下例子所示，在action操作之前就把快取釋放掉：

val data = sc.textFile(“data.csv”)
.flatMap(.split(“,”))
.cache() val data1 = data.map(word => (word, 1))
.reduceByKey( + ) val data2 = data.map(word => (word, word.length)).reduceByKey( + _)

data.unpersist()

val wordCount1 = data1.count()

val wordCount2 = data2.count()

如何釋放cache快取：unpersist，它是立即執行的，persist是lazy級別的（沒有計算），unpersist是eager級別的，RDD cache的生命周期是application級別的，也就是如果不顯示unpersist釋放快取，RDD會一直存在（雖然當記憶體不夠時按LRU演算法進行清除），如果不正確地進行unpersist，讓無用的RDD占用executor記憶體，會導致資源的浪費，影響任務的效率，

cache存盤級別

每個持久化的 RDD 可以使用不同的存盤級別進行快取，例如，持久化到磁盤、已序列化的 Java 物件形式持久化到記憶體（可以節省空間）、跨節點間復制、以 off-heap 的方式存盤在 Tachyon，這些存盤級別通過傳遞一個 StorageLevel 物件給 persist() 方法進行設定，

詳細的存盤級別介紹如下

MEMORY_ONLY

使用未序列化的Java物件格式，將資料保存在記憶體中，如果記憶體不夠存放所有的資料，則資料可能就不會進行持久化，那么下次對這個RDD執行算子操作時，那些沒有被持久化的資料，需要從源頭處重新計算一遍，如果RDD中資料量比較大時，會導致JVM的OOM記憶體溢位例外，這是RDD的默認持久化級別，

MEMORY_AND_DISK

使用未序列化的Java物件格式，優先嘗試將資料保存在記憶體中，如果記憶體不夠存放所有的資料，會將資料寫入磁盤檔案中，下次對這個RDD執行算子時，持久化在磁盤檔案中的資料會被讀取出來使用，這是DataFrame的默認持久化級別，

MEMORY_ONLY_SER/MEMORY_AND_DISK_SER

基本含義同MEMORY_ONLY/MEMORY_AND_DISK，唯一的區別是，會將RDD中的資料進行序列化，RDD的每個partition會被序列化成一個位元組陣列，這種方式更加節省記憶體，從而可以避免持久化的資料占用過多記憶體導致頻繁GC，

這兩種策略都是對MEMORY_ONLY/MEMORY_AND_DISK策略的補充，

DISK_ONLY

使用未序列化的Java物件格式，將資料全部寫入磁盤檔案中，

MEMORY_ONLY_2/MEMORY_AND_DISK_2 etc..

對于上述任意一種持久化策略，如果加上后綴_2，代表的是將每個持久化的資料，都復制一份副本，并將副本保存到其他節點上，這種基于副本的持久化機制主要用于進行容錯，假如某個節點掛掉，節點的記憶體或磁盤中的持久化資料丟失了，那么后續對RDD計算時還可以使用該資料在其他節點上的副本，如果沒有副本的話，就只能將這些資料從源頭處重新計算一遍了，

通過原始碼，實際上一共有12中快取級別

object StorageLevel {
val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
.....

}

如何選擇存盤級別

針對占用記憶體不是很大的中間計算結果優先采用MEMORY_ONLY，它的性能最好（前提是記憶體要足夠），也是RDD的默認策略，如果中間結果資料量稍大，可以采用MEMORY_ONLY_SER策略進行補充，但在實際的生產環境中，大多數情況下資料量都是超出記憶體容量的，這可能會導致JVM的OOM記憶體溢位例外，

如果記憶體無法容納中間資料，那么建議使用MEMORY_AND_DISK_SER策略，該策略會優先將資料快取在記憶體中，只有當記憶體不夠時才會把資料寫入磁盤，另外對于分布式任務，IO資源往往比CPU資源更加緊張，序列化后的資料可以節省記憶體和磁盤的空間開銷，

通常很少使用DISK_ONLY級別，它表示資料量已經非常大，遠大于記憶體的容量，這個時候需要慎重權衡重新計算RDD的消耗和從磁盤加載RDD的消耗，

除非對于高可用性的任務，否則不建議使用后綴為_2的級別，因為在記憶體中復制多份資料很難有足夠的記憶體資源滿足，而對于HDFS檔案本身已經有多備份保證資料的可靠性，

對于實際快取的效果，可以查看spark UI中的storage頁面，里面詳細描述了快取的每個RDD的資料快取分布情況，

checkPoint（檢查點）

spark計算里面計算流程DAG特別長,服務器需要將整個DAG計算完成得出結果,

如果中間算出的資料突然丟失了,spark又會根據RDD的依賴關系從頭到尾計算一遍,這樣子就很費性能

當然我們可以將中間的計算結果通過cache或者persist放到記憶體或者磁盤中,但是這樣也不能保證資料完全不會丟失,存盤的這個記憶體出問題了或者磁盤壞了,也會導致spark從頭再根據RDD計算一遍

所以就有了checkpoint,其中checkpoint的作用就是將DAG中比較重要的中間資料做一個檢查點將結果存盤到一個高可用的地方(HDFS，利用HDFS的默認存盤策略)

cache和checkpoint的區別：

快取（cache）把 RDD 計算出來然后放在記憶體中，但是RDD 的依賴鏈（相當于資料庫中的redo 日志），也不能丟掉，當某個點某個 executor 宕了，上面cache 的RDD就會丟掉，需要通過依賴鏈重放計算出來，不同的是，checkpoint是把 RDD 保存在 HDFS中，是多副本可靠存盤，所以依賴鏈就可以丟掉了，就斬斷了依賴鏈，是通過復制實作的高容錯，

RDD（彈性資料集）

Resilient Distributed Datasets 彈性分布式資料集，是一個容錯的、并行的資料結構，可以讓用戶顯式地將資料存盤到磁盤和記憶體中，并能控制資料的磁區，同時，RDD還提供了一組豐富的操作來操作這些資料，RDD是只讀的記錄磁區的集合，只能通過在其他RDD執行確定的轉換操作（transformation操作）而創建，RDD可看作一個spark的物件，它本身存在于記憶體中，如對檔案計算是一個RDD，等，

一個RDD可以包含多個磁區，每個磁區就是一個dataset片段，RDD可以相互依賴，如果RDD的每個磁區最多只能被一個Child RDD的一個磁區使用，則稱之為narrow dependency；若多個Child RDD磁區都可以依賴，則稱之為wide dependency，

寬窄依賴

RDD的依賴關系分為兩種模型，一種是窄依賴（narrow dependency）和寬依賴（wide dependency），

1.窄依賴（1對1 | N對1）

指父RDD的每一個磁區最多被一個子RDD的磁區所用，表現為一個父RDD的磁區對應于一個子RDD的磁區（第一類），或者是多個父RDD的磁區對應于一個RDD的磁區（第二類），也就是說一個父RDD的一個磁區不可能對應于一個子RDD的多個磁區，

如下圖所示，對輸入進行協同劃分（co-partitioned）的join屬于第二類，當子RDD的磁區依賴于單個父RDD的磁區的時候，磁區的結構不會發生改變，如下圖中的map,filter等操作，相反的，對于一個子RDD的磁區依賴于多個RDD的磁區的時候，磁區的結構會發生改變

2.寬依賴（一對多）

寬依賴是值子RDD的每一個磁區都要依賴于所有父RDD的所有磁區或者多個磁區，也就是說存在一個父RDD的一個磁區對應著一個子RDD的多個磁區，如下圖的groupByKey就屬于寬依賴，其中寬依賴會觸發shuffle操作

磁區優化

計算資料規模

一般單磁區資料規模一般為128M（一個Block一個磁區，完全做到本地化）借助Hadoop FileSystem 統計檔案位元組大小 Long FILE_SIZE

如果 FILE_SIZE/BLOCK_SIZE > 物理執行緒核心core * 2(執行緒) 則設定為物理執行緒核心core * 2(執行緒) 否則設定 Math.ceil（FILE_SIZE*1.0/BLOCK_SIZE）

Stage （階段）

概念

1.一個 job，就是由一個 rdd 的 action 觸發的動作，可以簡單的理解為，當你需要執行一個 rdd 的 action 的時候，會生成一個 job，

2.stage : stage 是一個 job 的組成單位，就是說，一個 job 會被切分成 1 個或 1 個以上的 stage，然后各個 stage 會按照執行順序依次執行，

3.task :即 stage 下的一個任務執行單元，一般來說，一個 rdd 有多少個partition，就會有多少個 task，因為每一個 task 只是處理一個partition 上的資料，

劃分規則

1.從后向前推理，遇到寬依賴就斷開，遇到窄依賴就把當前的RDD加入到Stage中；

2.每個Stage里面的Task的數量是由該Stage中最后一個RDD的Partition數量決定的；

3.最后一個Stage里面的任務的型別是ResultTask，前面所有其他Stage里面的任務型別都是ShuffleMapTask；

4.代表當前Stage的算子一定是該Stage的最后一個計算步驟；

總結：由于spark中stage的劃分是根據shuffle來劃分的，而寬依賴必然有shuffle程序，因此可以說spark是根據寬窄依賴來劃分stage的，

Spark案例

WordCount 、求每個用戶的觀影次數

資料提供，提取碼 arie

object App {
  def main(args: Array[String]): Unit = {
    val config: SparkConf = new SparkConf()
      .setAppName("spark_rdd_03")
      .setMaster("local[*]")
    val sc = new SparkContext(config)
    //Word Count
   wordCount
    sc.textFile("hdfs://single01:9000//spark/Broken to Harness.txt",3)
        .mapPartitions(it=>{
          it
            .filter(_.trim.size>0)
            .flatMap(_.replaceAll(",|\\.|!|\\?|;|\"","")
            .replaceAll("\\s{2,}"," ")
            .split(" ")
              .map((_,1))
                .groupBy(_._1)
                .map(tp2=>(tp2._1,tp2._2.size))
            )
        }).reduceByKey(_+_)
        .foreach(println)



    //每個用戶的觀影次數
    sc.textFile("hdfs://single01:9000//spark/tags.csv",3)
        .mapPartitionsWithIndex((index,it)=>{
          if (index==0) it.drop(1)  //洗掉第一個磁區的表頭
          it
            .map(line=>{
            val ps:Array[String]=line.split(",")
            (ps(0),1)
          }).toArray
            .groupBy(_._1)
            .map(tp2=>(tp2._1,tp2._2.size))
            .toIterator

        })
      .reduceByKey(_+_)
        .foreach(println)

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/437178.html

標籤：其他

上一篇：HBase 過濾器 Java API

下一篇：在ReactNativeClass中呼叫或分配變數值時遇到問題

Spark入門必知知識點

目錄

前言

關鍵字