Spark 生態(tài)系統(tǒng)組件

摘要： 隨著大數(shù)據(jù)技術(shù)的發(fā)展，實(shí)時(shí)流計(jì)算、機(jī)器學(xué)習(xí)、圖計(jì)算等領(lǐng)域成為較熱的研究方向，而Spark作為大數(shù)據(jù)處理的“利器”有著較為成熟的生態(tài)圈，能夠一站式解決類似場(chǎng)景的問題。那你知道Spark生態(tài)系統(tǒng)有哪些組件嗎？下面讓我們跟著本文一同了解下這些不可或缺的組件。本文選自《圖解Spark：核心技術(shù)與案例實(shí)戰(zhàn)》

　　Spark 生態(tài)系統(tǒng)以Spark Core 為核心，能夠讀取傳統(tǒng)文件（如文本文件）、HDFS、Amazon S3、Alluxio 和NoSQL 等數(shù)據(jù)源，利用Standalone、YARN 和Mesos 等資源調(diào)度管理，完成應(yīng)用程序分析與處理。這些應(yīng)用程序來自Spark 的不同組件，如Spark Shell 或Spark Submit 交互式批處理方式、Spark Streaming 的實(shí)時(shí)流處理應(yīng)用、Spark SQL 的即席查詢、采樣近似查詢引擎BlinkDB 的權(quán)衡查詢、MLbase/MLlib 的機(jī)器學(xué)習(xí)、GraphX 的圖處理和SparkR 的數(shù)學(xué)計(jì)算等，如下圖所示，正是這個(gè)生態(tài)系統(tǒng)實(shí)現(xiàn)了“One Stack to Rule Them All”目標(biāo)。

Spark Core

　　Spark Core 是整個(gè)BDAS 生態(tài)系統(tǒng)的核心組件，是一個(gè)分布式大數(shù)據(jù)處理框架。Spark Core提供了多種資源調(diào)度管理，通過內(nèi)存計(jì)算、有向無環(huán)圖（DAG）等機(jī)制保證分布式計(jì)算的快速，并引入了RDD 的抽象保證數(shù)據(jù)的高容錯(cuò)性，其重要特性描述如下。

• Spark Core提供了多種運(yùn)行模式，不僅可以使用自身運(yùn)行模式處理任務(wù)，如本地模式、Standalone，而且可以使用第三方資源調(diào)度框架來處理任務(wù)，如YARN、MESOS等。相比較而言，第三方資源調(diào)度框架能夠更細(xì)粒度管理資源。

• Spark Core提供了有向無環(huán)圖（DAG）的分布式并行計(jì)算框架，并提供內(nèi)存機(jī)制來支持多次迭代計(jì)算或者數(shù)據(jù)共享，大大減少迭代計(jì)算之間讀取數(shù)據(jù)的開銷，這對(duì)于需要進(jìn)行多次迭代的數(shù)據(jù)挖掘和分析性能有極大提升。另外，在任務(wù)處理過程中移動(dòng)計(jì)算而非移動(dòng)數(shù)據(jù)，RDDPartition 可以就近讀取分布式文件系統(tǒng)中的數(shù)據(jù)塊到各個(gè)節(jié)點(diǎn)內(nèi)存中進(jìn)行計(jì)算。

• 在Spark 中引入了RDD的抽象，它是分布在一組節(jié)點(diǎn)中的只讀對(duì)象集合，這些集合是彈性的，如果數(shù)據(jù)集一部分丟失，則可以根據(jù)“血統(tǒng)”對(duì)它們進(jìn)行重建，保證了數(shù)據(jù)的高容錯(cuò)性。

Spark Streaming

　　Spark Streaming 是一個(gè)對(duì)實(shí)時(shí)數(shù)據(jù)流進(jìn)行高吞吐、高容錯(cuò)的流式處理系統(tǒng)，可以對(duì)多種數(shù)據(jù)源（如Kafka、Flume、Twitter 和ZeroMQ 等）進(jìn)行類似Map、Reduce 和Join 等復(fù)雜操作，并將結(jié)果保存到外部文件系統(tǒng)、數(shù)據(jù)庫或應(yīng)用到實(shí)時(shí)儀表盤，如下圖。

