1. Spark介紹

Apache Spark是一個圍繞速度、易用性和復(fù)雜分析構(gòu)建的大數(shù)據(jù)處理框架，最初在2009年由加州大學(xué)伯克利分校的AMPLab開發(fā)，并于2010年成為Apache的開源項目之一，與Hadoop和Storm等其他大數(shù)據(jù)和MapReduce技術(shù)相比，Spark有如下優(yōu)勢：

1) 運行速度快, Spark擁有DAG執(zhí)行引擎，支持在內(nèi)存中對數(shù)據(jù)進行迭代計算。官方提供的數(shù)據(jù)表明，如果數(shù)據(jù)由磁盤讀取，速度是Hadoop MapReduce的10倍以上，如果數(shù)據(jù)從內(nèi)存中讀取，速度可以高達100多倍。

2) 適用場景廣泛, 大數(shù)據(jù)分析統(tǒng)計，實時數(shù)據(jù)處理，圖計算及機器學(xué)習(xí)

3) 易用性, 編寫簡單, 支持80種以上的高級算子，支持多種語言，數(shù)據(jù)源豐富，可部署在多種集群中

4) 容錯性高。Spark引進了彈性分布式數(shù)據(jù)集RDD (Resilient Distributed Dataset) 的抽象，它是分布在一組節(jié)點中的只讀對象集合，這些集合是彈性的，如果數(shù)據(jù)集一部分丟失，則可以根據(jù)“血統(tǒng)”（即充許基于數(shù)據(jù)衍生過程）對它們進行重建。另外在RDD計算時可以通過CheckPoint來實現(xiàn)容錯，而CheckPoint有兩種方式：CheckPoint Data，和Logging The Updates，用戶可以控制采用哪種方式來實現(xiàn)容錯.

Spark的適用場景有以下幾個類型：

1) 復(fù)雜的批量處理（Batch Data Processing），偏重點在于處理海量數(shù)據(jù)的能力，至于處理速度可忍受，通常的時間可能是在數(shù)十分鐘到數(shù)小時；

2) 基于歷史數(shù)據(jù)的交互式查詢（Interactive Query），通常的時間在數(shù)十秒到數(shù)十分鐘之間

3) 基于實時數(shù)據(jù)流的數(shù)據(jù)處理（Streaming Data Processing），通常在數(shù)百毫秒到數(shù)秒之間

2. Spark架構(gòu)以及生態(tài)

通常當需要處理的數(shù)據(jù)量超過了單機尺度(比如我們的計算機有4GB的內(nèi)存，而我們需要處理100GB以上的數(shù)據(jù))這時我們可以選擇spark集群進行計算，有時我們可能需要處理的數(shù)據(jù)量并不大，但是計算很復(fù)雜，需要大量的時間，這時我們也可以選擇利用spark集群強大的計算資源，并行化地計算，其架構(gòu)示意圖如下：

Spark Core：包含Spark的基本功能；尤其是定義RDD的API、操作以及這兩者上的動作。其他Spark的庫都是構(gòu)建在RDD和Spark Core之上的。

Spark SQL：提供通過Apache Hive的SQL變體Hive查詢語言（HiveQL）與Spark進行交互的API。每個數(shù)據(jù)庫表被當做一個RDD，Spark SQL查詢被轉(zhuǎn)換為Spark操作。Spark提供的sql形式的對接Hive、JDBC、HBase等各種數(shù)據(jù)渠道的API，用Java開發(fā)人員的思想來講就是面向接口、解耦合，ORMapping、Spring Cloud Stream等都是類似的思想。

Spark Streaming：基于SparkCore實現(xiàn)的可擴展、高吞吐、高可靠性的實時數(shù)據(jù)流處理。支持從Kafka、Flume等數(shù)據(jù)源處理后存儲到HDFS、DataBase、Dashboard中。對實時數(shù)據(jù)流進行處理和控制。Spark Streaming允許程序能夠像普通RDD一樣處理實時數(shù)據(jù)。

MLlib：一個常用機器學(xué)習(xí)算法庫，算法被實現(xiàn)為對RDD的Spark操作。這個庫包含可擴展的學(xué)習(xí)算法，比如分類、回歸等需要對大量數(shù)據(jù)集進行迭代的操作。

