怎么實現Spark Core的原理分析

怎么實現Spark Core的原理分析，相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

Spark為什么會流行?

原因1：優秀的數據模型和豐富計算抽象

Spark  產生之前，已經有MapReduce這類非常成熟的計算系統存在了，并提供了高層次的API(map/reduce)，把計算運行在集群中并提供容錯能力，從而實現分布式計算。

雖然MapReduce提供了對數據訪問和計算的抽象，但是對于數據的復用就是簡單的將中間數據寫到一個穩定的文件系統中(例如HDFS)，所以會產生數據的復制備份，磁盤的I/O以及數據的序列化，所以在遇到需要在多個計算之間復用中間結果的操作時效率就會非常的低。而這類操作是非常常見的，例如迭代式計算，交互式數據挖掘，圖計算等。

認識到這個問題后，學術界的 AMPLab 提出了一個新的模型，叫做 RDD。RDD  是一個可以容錯且并行的數據結構(其實可以理解成分布式的集合，操作起來和操作本地集合一樣簡單)，它可以讓用戶顯式的將中間結果數據集保存在內存中，并且通過控制數據集的分區來達到數據存放處理最優化。同時  RDD也提供了豐富的 API (map、reduce、filter、foreach、redeceByKey...)來操作數據集。后來RDD被 AMPLab  在一個叫做 Spark 的框架中提供并開源。

簡而言之，Spark 借鑒了 MapReduce  思想發展而來，保留了其分布式并行計算的優點并改進了其明顯的缺陷。讓中間數據存儲在內存中提高了運行速度、并提供豐富的操作數據的API提高了開發速度。

原因2：完善的生態圈-fullstack

目前，Spark已經發展成為一個包含多個子項目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子項目。

Spark Core：實現了 Spark 的基本功能，包含RDD、任務調度、內存管理、錯誤恢復、與存儲系統交互等模塊。

Spark SQL：Spark 用來操作結構化數據的程序包。通過 Spark SQL，我們可以使用 SQL操作數據。

Spark Streaming：Spark 提供的對實時數據進行流式計算的組件。提供了用來操作數據流的 API。

Spark MLlib：提供常見的機器學習(ML)功能的程序庫。包括分類、回歸、聚類、協同過濾等，還提供了模型評估、數據導入等額外的支持功能。

GraphX(圖計算)：Spark中用于圖計算的API，性能良好，擁有豐富的功能和運算符，能在海量數據上自如地運行復雜的圖算法。

集群管理器：Spark 設計為可以高效地在一個計算節點到數千個計算節點之間伸縮計算。

StructuredStreaming：處理結構化流,統一了離線和實時的API。

Spark VS Hadoop

注意：

盡管Spark相對于Hadoop而言具有較大優勢，但Spark并不能完全替代Hadoop，Spark主要用于替代Hadoop中的MapReduce計算模型。存儲依然可以使用HDFS，但是中間結果可以存放在內存中;調度可以使用Spark內置的，也可以使用更成熟的調度系統YARN等。

實際上，Spark已經很好地融入了Hadoop生態圈，并成為其中的重要一員，它可以借助于YARN實現資源調度管理，借助于HDFS實現分布式存儲。

此外，Hadoop可以使用廉價的、異構的機器來做分布式存儲與計算，但是，Spark對硬件的要求稍高一些，對內存與CPU有一定的要求。

Spark Core

一、RDD詳解

1. 為什么要有RDD?

在許多迭代式算法(比如機器學習、圖算法等)和交互式數據挖掘中，不同計算階段之間會重用中間結果，即一個階段的輸出結果會作為下一個階段的輸入。但是，之前的MapReduce框架采用非循環式的數據流模型，把中間結果寫入到HDFS中，帶來了大量的數據復制、磁盤IO和序列化開銷。且這些框架只能支持一些特定的計算模式(map/reduce)，并沒有提供一種通用的數據抽象。

AMP實驗室發表的一篇關于RDD的論文:《Resilient Distributed Datasets: A Fault-Tolerant  Abstraction for In-Memory Cluster Computing》就是為了解決這些問題的。

RDD提供了一個抽象的數據模型，讓我們不必擔心底層數據的分布式特性，只需將具體的應用邏輯表達為一系列轉換操作(函數)，不同RDD之間的轉換操作之間還可以形成依賴關系，進而實現管道化，從而避免了中間結果的存儲，大大降低了數據復制、磁盤IO和序列化開銷，并且還提供了更多的API(map/reduec/filter/groupBy...)。

2. RDD是什么?

RDD(Resilient Distributed  Dataset)叫做彈性分布式數據集，是Spark中最基本的數據抽象，代表一個不可變、可分區、里面的元素可并行計算的集合。單詞拆解：

Resilient ：它是彈性的，RDD里面的中的數據可以保存在內存中或者磁盤里面