　　相比其他的處理引擎要么只專注于流處理，要么只負(fù)責(zé)批處理（僅提供需要外部實(shí)現(xiàn)的流處理API 接口），而Spark Streaming 最大的優(yōu)勢(shì)是提供的處理引擎和RDD 編程模型可以同時(shí)進(jìn)行批處理與流處理。
　　對(duì)于傳統(tǒng)流處理中一次處理一條記錄的方式而言，Spark Streaming 使用的是將流數(shù)據(jù)離散化處理（Discretized Streams），通過該處理方式能夠進(jìn)行秒級(jí)以下的數(shù)據(jù)批處理。在SparkStreaming 處理過程中，Receiver 并行接收數(shù)據(jù)，并將數(shù)據(jù)緩存至Spark 工作節(jié)點(diǎn)的內(nèi)存中。經(jīng)過延遲優(yōu)化后，Spark 引擎對(duì)短任務(wù)（幾十毫秒）能夠進(jìn)行批處理，并且可將結(jié)果輸出至其他系統(tǒng)中。與傳統(tǒng)連續(xù)算子模型不同，其模型是靜態(tài)分配給一個(gè)節(jié)點(diǎn)進(jìn)行計(jì)算，而Spark 可基于數(shù)據(jù)的來源以及可用資源情況動(dòng)態(tài)分配給工作節(jié)點(diǎn)。

　　使用離散化流數(shù)據(jù)（DStreaming），Spark Streaming 將具有如下特性。

• 動(dòng)態(tài)負(fù)載均衡：Spark Streaming 將數(shù)據(jù)劃分為小批量，通過這種方式可以實(shí)現(xiàn)對(duì)資源更細(xì)粒度的分配。例如，傳統(tǒng)實(shí)時(shí)流記錄處理系統(tǒng)在輸入數(shù)據(jù)流以鍵值進(jìn)行分區(qū)處理情況下，如果一個(gè)節(jié)點(diǎn)計(jì)算壓力較大超出了負(fù)荷，該節(jié)點(diǎn)將成為瓶頸，進(jìn)而拖慢整個(gè)系統(tǒng)的處理速度。而在Spark　Streaming中，作業(yè)任務(wù)將會(huì)動(dòng)態(tài)地平衡分配給各個(gè)節(jié)點(diǎn)，如圖，即如果任務(wù)處理時(shí)間較長，分配的任務(wù)數(shù)量將少些；如果任務(wù)處理時(shí)間較短，則分配的任務(wù)數(shù)據(jù)將更多些。

• 快速故障恢復(fù)機(jī)制：在節(jié)點(diǎn)出現(xiàn)故障的情況下，傳統(tǒng)流處理系統(tǒng)會(huì)在其他的節(jié)點(diǎn)上重啟失敗的連續(xù)算子，并可能重新運(yùn)行先前數(shù)據(jù)流處理操作獲取部分丟失數(shù)據(jù)。在此過程中只有該節(jié)點(diǎn)重新處理失敗的過程，只有在新節(jié)點(diǎn)完成故障前所有計(jì)算后，整個(gè)系統(tǒng)才能夠處理其他任務(wù)。在Spark中，計(jì)算將分成許多小的任務(wù)，保證能在任何節(jié)點(diǎn)運(yùn)行后能夠正確進(jìn)行合并。因此，在某節(jié)點(diǎn)出現(xiàn)的故障的情況，這個(gè)節(jié)點(diǎn)的任務(wù)將均勻地分散到集群中的節(jié)點(diǎn)進(jìn)行計(jì)算，相對(duì)于傳遞故障恢復(fù)機(jī)制能夠更快地恢復(fù)。

　　批處理、流處理與交互式分析的一體化：Spark Streaming 是將流式計(jì)算分解成一系列短小的批處理作業(yè)，也就是把Spark Streaming 的輸入數(shù)據(jù)按照批處理大?。ㄈ鐜酌耄┓殖梢欢我欢蔚碾x散數(shù)據(jù)流（DStream），每一段數(shù)據(jù)都轉(zhuǎn)換成Spark 中的RDD，然后將Spark Streaming 中對(duì)DStream 流處理操作變?yōu)獒槍?duì)Spark 中對(duì)RDD 的批處理操作。另外，流數(shù)據(jù)都儲(chǔ)存在Spark 節(jié)點(diǎn)的內(nèi)存里，用戶便能根據(jù)所需進(jìn)行交互查詢。正是利用了Spark 這種工作機(jī)制將批處理、流處理與交互式工作結(jié)合在一起。