GraphX：控制圖、并行圖操作和計算的一組算法和工具的集合。GraphX擴展了RDD API，包含控制圖、創(chuàng)建子圖、訪問路徑上所有頂點的操作

3. Spark的架構(gòu)設(shè)計

(1) 運行流程及特點

Hadoop存在缺陷: 基于磁盤，無論是MapReduce還是YARN都是將數(shù)據(jù)從磁盤中加載出來，經(jīng)過DAG，然后重新寫回到磁盤中，計算過程的中間數(shù)據(jù)又需要寫入到HDFS的臨時文件，這些都使得Hadoop在大數(shù)據(jù)運算上表現(xiàn)太“慢”，Spark應(yīng)運而生。

Cluster Manager在standalone模式中即為Master主節(jié)點，控制整個集群，監(jiān)控worker。在YARN模式中為資源管理器負責(zé)分配資源，有點像YARN中ResourceManager那個角色，大管家握有所有的干活的資源，屬于乙方的總包。

WorkerNode是可以干活的節(jié)點，聽大管家ClusterManager差遣，是真正有資源干活的主。從節(jié)點，負責(zé)控制計算節(jié)點，啟動Executor或者Driver。

Executor是在WorkerNode上起的一個進程，相當于一個包工頭，負責(zé)準備Task環(huán)境和執(zhí)行

Task，負責(zé)內(nèi)存和磁盤的使用。Task是施工項目里的每一個具體的任務(wù)。

Driver是統(tǒng)管Task的產(chǎn)生與發(fā)送給Executor的，運行Application 的main()函數(shù)，是甲方的司令員。

SparkContext是與ClusterManager打交道的，負責(zé)給錢申請資源的，是甲方的接口人。

整個互動流程是這樣的：

1) Spark Apllication 創(chuàng)建SparkContext , 通過SparkContext向Cluster manager(資源管理器)申請所需執(zhí)行的資源（cpu、內(nèi)存等）

2) Cluster manager分配應(yīng)用程序執(zhí)行需要的資源，在Worker(Yarn是NodeManager)節(jié)點上創(chuàng)建Executor

3) SparkContext 將程序代碼（jar包或者python文件）和Task任務(wù)發(fā)送給Executor執(zhí)行，并收集結(jié)果給Driver。

4) 釋放SparkContext

運行流程及特點

涉及的幾個定義和詳細的運行過程如下：

1) Application：Spark應(yīng)用程序

指的是用戶編寫的Spark應(yīng)用程序，包含了Driver功能代碼和分布在集群中多個節(jié)點上運行的Executor代碼。

Spark應(yīng)用程序，由一個或多個作業(yè)JOB組成，如下圖所示。

2) Driver：驅(qū)動程序

Spark中的Driver即運行上述Application的Main()函數(shù)并且創(chuàng)建SparkContext，其中創(chuàng)建SparkContext的目的是為了準備Spark應(yīng)用程序的運行環(huán)境。在Spark中由SparkContext負責(zé)和ClusterManager通信，進行資源的申請、任務(wù)的分配和監(jiān)控等；當Executor部分運行完畢后，Driver負責(zé)將SparkContext關(guān)閉。通常SparkContext代表Driver，如下圖所示。

3) Cluster Manager：資源管理器

指的是在集群上獲取資源的外部服務(wù)，常用的有：Standalone，Spark原生的資源管理器，由Master負責(zé)資源的分配；Hadoop Yarn，由Yarn中的ResourceManager負責(zé)資源的分配；Messos，由Messos中的Messos Master負責(zé)資源管理。

4) Executor：執(zhí)行器

Executor是Application運行在Worker節(jié)點上的一個進程，該進程負責(zé)運行Task，并且負責(zé)將數(shù)據(jù)存在內(nèi)存或者磁盤上，每個Application都有各自獨立的一批Executor，如下圖所示。

5）Worker：計算節(jié)點

