大家好，我是Tom哥。

互聯(lián)網(wǎng)時代，隨著業(yè)務數(shù)據(jù)化，數(shù)據(jù)越來越多。如何用好數(shù)據(jù)，做好數(shù)據(jù)業(yè)務化，我們需要有個利器。

很多人都用過Hadoop，包含兩部分 HDFS 和 MapReduce，其中 MapReduce 是Hadoop的分布式計算引擎，計算過程中需要頻繁落盤，性能會弱一些。

今天，帶大家 快速熟悉一個大數(shù)據(jù)框架，Spark。

Spark 是內(nèi)存計算引擎，性能更好一些。盛行自 2014年，支持 流計算 Streaming、數(shù)據(jù)分析 SQL、機器學習 MLlib、圖計算 GraphFrames 等多種場景。

語言支持很多，如 Python、Java、Scala、R 和 SQL。提供了種類豐富的開發(fā)算子，如 RDD、DataFrame、Dataset。

有了這些基礎工具，開發(fā)者就可以像搭樂高一樣，快速完成各種業(yè)務場景系統(tǒng)開發(fā)。

一、先來個體感

首先，我們看一個簡單的代碼示例，讓大家有個體感。

import org.apache.spark.rdd.RDD
val file: String = "/Users/onlyone/spark/demo.txt"
// 加載文件
val lineRDD: RDD[String] = spark.sparkContext.textFile(file)
val wordRDD: RDD[String] = lineRDD.flatMap(line => line.split(" "))
val kvRDD: RDD[(String, Int)] = wordRDD.map(word => (word, 1))
val wordCounts: RDD[(String, Int)] = kvRDD.reduceByKey((x, y) => x + y)
wordCounts.foreach(wordCount => println(wordCount._1 + " appeared " + wordCount._2 + " times"))

我們看到，入口代碼是從第四行的 spark 變量開始。

在 spark-shell 中 由系統(tǒng)自動創(chuàng)建，是 SparkSession 的實例化對象，可以直接使用，不需要每次自己 new 一個新對象。

SparkSession 是 Spark 程序的統(tǒng)一開發(fā)入口。開發(fā)一個 Spark 應用，必須先創(chuàng)建 SparkSession。

二、RDD

彈性分布式數(shù)據(jù)集，全稱 Resilient Distributed Datasets，是一種抽象，囊括所有內(nèi)存和磁盤中的分布式數(shù)據(jù)實體，是Spark最核心的模塊和類。

RDD 中承載數(shù)據(jù)的基本單元是數(shù)據(jù)分片。在分布式計算環(huán)境中，一份完整的數(shù)據(jù)集，會按照某種規(guī)則切割成多份數(shù)據(jù)分片。這些數(shù)據(jù)分片被均勻地分發(fā)給集群內(nèi)不同的計算節(jié)點和執(zhí)行進程，從而實現(xiàn)分布式并行計算。

RDD 包含 4大屬性：

• 數(shù)據(jù)分片，partitions。
• 分片切割規(guī)則， partitioner。
• RDD 依賴關系， dependencies。
• 轉換函數(shù)，compute。

RDD 表示的是分布式數(shù)據(jù)形態(tài)，RDD 到 RDD 之間的轉換，本質(zhì)上是數(shù)據(jù)形態(tài)上的轉換，這里面的一個重要角色就是算子。

三、算子

算子分為兩大類，Transformations 和 Actions。

• Transformations 算子：通過函數(shù)方法對數(shù)據(jù)從一種形態(tài)轉換為另一種形態(tài)。
• Actions 算子：收集計算結果，或者將數(shù)據(jù)物化到磁盤。

劃重點：mapPartitions 與 map 的功能類似，但是mapPartitions 算子是以數(shù)據(jù)分區(qū)為粒度初始化共享對象，比如：數(shù)據(jù)庫連接對象，S3文件句柄等。

結合上面的兩類算子，Spark 運行劃分為兩個環(huán)節(jié)：