Spark SQL

　　Spark SQL 的前身是Shark，它發(fā)布時(shí)Hive 可以說是SQL on Hadoop 的唯一選擇（Hive 負(fù)責(zé)將SQL 編譯成可擴(kuò)展的MapReduce 作業(yè)），鑒于Hive 的性能以及與Spark 的兼容，Shark 由此而生。
　　Shark 即Hive on Spark，本質(zhì)上是通過Hive 的HQL 進(jìn)行解析，把HQL 翻譯成Spark 上對(duì)應(yīng)的RDD 操作，然后通過Hive 的Metadata 獲取數(shù)據(jù)庫里的表信息，實(shí)際為HDFS 上的數(shù)據(jù)和文件，最后由Shark 獲取并放到Spark 上運(yùn)算。Shark 的最大特性就是速度快，能與Hive 的完全兼容，并且可以在Shell 模式下使用rdd2sql 這樣的API，把HQL 得到的結(jié)果集繼續(xù)在Scala環(huán)境下運(yùn)算，支持用戶編寫簡單的機(jī)器學(xué)習(xí)或簡單分析處理函數(shù)，對(duì)HQL 結(jié)果進(jìn)一步分析計(jì)算。
　　在2014 年7 月1 日的Spark Summit 上，Databricks 宣布終止對(duì)Shark 的開發(fā)，將重點(diǎn)放到Spark SQL 上。在此次會(huì)議上，Databricks 表示，Shark 更多是對(duì)Hive 的改造，替換了Hive 的物理執(zhí)行引擎，使之有一個(gè)較快的處理速度。然而，不容忽視的是，Shark 繼承了大量的Hive代碼，因此給優(yōu)化和維護(hù)帶來大量的麻煩。隨著性能優(yōu)化和先進(jìn)分析整合的進(jìn)一步加深，基于MapReduce 設(shè)計(jì)的部分無疑成為了整個(gè)項(xiàng)目的瓶頸。因此，為了更好的發(fā)展，給用戶提供一個(gè)更好的體驗(yàn)，Databricks 宣布終止Shark 項(xiàng)目，從而將更多的精力放到Spark SQL 上。
　　Spark SQL 允許開發(fā)人員直接處理RDD，同時(shí)也可查詢?cè)?Hive 上存在的外部數(shù)據(jù)。SparkSQL 的一個(gè)重要特點(diǎn)是能夠統(tǒng)一處理關(guān)系表和RDD，使得開發(fā)人員可以輕松地使用SQL 命令進(jìn)行外部查詢，同時(shí)進(jìn)行更復(fù)雜的數(shù)據(jù)分析。

　　Spark SQL 的特點(diǎn)如下。

• 引入了新的RDD 類型SchemaRDD，可以像傳統(tǒng)數(shù)據(jù)庫定義表一樣來定義SchemaRDD。 SchemaRDD由定義了列數(shù)據(jù)類型的行對(duì)象構(gòu)成。SchemaRDD 既可以從RDD 轉(zhuǎn)換過 來，也可以從Parquet 文件讀入，還可以使用HiveQL從Hive 中獲取。

• 內(nèi)嵌了Catalyst 查詢優(yōu)化框架，在把SQL 解析成邏輯執(zhí)行計(jì)劃之后，利用Catalyst 包里的一些類和接口，執(zhí)行了一些簡單的執(zhí)行計(jì)劃優(yōu)化，最后變成RDD 的計(jì)算。

• 在應(yīng)用程序中可以混合使用不同來源的數(shù)據(jù)，如可以將來自HiveQL的數(shù)據(jù)和來自SQL的數(shù)據(jù)進(jìn)行Join 操作。 Shark的出現(xiàn)使得SQL-on-Hadoop 的性能比Hive 有了10～100 倍的提高，那么，擺脫了 Hive 的限制，Spark SQL的性能又有怎么樣的表現(xiàn)呢？雖然沒有Shark 相對(duì)于Hive 那樣矚目的 性能提升，但也表現(xiàn)得優(yōu)異，如圖（其中，右側(cè)數(shù)據(jù)為Spark　SQL）。