集群中任何可以運行Application代碼的節(jié)點，類似于Yarn中的NodeManager節(jié)點。在Standalone模式中指的就是通過Slave文件配置的Worker節(jié)點，在Spark on Yarn模式中指的就是NodeManager節(jié)點，在Spark on Messos模式中指的就是Messos Slave節(jié)點，如下圖所示。

6) DAGScheduler：有向無環(huán)圖調(diào)度器

在Spark里每一個操作生成一個RDD，RDD之間連一條邊，最后這些RDD和他們之間的邊組成一個有向無環(huán)圖，這個就是DAG。

Spark會根據(jù)寬依賴窄依賴來劃分具體的Stage，而依賴有2個作用：

用來解決數(shù)據(jù)容錯的高效性和用來劃分stage。

RDD的依賴關(guān)系分為兩種：窄依賴(Narrow Dependencies)與寬依賴(Wide Dependencies)

a. 窄依賴

每個父RDD的一個Partition最多被子RDD的一個Partition所使用（1:1 或 n:1）。例如map、filter、union等操作都會產(chǎn)生窄依賴；

子RDD分區(qū)通常對應(yīng)常數(shù)個父RDD分區(qū)(O(1)，與數(shù)據(jù)規(guī)模無關(guān)）。

b. 寬依賴

一個父RDD的Partition會被多個子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都會產(chǎn)生寬依賴；（1:m 或 n:m）

子RDD分區(qū)通常對應(yīng)所有的父RDD分區(qū)(O(n)，與數(shù)據(jù)規(guī)模有關(guān)）

原始的RDD通過一系列的轉(zhuǎn)換就形成了DAG，有了計算的DAG圖，Spark內(nèi)核下一步的任務(wù)就是根據(jù)DAG圖將計算劃分成任務(wù)集，也就是Stage，這樣可以將任務(wù)提交到計算節(jié)點進行真正的計算。Spark計算的中間結(jié)果默認是保存在內(nèi)存中的，Spark在劃分Stage的時候會充分考慮在分布式計算中可流水線計算（pipeline）的部分來提高計算的效率，而在這個過程中Spark根據(jù)RDD之間依賴關(guān)系的不同將DAG劃分成不同的Stage(調(diào)度階段)。對于窄依賴，partition的轉(zhuǎn)換處理在一個Stage中完成計算。對于寬依賴，由于有Shuffle的存在，只能在parent RDD處理完成后，才能開始接下來的計算，因此寬依賴是劃分Stage的依據(jù)。

Spark 執(zhí)行時有下面所列的流程：

a. 用戶代碼定義RDD的有向無環(huán)圖

RDD上的操作會創(chuàng)建新的RDD，并引用它們的父節(jié)點，這樣就創(chuàng)建了一個圖。

b. 行動(action)操作把有向無環(huán)圖強制轉(zhuǎn)譯為執(zhí)行計劃stage

當調(diào)用RDD的一個行動操作時，這個RDD就必須被計算出來。這也要求計算出該RDD的父節(jié)點。Spark調(diào)度器提交一個作業(yè)來計算出所有必要的RDD。這個作業(yè)會包含一個或多個步驟，每個步驟其實也就是一波并行執(zhí)行的計算任務(wù)。一個stage步驟對應(yīng)有向五環(huán)圖中的一個或多個RDD，一個步驟對應(yīng)多個RDD是因為發(fā)生了流水線執(zhí)行。

c. 任務(wù)于集群中調(diào)度并執(zhí)行

步驟stage是按順序處理的，任務(wù)則獨立的啟動來計算出RDD的一部分。一旦作業(yè)的最后一個步驟結(jié)束，一個行動操作也就執(zhí)行完了。

DAGScheduler基于DAG劃分Stage 并以TaskSet的形勢提交Stage給TaskScheduler；負責(zé)將作業(yè)拆分成不同階段的具有依賴關(guān)系的多批任務(wù)；最重要的任務(wù)之一就是：計算作業(yè)和任務(wù)的依賴關(guān)系，制定調(diào)度邏輯。在SparkContext初始化的過程中被實例化，一個SparkContext對應(yīng)創(chuàng)建一個DAGScheduler。

7) TaskScheduler：任務(wù)調(diào)度器

將Taskset提交給worker（集群）運行并回報結(jié)果；負責(zé)每個具體任務(wù)的實際物理調(diào)度。如圖所示。

8) Job：作業(yè)