• 不同數(shù)據(jù)形態(tài)之間的轉換，構建計算流圖 (DAG)。
• 通過 Actions 類算子，以回溯的方式去觸發(fā)執(zhí)行這個計算流圖。

題外話，回溯在Java 中也有引入，比如 Stream 流也是類似機制。

一個流程可能會引入很多算子，但是他們并不會立即執(zhí)行，只有當開發(fā)者調(diào)用了 Actions 算子，之前調(diào)用的轉換算子才會執(zhí)行。這個也稱為 延遲計算。

延遲計算是 Spark 分布式運行機制的一大亮點。可以讓執(zhí)行引擎從全局角度來優(yōu)化執(zhí)行流程。

四、分布式計算

Spark 應用中，程序的入口是帶有 SparkSession 的 main 函數(shù)。

SparkSession 提供了 Spark 運行時的上下文，如 調(diào)度系統(tǒng)、存儲系統(tǒng)、內(nèi)存管理、RPC 通信)，同時為開發(fā)者提供創(chuàng)建、轉換、計算分布式數(shù)據(jù)集的開發(fā) API。

運行這個 SparkSession 的main函數(shù)的JVM進程，我們稱為 Driver。

Driver 職責：

解析用戶代碼，構建 DAG 圖，然后將計算流圖轉化為分布式任務，將任務分發(fā)給集群的 Executor 執(zhí)行。定期與每個 Executor 通信，及時獲取任務的進展，從而協(xié)調(diào)整體的執(zhí)行進度。

Executors 職責：

調(diào)用內(nèi)部線程池，結合事先分配好的數(shù)據(jù)分片，并發(fā)地執(zhí)行任務代碼。每個 Executors 負責處理 RDD 的一個數(shù)據(jù)分片子集。

分布式計算的核心是任務調(diào)度，主要是 Driver 與 Executors 之間的交互。

Driver 的任務調(diào)度依賴于 DAGScheduler、TaskScheduler 和 SchedulerBackend。

計算過程：

Driver 通過 foreach 這個 Action 算子，觸發(fā)計算流圖的執(zhí)行，上圖自左向右執(zhí)行，以 shuffle 為邊界，創(chuàng)建、分發(fā)分布式任務。

其中的 textFile、flatMap、map 三個算子合并成一份任務，分發(fā)給每一個 Executor。Executor 收到任務后，對任務進行解析，把任務拆解成 textFile、flatMap、map 3 個步驟，然后分別對自己負責的數(shù)據(jù)分片進行處理。

每個 Executor 執(zhí)行完得到中間結果，然后向 Driver 匯報任務進度。接著 Driver 進行后續(xù)的聚合計算，由于數(shù)據(jù)分散在多個分片，會觸發(fā) shuffle 操作。

shuffle 機制是將原來多個 Executor中的計算結果重新路由、分發(fā)到同一個 Executor，然后對匯總后的數(shù)據(jù)再次處理。在集群范圍內(nèi)跨進程、跨節(jié)點的數(shù)據(jù)交換?？赡艽嬖诰W(wǎng)絡性能瓶頸，需要特別關注。

在不同 Executor 完成數(shù)據(jù)交換之后，Driver 分發(fā)下一個階段的任務，對單詞計數(shù)。

同一個key的數(shù)據(jù)已經(jīng)分發(fā)到相同的 Executor ，每個 Executor 獨自完成計數(shù)統(tǒng)計。

最后，Executors 把最終的計算結果統(tǒng)一返回給 Driver。

劃重點：DAG 到 Stages 的拆分過程，以 Actions 算子為觸發(fā)起點，從后往前回溯 DAG，以 Shuffle 為邊界劃分 Stages。

收集結果：

收集結果，按照收集的路徑不同，主要分為兩類：

1. 把計算結果從各個 Executors 收集到 Driver 端。
2. 把計算結果通過 Executors 直接持久化到文件系統(tǒng)。如：HDFS 或 S3 分布式文件系統(tǒng)。