　　為什么Spark SQL 的性能會(huì)得到這么大的提升呢？主要是Spark SQL 在以下幾點(diǎn)做了優(yōu)化。

• 內(nèi)存列存儲(chǔ)（In-Memory Columnar Storage）：Spark SQL 的表數(shù)據(jù)在內(nèi)存中存儲(chǔ)不是采用原生態(tài)的JVM對(duì)象存儲(chǔ)方式，而是采用內(nèi)存列存儲(chǔ)。

• 字節(jié)碼生成技術(shù)（Bytecode Generation）：Spark 1.1.0 在Catalyst 模塊的Expressions 增加了Codegen 模塊，使用動(dòng)態(tài)字節(jié)碼生成技術(shù)，對(duì)匹配的表達(dá)式采用特定的代碼動(dòng)態(tài)編譯。另外對(duì)SQL 表達(dá)式都做了CG 優(yōu)化。CG優(yōu)化的實(shí)現(xiàn)主要還是依靠Scala 2.10運(yùn)行時(shí)的反射機(jī)制（Runtime Reflection）。

• Scala 代碼優(yōu)化：Spark SQL 在使用Scala編寫代碼的時(shí)候，盡量避免低效的、容易GC的代碼；盡管增加了編寫代碼的難度，但對(duì)于用戶來說接口統(tǒng)一。

BlinkDB

　　BlinkDB 是一個(gè)用于在海量數(shù)據(jù)上運(yùn)行交互式SQL 查詢的大規(guī)模并行查詢引擎，它允許用戶通過權(quán)衡數(shù)據(jù)精度來提升查詢響應(yīng)時(shí)間，其數(shù)據(jù)的精度被控制在允許的誤差范圍內(nèi)。為了達(dá)到這個(gè)目標(biāo)，BlinkDB 使用如下核心思想：

• 自適應(yīng)優(yōu)化框架，從原始數(shù)據(jù)隨著時(shí)間的推移建立并維護(hù)一組多維樣本。

• 動(dòng)態(tài)樣本選擇策略，選擇一個(gè)適當(dāng)大小的示例，該示例基于查詢的準(zhǔn)確性和響應(yīng)時(shí)間的緊迫性。和傳統(tǒng)關(guān)系型數(shù)據(jù)庫不同，BlinkDB是一個(gè)交互式查詢系統(tǒng)，就像一個(gè)蹺蹺板，用戶需要在查詢精度和查詢時(shí)間上做權(quán)衡；如果用戶想更快地獲取查詢結(jié)果，那么將犧牲查詢結(jié)果的精度；反之，用戶如果想獲取更高精度的查詢結(jié)果，就需要犧牲查詢響應(yīng)時(shí)間。下圖為BlinkDB架構(gòu)。

MLBase/MLlib

　　MLBase 是Spark 生態(tài)系統(tǒng)中專注于機(jī)器學(xué)習(xí)的組件，它的目標(biāo)是讓機(jī)器學(xué)習(xí)的門檻更低，讓一些可能并不了解機(jī)器學(xué)習(xí)的用戶能夠方便地使用MLBase。MLBase 分為4 個(gè)部分：MLRuntime、MLlib、MLI 和ML Optimizer。

• MLRuntime：是由Spark Core 提供的分布式內(nèi)存計(jì)算框架，運(yùn)行由Optimizer優(yōu)化過的算法進(jìn)行數(shù)據(jù)的計(jì)算并輸出分析結(jié)果。

• MLlib：是Spark 實(shí)現(xiàn)一些常見的機(jī)器學(xué)習(xí)算法和實(shí)用程序，包括分類、回歸、聚類、協(xié)同過濾、降維以及底層優(yōu)化。該算法可以進(jìn)行可擴(kuò)充。

• MLI：是一個(gè)進(jìn)行特征抽取和高級(jí)ML 編程抽象算法實(shí)現(xiàn)的API 或平臺(tái)。

• MLOptimizer：會(huì)選擇它認(rèn)為最適合的已經(jīng)在內(nèi)部實(shí)現(xiàn)好了的機(jī)器學(xué)習(xí)算法和相關(guān)參數(shù)，來處理用戶輸入的數(shù)據(jù)，并返回模型或其他幫助分析的結(jié)果。