由一個或多個調(diào)度階段所組成的一次計算作業(yè)；包含多個Task組成的并行計算，往往由Spark Action催生，一個JOB包含多個RDD及作用于相應(yīng)RDD上的各種Operation。如圖所示。

9) Stage：調(diào)度階段

一個任務(wù)集對應(yīng)的調(diào)度階段；每個Job會被拆分很多組Task，每組任務(wù)被稱為Stage，也可稱TaskSet，一個作業(yè)分為多個階段；Stage分成兩種類型ShuffleMapStage、ResultStage。如圖所示。

Application多個job多個Stage：Spark Application中可以因為不同的Action觸發(fā)眾多的job，一個Application中可以有很多的job，每個job是由一個或者多個Stage構(gòu)成的，后面的Stage依賴于前面的Stage，也就是說只有前面依賴的Stage計算完畢后，后面的Stage才會運行。

劃分依據(jù)：Stage劃分的依據(jù)就是寬依賴，何時產(chǎn)生寬依賴，reduceByKey, groupByKey等算子，會導(dǎo)致寬依賴的產(chǎn)生。

核心算法：從后往前回溯，遇到窄依賴加入本stage，遇見寬依賴進行Stage切分。Spark內(nèi)核會從觸發(fā)Action操作的那個RDD開始從后往前推，首先會為最后一個RDD創(chuàng)建一個stage，然后繼續(xù)倒推，如果發(fā)現(xiàn)對某個RDD是寬依賴，那么就會將寬依賴的那個RDD創(chuàng)建一個新的stage，那個RDD就是新的stage的最后一個RDD。然后依次類推，繼續(xù)繼續(xù)倒推，根據(jù)窄依賴或者寬依賴進行stage的劃分，直到所有的RDD全部遍歷完成為止。

將DAG劃分為Stage剖析：如上圖，從HDFS中讀入數(shù)據(jù)生成3個不同的RDD，通過一系列transformation操作后再將計算結(jié)果保存回HDFS。可以看到這個DAG中只有join操作是一個寬依賴，Spark內(nèi)核會以此為邊界將其前后劃分成不同的Stage. 同時我們可以注意到，在圖中Stage2中，從map到union都是窄依賴，這兩步操作可以形成一個流水線操作，通過map操作生成的partition可以不用等待整個RDD計算結(jié)束，而是繼續(xù)進行union操作，這樣大大提高了計算的效率。

10) TaskSet：任務(wù)集

由一組關(guān)聯(lián)的，但相互之間沒有Shuffle依賴關(guān)系的任務(wù)所組成的任務(wù)集。如圖所示。

提示：

1）一個Stage創(chuàng)建一個TaskSet；

2）為Stage的每個Rdd分區(qū)創(chuàng)建一個Task,多個Task封裝成TaskSet

summary:

4. Spark RDD 操作

map的作用很容易理解就是對rdd之中的元素進行逐一進行函數(shù)操作映射為另外一個rdd。

flatMap的操作是將函數(shù)應(yīng)用于rdd之中的每一個元素，將返回的迭代器的所有內(nèi)容構(gòu)成新的rdd。通常用來切分單詞。

Spark 中 map函數(shù)會對每一條輸入進行指定的操作，然后為每一條輸入返回一個對象。 而flatMap函數(shù)則是兩個操作的集合——正是“先映射后扁平化”：

-------flat map------

object fla_map {

def main(args: Array[String]): Unit = {

val rdd1= List(List("A","B"),List("C","D"))

rdd1.map( i => println(i))

println(rdd1)

println("----------------------")

val strings = rdd1.flatMap(f => f)

println(strings)

strings.foreach( i => println(i))

}

}

-----------------

map：List里有小的List

flatmap：先flat再map，只能壓一次，形成一個新的List集合，把原元素放進新的集合里面

總結(jié)

以上是生活随笔為你收集整理的spark中local模式与cluster模式使用场景_Spark 知识点 ( 架构 RDD Task )的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。