怎么用實例解析Spark Core

今天就跟大家聊聊有關怎么用實例解析Spark Core，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

先來一個問題，也是面試中常問的：

Spark為什么會流行？

原因1：優(yōu)秀的數(shù)據模型和豐富計算抽象

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

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

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

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

原因2：完善的生態(tài)圈-fullstack

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

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

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

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

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

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

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

StructuredStreaming:處理結構化流,統(tǒng)一了離線和實時的API。

Spark VS Hadoop

????注意：
盡管Spark相對于Hadoop而言具有較大優(yōu)勢，但Spark并不能完全替代Hadoop，Spark主要用于替代Hadoop中的MapReduce計算模型。存儲依然可以使用HDFS，但是中間結果可以存放在內存中；調度可以使用Spark內置的，也可以使用更成熟的調度系統(tǒng)YARN等。
實際上，Spark已經很好地融入了Hadoop生態(tài)圈，并成為其中的重要一員，它可以借助于YARN實現(xiàn)資源調度管理，借助于HDFS實現(xiàn)分布式存儲。
此外，Hadoop可以使用廉價的、異構的機器來做分布式存儲與計算，但是，Spark對硬件的要求稍高一些，對內存與CPU有一定的要求。

Spark Core

一、RDD詳解

1. 為什么要有RDD?

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

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

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

2. RDD是什么?

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

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

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

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

3. RDD主要屬性

進入RDD的源碼中看下：

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

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

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

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

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

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

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

1. 分區(qū)列表

2. 計算函數(shù)

3. 依賴關系

4. 分區(qū)函數(shù)(默認是hash)

5. 最佳位置

分區(qū)列表、分區(qū)函數(shù)、最佳位置，這三個屬性其實說的就是數(shù)據集在哪，在哪計算更合適，如何分區(qū)；
計算函數(shù)、依賴關系，這兩個屬性其實說的是數(shù)據集怎么來的。

二、RDD-API

1. RDD的創(chuàng)建方式

1. 由外部存儲系統(tǒng)的數(shù)據集創(chuàng)建，包括本地的文件系統(tǒng)，還有所有Hadoop支持的數(shù)據集，比如HDFS、Cassandra、HBase等：
   val rdd1 = sc.textFile("hdfs://node1:8020/wordcount/input/words.txt")

2. 通過已有的RDD經過算子轉換生成新的RDD：
   val rdd2=rdd1.flatMap(_.split(" "))

3. 由一個已經存在的Scala集合創(chuàng)建：
   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方法：

2. RDD的算子分類

RDD的算子分為兩類:

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

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

??注意:
1、RDD不實際存儲真正要計算的數(shù)據，而是記錄了數(shù)據的位置在哪里，數(shù)據的轉換關系(調用了什么方法，傳入什么函數(shù))。
2、RDD中的所有轉換都是惰性求值/延遲執(zhí)行的，也就是說并不會直接計算。只有當發(fā)生一個要求返回結果給Driver的Action動作時，這些轉換才會真正運行。
3、之所以使用惰性求值/延遲執(zhí)行，是因為這樣可以在Action時對RDD操作形成DAG有向無環(huán)圖進行Stage的劃分和并行優(yōu)化，這種設計讓Spark更加有效率地運行。

3. Transformation轉換算子

4. Action動作算子

統(tǒng)計操作：

三、RDD的持久化/緩存

在實際開發(fā)中某些RDD的計算或轉換可能會比較耗費時間，如果這些RDD后續(xù)還會頻繁的被使用到，那么可以將這些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//觸發(fā)action,會去讀取HDFS的文件,rdd2會真正執(zhí)行持久化
rdd2.sortBy(_._2,false).collect//觸發(fā)action,會去讀緩存中的數(shù)據,執(zhí)行速度會比之前快,因為rdd2已經持久化到內存中了

持久化/緩存API詳解

• ersist方法和cache方法

RDD通過persist或cache方法可以將前面的計算結果緩存，但是并不是這兩個方法被調用時立即緩存，而是觸發(fā)后面的action時，該RDD將會被緩存在計算節(jié)點的內存中，并供后面重用。
通過查看RDD的源碼發(fā)現(xiàn)cache最終也是調用了persist無參方法(默認存儲只存在內存中)：

• 存儲級別

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

總結：

1. RDD持久化/緩存的目的是為了提高后續(xù)操作的速度

2. 緩存的級別有很多，默認只存在內存中,開發(fā)中使用memory_and_disk

3. 只有執(zhí)行action操作的時候才會真正將RDD數(shù)據進行持久化/緩存

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

四、RDD容錯機制Checkpoint

• 持久化的局限：

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

• 問題解決：

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

用法：

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

RDD.checkpoint

總結：

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

• 持久化和Checkpoint的區(qū)別：

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

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

五、RDD依賴關系

1. 寬窄依賴

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

• 圖解：

• 如何區(qū)分寬窄依賴：

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

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

1. 對于窄依賴：

窄依賴的多個分區(qū)可以并行計算；
窄依賴的一個分區(qū)的數(shù)據如果丟失只需要重新計算對應的分區(qū)的數(shù)據就可以了。

2. 對于寬依賴：

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

六、DAG的生成和劃分Stage

1. DAG介紹

• DAG是什么：

DAG(Directed Acyclic Graph有向無環(huán)圖)指的是數(shù)據轉換執(zhí)行的過程，有方向，無閉環(huán)(其實就是RDD執(zhí)行的流程)；
原始的RDD通過一系列的轉換操作就形成了DAG有向無環(huán)圖，任務執(zhí)行時，可以按照DAG的描述，執(zhí)行真正的計算(數(shù)據被操作的一個過程)。

• DAG的邊界

開始:通過SparkContext創(chuàng)建的RDD；
結束:觸發(fā)Action，一旦觸發(fā)Action就形成了一個完整的DAG。

2.DAG劃分Stage

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

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

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

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

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

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

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

• 如何劃分DAG的stage？

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

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

Spark會根據shuffle/寬依賴使用回溯算法來對DAG進行Stage劃分，從后往前，遇到寬依賴就斷開，遇到窄依賴就把當前的RDD加入到當前的stage/階段中。

看完上述內容，你們對怎么用實例解析Spark Core有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業(yè)資訊頻道，感謝大家的支持。