　　MLBase 的核心是其優(yōu)化器（ML Optimizer），它可以把聲明式的任務(wù)轉(zhuǎn)化成復(fù)雜的學(xué)習(xí)計(jì)劃，最終產(chǎn)出最優(yōu)的模型和計(jì)算結(jié)果。MLBase 與其他機(jī)器學(xué)習(xí)Weka 和Mahout 不同，三者各有特色，具體內(nèi)容如下。

• MLBase 基于Spark，它是使用的是分布式內(nèi)存計(jì)算的；Weka 是一個(gè)單機(jī)的系統(tǒng)，而Mahout 是使用MapReduce 進(jìn)行處理數(shù)據(jù)（Mahout 正向使用Spark 處理數(shù)據(jù)轉(zhuǎn)變）。

• MLBase 是自動(dòng)化處理的；Weka 和Mahout 都需要使用者具備機(jī)器學(xué)習(xí)技能，來選擇自己想要的算法和參數(shù)來做處理。

• MLBase 提供了不同抽象程度的接口，可以由用戶通過該接口實(shí)現(xiàn)算法的擴(kuò)展。

GraphX

　　GraphX 最初是伯克利AMP 實(shí)驗(yàn)室的一個(gè)分布式圖計(jì)算框架項(xiàng)目，后來整合到Spark 中成為一個(gè)核心組件。它是Spark 中用于圖和圖并行計(jì)算的API，可以認(rèn)為是GraphLab 和Pregel 在Spark 上的重寫及優(yōu)化。跟其他分布式圖計(jì)算框架相比，GraphX 最大的優(yōu)勢(shì)是：在Spark 基礎(chǔ)上提供了一棧式數(shù)據(jù)解決方案，可以高效地完成圖計(jì)算的完整的流水作業(yè)。
　　GraphX 的核心抽象是Resilient Distributed Property Graph，一種點(diǎn)和邊都帶屬性的有向多重圖。GraphX 擴(kuò)展了Spark RDD 的抽象，它有Table 和Graph 兩種視圖，但只需要一份物理存儲(chǔ)，兩種視圖都有自己獨(dú)有的操作符，從而獲得了靈活操作和執(zhí)行效率。GraphX 的整體架構(gòu)中大部分的實(shí)現(xiàn)都是圍繞Partition 的優(yōu)化進(jìn)行的，這在某種程度上說明了，點(diǎn)分割的存儲(chǔ)和相應(yīng)的計(jì)算優(yōu)化的確是圖計(jì)算框架的重點(diǎn)和難點(diǎn)。
GraphX 的底層設(shè)計(jì)有以下幾個(gè)關(guān)鍵點(diǎn)。
（1）對(duì)Graph 視圖的所有操作，最終都會(huì)轉(zhuǎn)換成其關(guān)聯(lián)的Table 視圖的RDD 操作來完成。這樣對(duì)一個(gè)圖的計(jì)算，最終在邏輯上，等價(jià)于一系列RDD 的轉(zhuǎn)換過程。因此，Graph 最終具備了RDD 的3 個(gè)關(guān)鍵特性：Immutable、Distributed 和Fault-Tolerant。其中最關(guān)鍵的是Immutable（不變性）。邏輯上，所有圖的轉(zhuǎn)換和操作都產(chǎn)生了一個(gè)新圖；物理上，GraphX 會(huì)有一定程度的不變頂點(diǎn)和邊的復(fù)用優(yōu)化，對(duì)用戶透明。
（2）兩種視圖底層共用的物理數(shù)據(jù)，由RDD[Vertex-Partition]和RDD[EdgePartition]這兩個(gè)RDD 組成。點(diǎn)和邊實(shí)際都不是以表Collection[tuple] 的形式存儲(chǔ)的， 而是由VertexPartition/EdgePartition 在內(nèi)部存儲(chǔ)一個(gè)帶索引結(jié)構(gòu)的分片數(shù)據(jù)塊，以加速不同視圖下的遍歷速度。不變的索引結(jié)構(gòu)在RDD 轉(zhuǎn)換過程中是共用的，降低了計(jì)算和存儲(chǔ)開銷。
（3）圖的分布式存儲(chǔ)采用點(diǎn)分割模式，而且使用partitionBy 方法，由用戶指定不同的劃分策略（PartitionStrategy）。劃分策略會(huì)將邊分配到各個(gè)EdgePartition，頂點(diǎn)Master 分配到各個(gè)VertexPartition，EdgePartition 也會(huì)緩存本地邊關(guān)聯(lián)點(diǎn)的Ghost 副本。劃分策略的不同會(huì)影響到所需要緩存的Ghost 副本數(shù)量，以及每個(gè)EdgePartition 分配的邊的均衡程度，需要根據(jù)圖的結(jié)構(gòu)特征選取最佳策略。