Distributed ：它里面的元素是分布式存儲的，可以用于分布式計算

Dataset: 它是一個集合，可以存放很多元素

3. RDD主要屬性

進入RDD的源碼中看下：

RDD源碼

在源碼中可以看到有對RDD介紹的注釋，我們來翻譯下：

A list of partitions  ：一組分片(Partition)/一個分區(Partition)列表，即數據集的基本組成單位。對于RDD來說，每個分片都會被一個計算任務處理，分片數決定并行度。用戶可以在創建RDD時指定RDD的分片個數，如果沒有指定，那么就會采用默認值。

A function for computing each split  ：一個函數會被作用在每一個分區。Spark中RDD的計算是以分片為單位的，compute函數會被作用到每個分區上。

A list of dependencies on other RDDs  ：一個RDD會依賴于其他多個RDD。RDD的每次轉換都會生成一個新的RDD，所以RDD之間就會形成類似于流水線一樣的前后依賴關系。在部分分區數據丟失時，Spark可以通過這個依賴關系重新計算丟失的分區數據，而不是對RDD的所有分區進行重新計算。(Spark的容錯機制)

Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is  hash-partitioned)：可選項，對于KV類型的RDD會有一個Partitioner，即RDD的分區函數，默認為HashPartitioner。

Optionally, a list of preferred locations to compute each split on (e.g.  block locations for an HDFS file)：可選項,一個列表，存儲存取每個Partition的優先位置(preferred  location)。對于一個HDFS文件來說，這個列表保存的就是每個Partition所在的塊的位置。按照"移動數據不如移動計算"的理念，Spark在進行任務調度的時候，會盡可能選擇那些存有數據的worker節點來進行任務計算。

總結

RDD 是一個數據集的表示，不僅表示了數據集，還表示了這個數據集從哪來，如何計算，主要屬性包括：

1. 分區列表

2. 計算函數

3. 依賴關系

4. 分區函數(默認是hash)

5. 最佳位置

分區列表、分區函數、最佳位置，這三個屬性其實說的就是數據集在哪，在哪計算更合適，如何分區;

計算函數、依賴關系，這兩個屬性其實說的是數據集怎么來的。

二、RDD-API

1. RDD的創建方式

由外部存儲系統的數據集創建，包括本地的文件系統，還有所有Hadoop支持的數據集，比如HDFS、Cassandra、HBase等：

val rdd1 = sc.textFile("hdfs://node1:8020/wordcount/input/words.txt")

通過已有的RDD經過算子轉換生成新的RDD：

val rdd2=rdd1.flatMap(_.split(" "))

由一個已經存在的Scala集合創建：

val rdd3 = sc.parallelize(Array(1,2,3,4,5,6,7,8))或 val rdd4 = sc.makeRDD(List(1,2,3,4,5,6,7,8))

makeRDD方法底層調用了parallelize方法：

RDD源碼

2. RDD的算子分類

RDD的算子分為兩類:

Transformation轉換操作:返回一個新的RDD

Action動作操作:返回值不是RDD(無返回值或返回其他的)

注意:

1、RDD不實際存儲真正要計算的數據，而是記錄了數據的位置在哪里，數據的轉換關系(調用了什么方法，傳入什么函數)。

2、RDD中的所有轉換都是惰性求值/延遲執行的，也就是說并不會直接計算。只有當發生一個要求返回結果給Driver的Action動作時，這些轉換才會真正運行。

3、之所以使用惰性求值/延遲執行，是因為這樣可以在Action時對RDD操作形成DAG有向無環圖進行Stage的劃分和并行優化，這種設計讓Spark更加有效率地運行。

3. Transformation轉換算子

4. Action動作算子

統計操作：

三、RDD的持久化/緩存

在實際開發中某些RDD的計算或轉換可能會比較耗費時間，如果這些RDD后續還會頻繁的被使用到，那么可以將這些RDD進行持久化/緩存，這樣下次再使用到的時候就不用再重新計算了，提高了程序運行的效率。

val rdd1 = sc.textFile("hdfs://node01:8020/words.txt") val rdd2 = rdd1.flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_) rdd2.cache //緩存/持久化 rdd2.sortBy(_._2,false).collect//觸發action,會去讀取HDFS的文件,rdd2會真正執行持久化 rdd2.sortBy(_._2,false).collect//觸發action,會去讀緩存中的數據,執行速度會比之前快,因為rdd2已經持久化到內存中了

持久化/緩存API詳解

ersist方法和cache方法

RDD通過persist或cache方法可以將前面的計算結果緩存，但是并不是這兩個方法被調用時立即緩存，而是觸發后面的action時，該RDD將會被緩存在計算節點的內存中，并供后面重用。

通過查看RDD的源碼發現cache最終也是調用了persist無參方法(默認存儲只存在內存中)：

RDD源碼

存儲級別