五、調(diào)度系統(tǒng)

1、DAGScheduler

根據(jù)用戶代碼構建 DAG，以 Shuffle 為邊界切割 Stages。每個Stage 根據(jù) RDD中的Partition分區(qū)個數(shù)決定Task的個數(shù)，然后構建 TaskSets，然后將 TaskSets 提交給 TaskScheduler 請求調(diào)度。

2、TaskScheduler

按照任務的本地傾向性，挑選出 TaskSet 中適合調(diào)度的 Task，然后將 Task 分配到 Executor 上執(zhí)行。

3、SchedulerBackend

通過ExecutorDataMap 數(shù)據(jù)結構，來記錄每一個計算節(jié)點中 Executors 的資源狀態(tài)，如 RPC 地址、主機地址、可用 CPU 核數(shù)和滿配 CPU 核數(shù)等。

4、Task

運行在Executor上的工作單元。

5、Job

SparkContext提交的具體Action操作，常和Action對應。

6、Stage

每個Job會被拆分很多組任務(task)，每組任務被稱為Stage，也稱 TaskSet。

調(diào)度系統(tǒng)的核心思想：數(shù)據(jù)不動、代碼動。

六、內(nèi)存管理

Spark 的內(nèi)存分為 4 個區(qū)域，Reserved Memory、User Memory、Execution Memory 和 Storage Memory。

1. Reserved Memory：固定為 300MB，Spark 預留的，用來存儲各種 Spark 內(nèi)部對象的內(nèi)存區(qū)域。
2. User Memory：存儲開發(fā)者自定義的數(shù)據(jù)結構，例如 RDD 算子中引用的數(shù)組、列表、映射。
3. Execution Memory：執(zhí)行分布式任務。分布式任務的計算，主要包括數(shù)據(jù)的轉換、過濾、映射、排序、聚合、歸并等。
4. Storage Memory：緩存分布式數(shù)據(jù)集，如 RDD Cache、廣播變量等。

整個內(nèi)存區(qū)域，Execution Memory 和 Storage Memory 最重要。在 1.6 版本之后，Spark 推出了統(tǒng)一內(nèi)存管理模式，這兩者可以相互轉化。

七、共享變量

Spark 提供兩類共享變量，分別是廣播變量(Broadcast variables)和累加器(Accumulators)。

1、廣播變量

val list: List[String] = List("Tom哥", "Spark")
// sc為SparkContext實例
val bc = sc.broadcast(list)

廣播變量的用法很簡單，通過調(diào)用 SparkContext 下的 broadcast 即可完成廣播變量的創(chuàng)建。

如果要讀取封裝的共享數(shù)據(jù)內(nèi)容，調(diào)用它的 bc.value 函數(shù)。

好奇寶寶會問，既然 list 可以獲取字符串列表，為什么還要封裝廣播變量呢?

答案：

Driver 端對普通的共享變量的分發(fā)是以 Task 為粒度的，系統(tǒng)中有多少個 Task，變量就需要在網(wǎng)絡中分發(fā)多少次，存在巨大的內(nèi)存資源浪費。

使用廣播變量后，共享變量分發(fā)的粒度以 Executors 為單位，同一個 Executor 內(nèi)多個不同的 Tasks 只需訪問同一份數(shù)據(jù)拷貝即可。也就是說，變量在網(wǎng)絡中分發(fā)與存儲的次數(shù)，從 RDD 的分區(qū)數(shù)，減少為集群中 Executors 的個數(shù)。

2、累加器

累加器也是在 Driver 端定義，累計過程是通過在 RDD 算子中調(diào)用 add 函數(shù)為累加器計數(shù)，從而更新累加器狀態(tài)。

應用執(zhí)行完畢之后，開發(fā)者在 Driver 端調(diào)用累加器的 value 函數(shù)，獲取全局計數(shù)結果。

Spark 提供了 3 種累加器，longAccumulator、doubleAccumulator 和 collectionAccumulator ，滿足不同的業(yè)務場景。