SparkR

　　R 是遵循GNU 協(xié)議的一款開源、免費(fèi)的軟件，廣泛應(yīng)用于統(tǒng)計(jì)計(jì)算和統(tǒng)計(jì)制圖，但是它只能單機(jī)運(yùn)行。為了能夠使用R 語言分析大規(guī)模分布式的數(shù)據(jù)，伯克利分校AMP 實(shí)驗(yàn)室開發(fā)了SparkR，并在Spark 1.4 版本中加入了該組件。通過SparkR 可以分析大規(guī)模的數(shù)據(jù)集，并通過R Shell 交互式地在SparkR 上運(yùn)行作業(yè)。SparkR 特性如下：

• 提供了Spark 中彈性分布式數(shù)據(jù)集（RDDs）的API，用戶可以在集群上通過R Shell交互性地運(yùn)行Spark 任務(wù)。

• 支持序化閉包功能，可以將用戶定義函數(shù)中所引用到的變量自動(dòng)序化發(fā)送到集群中其他的機(jī)器上。

• SparkR 還可以很容易地調(diào)用R 開發(fā)包，只需要在集群上執(zhí)行操作前用includePackage讀取R 開發(fā)包就可以了。

下為SparkR 的處理流程示意圖。

Alluxio

　　Alluxio 是一個(gè)分布式內(nèi)存文件系統(tǒng)，它是一個(gè)高容錯(cuò)的分布式文件系統(tǒng)，允許文件以內(nèi)存的速度在集群框架中進(jìn)行可靠的共享，就像Spark 和 MapReduce 那樣。Alluxio 是架構(gòu)在最底層的分布式文件存儲(chǔ)和上層的各種計(jì)算框架之間的一種中間件。其主要職責(zé)是將那些不需要落地到DFS 里的文件，落地到分布式內(nèi)存文件系統(tǒng)中，來達(dá)到共享內(nèi)存，從而提高效率。同時(shí)可以減少內(nèi)存冗余、GC 時(shí)間等。
　　和Hadoop 類似，Alluxio 的架構(gòu)是傳統(tǒng)的Master-Slave 架構(gòu)，所有的Alluxio Worker 都被Alluxio Master 所管理，Alluxio Master 通過Alluxio Worker 定時(shí)發(fā)出的心跳來判斷Worker 是否已經(jīng)崩潰以及每個(gè)Worker 剩余的內(nèi)存空間量，為了防止單點(diǎn)問題使用了ZooKeeper 做了HA。
　　Alluxio 具有如下特性。

• AVA-Like File API：Alluxio 提供類似Java File 類的API。

• 兼容性：Alluxio 實(shí)現(xiàn)了HDFS 接口，所以Spark 和MapReduce 程序不需要任何修改即可運(yùn)行。

• 可插拔的底層文件系統(tǒng)：Alluxio是一個(gè)可插拔的底層文件系統(tǒng)，提供容錯(cuò)功能，它將內(nèi)存數(shù)據(jù)記錄在底層文件系統(tǒng)。它有一個(gè)通用的接口，可以很容易地插入到不同的底層文件系統(tǒng)。目前支持HDFS、S3、GlusterFS和單節(jié)點(diǎn)的本地文件系統(tǒng)，以后將支持更多的文件系統(tǒng)。Alluxio 所支持的應(yīng)用如下。

　　本文選自《圖解Spark：核心技術(shù)與案例實(shí)戰(zhàn)》，想及時(shí)獲得更多精彩文章，可在微信中搜索“博文視點(diǎn)”或者掃描下方二維碼并關(guān)注。