默認的存儲級別都是僅在內存存儲一份，Spark的存儲級別還有好多種，存儲級別在object StorageLevel中定義的。

總結：

RDD持久化/緩存的目的是為了提高后續操作的速度

緩存的級別有很多，默認只存在內存中,開發中使用memory_and_disk

只有執行action操作的時候才會真正將RDD數據進行持久化/緩存

實際開發中如果某一個RDD后續會被頻繁的使用，可以將該RDD進行持久化/緩存

四、RDD容錯機制Checkpoint

持久化的局限：

持久化/緩存可以把數據放在內存中，雖然是快速的，但是也是最不可靠的;也可以把數據放在磁盤上，也不是完全可靠的!例如磁盤會損壞等。

問題解決：

Checkpoint的產生就是為了更加可靠的數據持久化，在Checkpoint的時候一般把數據放在在HDFS上，這就天然的借助了HDFS天生的高容錯、高可靠來實現數據最大程度上的安全，實現了RDD的容錯和高可用。

用法：

SparkContext.setCheckpointDir("目錄") //HDFS的目錄  RDD.checkpoint

總結：

開發中如何保證數據的安全性性及讀取效率：可以對頻繁使用且重要的數據，先做緩存/持久化，再做checkpint操作。

持久化和Checkpoint的區別：

1. 位置：Persist 和 Cache 只能保存在本地的磁盤和內存中(或者堆外內存--實驗中) Checkpoint 可以保存數據到 HDFS  這類可靠的存儲上。

2. 生命周期：Cache和Persist的RDD會在程序結束后會被清除或者手動調用unpersist方法  Checkpoint的RDD在程序結束后依然存在，不會被刪除。

五、RDD依賴關系

1. 寬窄依賴

兩種依賴關系類型：RDD和它依賴的父RDD的關系有兩種不同的類型，即寬依賴(wide dependency/shuffle  dependency)窄依賴(narrow dependency)

圖解：

寬窄依賴

如何區分寬窄依賴：

窄依賴:父RDD的一個分區只會被子RDD的一個分區依賴;

寬依賴:父RDD的一個分區會被子RDD的多個分區依賴(涉及到shuffle)。

2. 為什么要設計寬窄依賴

對于窄依賴：

窄依賴的多個分區可以并行計算;

窄依賴的一個分區的數據如果丟失只需要重新計算對應的分區的數據就可以了。

對于寬依賴：

劃分Stage(階段)的依據:對于寬依賴,必須等到上一階段計算完成才能計算下一階段。

六、DAG的生成和劃分Stage

1. DAG介紹

DAG是什么：

DAG(Directed Acyclic Graph有向無環圖)指的是數據轉換執行的過程，有方向，無閉環(其實就是RDD執行的流程);

原始的RDD通過一系列的轉換操作就形成了DAG有向無環圖，任務執行時，可以按照DAG的描述，執行真正的計算(數據被操作的一個過程)。

DAG的邊界

開始:通過SparkContext創建的RDD;

結束:觸發Action，一旦觸發Action就形成了一個完整的DAG。

2.DAG劃分Stage

DAG劃分Stage

一個Spark程序可以有多個DAG(有幾個Action，就有幾個DAG，上圖最后只有一個Action(圖中未表現),那么就是一個DAG)。

一個DAG可以有多個Stage(根據寬依賴/shuffle進行劃分)。

同一個Stage可以有多個Task并行執行(task數=分區數，如上圖，Stage1 中有三個分區P1、P2、P3，對應的也有三個 Task)。

可以看到這個DAG中只reduceByKey操作是一個寬依賴，Spark內核會以此為邊界將其前后劃分成不同的Stage。

同時我們可以注意到，在圖中Stage1中，從textFile到flatMap到map都是窄依賴，這幾步操作可以形成一個流水線操作，通過flatMap操作生成的partition可以不用等待整個RDD計算結束，而是繼續進行map操作，這樣大大提高了計算的效率。

• 為什么要劃分Stage? --并行計算

一個復雜的業務邏輯如果有shuffle，那么就意味著前面階段產生結果后，才能執行下一個階段，即下一個階段的計算要依賴上一個階段的數據。那么我們按照shuffle進行劃分(也就是按照寬依賴就行劃分)，就可以將一個DAG劃分成多個Stage/階段，在同一個Stage中，會有多個算子操作，可以形成一個pipeline流水線，流水線內的多個平行的分區可以并行執行。

• 如何劃分DAG的stage?

對于窄依賴，partition的轉換處理在stage中完成計算，不劃分(將窄依賴盡量放在在同一個stage中，可以實現流水線計算)。

對于寬依賴，由于有shuffle的存在，只能在父RDD處理完成后，才能開始接下來的計算，也就是說需要要劃分stage。

看完上述內容，你們掌握怎么實現Spark Core的原理分析的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！