簡介

Spark是基于內(nèi)存的分布式計算框架。在迭代計算的場景下，數(shù)據(jù)處理過程中的數(shù)據(jù)可以存儲在內(nèi)存中，提供了比MapReduce高10到100倍的計算能力。Spark可以使用HDFS作為底層存儲，使用戶能夠快速地從MapReduce切換到Spark計算平臺上去。Spark提供一站式數(shù)據(jù)分析能力，包括小批量流式處理、離線批處理、SQL查詢、數(shù)據(jù)挖掘等，用戶可以在同一個應(yīng)用中無縫結(jié)合使用這些能力。Spark2x的開源新特性請參考Spark2x開源新特性說明。

更多關(guān)于Spark2x組件操作指導(dǎo)，請參考使用Spark/Spark2x。

Spark的特點如下：

結(jié)構(gòu)

Spark的架構(gòu)如圖1所示，各模塊的說明如表1所示。

圖1 Spark架構(gòu)

Spark原理

Spark的應(yīng)用運行架構(gòu)如圖2所示，運行流程如下所示：

1. 應(yīng)用程序（Application）是作為一個進(jìn)程的集合運行在集群上的，由Driver進(jìn)行協(xié)調(diào)。
2. 在運行一個應(yīng)用時，Driver會去連接集群管理器（Standalone、Mesos、YARN）申請運行Executor資源，并啟動ExecutorBackend。然后由集群管理器在不同的應(yīng)用之間調(diào)度資源。Driver同時會啟動應(yīng)用程序DAG調(diào)度、Stage劃分、Task生成。
3. 然后Spark會把應(yīng)用的代碼（傳遞給SparkContext的JAR或者Python定義的代碼）發(fā)送到Executor上。
4. 所有的Task執(zhí)行完成后，用戶的應(yīng)用程序運行結(jié)束。

圖2 Spark應(yīng)用運行架構(gòu)

Spark采用Master和Worker的模式，如圖3所示。用戶在Spark客戶端提交應(yīng)用程序，調(diào)度器將Job分解為多個Task發(fā)送到各個Worker中執(zhí)行，各個Worker將計算的結(jié)果上報給Driver（即Master），Driver聚合結(jié)果返回給客戶端。

圖3 Spark的Master和Worker

在此結(jié)構(gòu)中，有幾個說明點：

• 應(yīng)用之間是獨立的。

  每個應(yīng)用有自己的executor進(jìn)程，Executor啟動多個線程，并行地執(zhí)行任務(wù)。無論是在調(diào)度方面，或者是executor方面。各個Driver獨立調(diào)度自己的任務(wù)；不同的應(yīng)用任務(wù)運行在不同的JVM上，即不同的Executor。

• 不同Spark應(yīng)用之間是不共享數(shù)據(jù)的，除非把數(shù)據(jù)存儲在外部的存儲系統(tǒng)上（比如HDFS）。
• 因為Driver程序在集群上調(diào)度任務(wù)，所以Driver程序需要和worker節(jié)點比較近，比如在一個相同的局部網(wǎng)絡(luò)內(nèi)。

Spark on YARN有兩種部署模式：

• YARN-Cluster模式下，Spark的Driver會運行在YARN集群內(nèi)的ApplicationMaster進(jìn)程中，ApplicationMaster已經(jīng)啟動之后，提交任務(wù)的客戶端退出也不會影響任務(wù)的運行。
• YARN-Client模式下，Driver啟動在客戶端進(jìn)程內(nèi)，ApplicationMaster進(jìn)程只用來向YARN集群申請資源。

Spark Streaming原理

Spark Streaming是一種構(gòu)建在Spark上的實時計算框架，擴(kuò)展了Spark處理大規(guī)模流式數(shù)據(jù)的能力。當(dāng)前Spark支持兩種數(shù)據(jù)處理方式：Direct Streaming和Receiver方式。

Direct Streaming計算流程

Direct Streaming方式主要通過采用Direct API對數(shù)據(jù)進(jìn)行處理。以Kafka Direct接口為例，與啟動一個Receiver來連續(xù)不斷地從Kafka中接收數(shù)據(jù)并寫入到WAL中相比，Direct API簡單地給出每個batch區(qū)間需要讀取的偏移量位置。然后，每個batch的Job被運行，而對應(yīng)偏移量的數(shù)據(jù)在Kafka中已準(zhǔn)備好。這些偏移量信息也被可靠地存儲在checkpoint文件中，應(yīng)用失敗重啟時可以直接讀取偏移量信息。

圖4 Direct Kafka接口數(shù)據(jù)傳輸

需要注意的是，Spark Streaming可以在失敗后重新從Kafka中讀取并處理數(shù)據(jù)段。然而，由于語義僅被處理一次，重新處理的結(jié)果和沒有失敗處理的結(jié)果是一致的。

因此，Direct API消除了需要使用WAL和Receivers的情況，且確保每個Kafka記錄僅被接收一次，這種接收更加高效。使得Spark Streaming和Kafka可以很好地整合在一起。總體來說，這些特性使得流處理管道擁有高容錯性、高效性及易用性，因此推薦使用Direct Streaming方式處理數(shù)據(jù)。

Receiver計算流程

在一個Spark Streaming應(yīng)用開始時（也就是Driver開始時），相關(guān)的StreamingContext（所有流功能的基礎(chǔ)）使用SparkContext啟動Receiver成為長駐運行任務(wù)。這些Receiver接收并保存流數(shù)據(jù)到Spark內(nèi)存中以供處理。用戶傳送數(shù)據(jù)的生命周期如圖5所示：

圖5 數(shù)據(jù)傳輸生命周期

1. 接收數(shù)據(jù)（藍(lán)色箭頭）

   Receiver將數(shù)據(jù)流分成一系列小塊，存儲到Executor內(nèi)存中。另外，在啟用預(yù)寫日志（Write-ahead Log，簡稱WAL）以后，數(shù)據(jù)同時還寫入到容錯文件系統(tǒng)的預(yù)寫日志中。

2. 通知Driver（綠色箭頭）

   接收塊中的元數(shù)據(jù)（Metadata）被發(fā)送到Driver的StreamingContext。這個元數(shù)據(jù)包括：

   • 定位其在Executor內(nèi)存中數(shù)據(jù)位置的塊Reference ID。
   • 若啟用了WAL，還包括塊數(shù)據(jù)在日志中的偏移信息。
3. 處理數(shù)據(jù)（紅色箭頭）

   對每個批次的數(shù)據(jù)，StreamingContext使用Block信息產(chǎn)生RDD及其Job。StreamingContext通過運行任務(wù)處理Executor內(nèi)存中的Block來執(zhí)行Job。

4. 周期性地設(shè)置檢查點（橙色箭頭）
5. 為了容錯的需要，StreamingContext會周期性地設(shè)置檢查點，并保存到外部文件系統(tǒng)中。

容錯性

Spark及其RDD允許無縫地處理集群中任何Worker節(jié)點的故障。鑒于Spark Streaming建立于Spark之上，因此其Worker節(jié)點也具備了同樣的容錯能力。然而，由于Spark Streaming的長正常運行需求，其應(yīng)用程序必須也具備從Driver進(jìn)程（協(xié)調(diào)各個Worker的主要應(yīng)用進(jìn)程）故障中恢復(fù)的能力。使Spark Driver能夠容錯是件很棘手的事情，因為可能是任意計算模式實現(xiàn)的任意用戶程序。不過Spark Streaming應(yīng)用程序在計算上有一個內(nèi)在的結(jié)構(gòu)：在每批次數(shù)據(jù)周期性地執(zhí)行同樣的Spark計算。這種結(jié)構(gòu)允許把應(yīng)用的狀態(tài)（亦稱Checkpoint）周期性地保存到可靠的存儲空間中，并在Driver重新啟動時恢復(fù)該狀態(tài)。

對于文件這樣的源數(shù)據(jù)，這個Driver恢復(fù)機(jī)制足以做到零數(shù)據(jù)丟失，因為所有的數(shù)據(jù)都保存在了像HDFS這樣的容錯文件系統(tǒng)中。但對于像Kafka和Flume等其他數(shù)據(jù)源，有些接收到的數(shù)據(jù)還只緩存在內(nèi)存中，尚未被處理，就有可能會丟失。這是由于Spark應(yīng)用的分布操作方式引起的。當(dāng)Driver進(jìn)程失敗時，所有在Cluster Manager中運行的Executor，連同在內(nèi)存中的所有數(shù)據(jù)，也同時被終止。為了避免這種數(shù)據(jù)損失，Spark Streaming引進(jìn)了WAL功能。

WAL通常被用于 數(shù)據(jù)庫 和文件系統(tǒng)中，用來保證任何數(shù)據(jù)操作的持久性，即先將操作記入一個持久的日志，再對數(shù)據(jù)施加這個操作。若施加操作的過程中執(zhí)行失敗了，則通過讀取日志并重新施加前面指定的操作，系統(tǒng)就得到了恢復(fù)。下面介紹了如何利用這樣的概念保證接收到的數(shù)據(jù)的持久性。

Kafka數(shù)據(jù)源使用Receiver來接收數(shù)據(jù)，是Executor中的長運行任務(wù)，負(fù)責(zé)從數(shù)據(jù)源接收數(shù)據(jù)，并且在數(shù)據(jù)源支持時還負(fù)責(zé)確認(rèn)收到數(shù)據(jù)的結(jié)果（收到的數(shù)據(jù)被保存在Executor的內(nèi)存中，然后Driver在Executor中運行來處理任務(wù)）。

當(dāng)啟用了預(yù)寫日志以后，所有收到的數(shù)據(jù)同時還保存到了容錯文件系統(tǒng)的日志文件中。此時即使Spark Streaming失敗，這些接收到的數(shù)據(jù)也不會丟失。另外，接收數(shù)據(jù)的正確性只在數(shù)據(jù)被預(yù)寫到日志以后Receiver才會確認(rèn)，已經(jīng)緩存但還沒有保存的數(shù)據(jù)可以在Driver重新啟動之后由數(shù)據(jù)源再發(fā)送一次。這兩個機(jī)制確保了零數(shù)據(jù)丟失，即所有的數(shù)據(jù)或者從日志中恢復(fù)，或者由數(shù)據(jù)源重發(fā)。

如果需要啟用預(yù)寫日志功能，可以通過如下動作實現(xiàn)：

• 通過“streamingContext.checkpoint”(path-to-directory)設(shè)置checkpoint的目錄，這個目錄是一個HDFS的文件路徑，既用作保存流的checkpoint，又用作保存預(yù)寫日志。
• 設(shè)置SparkConf的屬性“spark.streaming.receiver.writeAheadLog.enable”為“true”（默認(rèn)值是“false”）。

在WAL被啟用以后，所有Receiver都獲得了能夠從可靠收到的數(shù)據(jù)中恢復(fù)的優(yōu)勢。建議緩存RDD時不采取多備份選項，因為用于預(yù)寫日志的容錯文件系統(tǒng)很可能也復(fù)制了數(shù)據(jù)。

在啟用了預(yù)寫日志以后，數(shù)據(jù)接收吞吐率會有降低。由于所有數(shù)據(jù)都被寫入容錯文件系統(tǒng)，文件系統(tǒng)的寫入吞吐率和用于 數(shù)據(jù)復(fù)制 的網(wǎng)絡(luò)帶寬，可能就是潛在的瓶頸了。在此情況下，需要創(chuàng)建更多的Receiver增加數(shù)據(jù)接收的并行度，或使用更好的硬件以增加容錯文件系統(tǒng)的吞吐率。

恢復(fù)流程

當(dāng)一個失敗的Driver重啟時，按如下流程啟動：

圖6 計算恢復(fù)流程

1. 恢復(fù)計算（橙色箭頭）

   使用checkpoint信息重啟Driver，重新構(gòu)造SparkContext并重啟Receiver。

2. 恢復(fù)元數(shù)據(jù)塊（綠色箭頭）

   為了保證能夠繼續(xù)下去所必備的全部元數(shù)據(jù)塊都被恢復(fù)。

3. 未完成作業(yè)的重新形成（紅色箭頭）

   由于失敗而沒有處理完成的批處理，將使用恢復(fù)的元數(shù)據(jù)再次產(chǎn)生RDD和對應(yīng)的作業(yè)。

4. 讀取保存在日志中的塊數(shù)據(jù)（藍(lán)色箭頭）

   在這些作業(yè)執(zhí)行時，塊數(shù)據(jù)直接從預(yù)寫日志中讀出。這將恢復(fù)在日志中可靠地保存的所有必要數(shù)據(jù)。

5. 重發(fā)尚未確認(rèn)的數(shù)據(jù)（紫色箭頭）

   失敗時沒有保存到日志中的緩存數(shù)據(jù)將由數(shù)據(jù)源再次發(fā)送。因為Receiver尚未對其確認(rèn)。

因此通過預(yù)寫日志和可靠的Receiver，Spark Streaming就可以保證沒有輸入數(shù)據(jù)會由于Driver的失敗而丟失。

SparkSQL和DataSet原理

SparkSQL

圖7 SparkSQL和DataSet

Spark SQL是Spark中用于結(jié)構(gòu)化數(shù)據(jù)處理的模塊。在Spark應(yīng)用中，可以無縫地使用SQL語句亦或是DataSet API對結(jié)構(gòu)化數(shù)據(jù)進(jìn)行查詢。

Spark SQL以及DataSet還提供了一種通用的訪問多數(shù)據(jù)源的方式，可訪問的數(shù)據(jù)源包括Hive、 CS V、Parquet、ORC、JSON和JDBC數(shù)據(jù)源，這些不同的數(shù)據(jù)源之間也可以實現(xiàn)互相操作。Spark SQL復(fù)用了Hive的前端處理邏輯和元數(shù)據(jù)處理模塊，使用Spark SQL可以直接對已有的Hive數(shù)據(jù)進(jìn)行查詢。

另外，SparkSQL還提供了諸如API、CLI、JDBC等諸多接口，對客戶端提供多樣接入形式。

Spark SQL Native DDL/DML

Spark1.5將很多DDL/DML命令下壓到Hive執(zhí)行，造成了與Hive的耦合，且在一定程度上不夠靈活（比如報錯不符合預(yù)期、結(jié)果與預(yù)期不一致等）。

Spark2x實現(xiàn)了命令的本地化，使用Spark SQL Native DDL/DML取代Hive執(zhí)行DDL/DML命令。一方面實現(xiàn)和Hive的解耦，另一方面可以對命令進(jìn)行定制化。

DataSet

DataSet是一個由特定域的對象組成的強類型集合，可通過功能或關(guān)系操作并行轉(zhuǎn)換其中的對象。 每個Dataset還有一個非類型視圖，即由多個列組成的DataSet，稱為DataFrame。

DataFrame是一個由多個列組成的結(jié)構(gòu)化的分布式數(shù)據(jù)集合，等同于關(guān)系數(shù)據(jù)庫中的一張表，或者是R/Python中的data frame。DataFrame是Spark SQL中的最基本的概念，可以通過多種方式創(chuàng)建，例如結(jié)構(gòu)化的數(shù)據(jù)集、Hive表、外部數(shù)據(jù)庫或者是RDD。

可用于DataSet的操作分為Transformation和Action。

• Transformation操作可生成新的DataSet。

  如map、filter、select和aggregate (groupBy)。

• Action操作可觸發(fā)計算及返回記結(jié)果。

  如count、show或向文件系統(tǒng)寫數(shù)據(jù)。

通常使用兩種方法創(chuàng)建一個DataSet：

• 最常見的方法是通過使用SparkSession上的read函數(shù)將Spark指向存儲系統(tǒng)上的某些文件。

  val people = spark.read.parquet("...").as[Person]  // Scala

  DataSet<Person> people = spark.read().parquet("...").as(Encoders.bean(Person.class));//Java

• 還可通過已存在的DataSet上可用的transformation操作來創(chuàng)建數(shù)據(jù)集。 例如，在已存在的DataSet上應(yīng)用map操作來創(chuàng)建新的DataSet：

  val names = people.map(_.name)  // 使用Scala語言，且names為一個Dataset

  Dataset<String> names = people.map((Person p) -> p.name, Encoders.STRING)); // Java

CLI和JD BCS erver

除了API編程接口之外，Spark SQL還對外提供CLI/JDBC接口：

• spark-shell和spark-sql腳本均可以提供CLI，以便于調(diào)試。
• JDBCServer提供JDBC接口，外部可直接通過發(fā)送JDBC請求來完成結(jié)構(gòu)化數(shù)據(jù)的計算和解析。

SparkSession原理

SparkSession是Spark2x編程的統(tǒng)一API，也可看作是讀取數(shù)據(jù)的統(tǒng)一入口。SparkSession提供了一個統(tǒng)一的入口點來執(zhí)行以前分散在多個類中的許多操作，并且還為那些較舊的類提供了訪問器方法，以實現(xiàn)最大的兼容性。

使用構(gòu)建器模式創(chuàng)建SparkSession。如果存在SparkSession，構(gòu)建器將自動重用現(xiàn)有的SparkSession；如果不存在則會創(chuàng)建一個SparkSession。 在I/O期間，在構(gòu)建器中設(shè)置的配置項將自動同步到Spark和Hadoop。

import org.apache.spark.sql.SparkSessionval sparkSession = SparkSession.builder  .master("local")  .appName("my-spark-app")  .config("spark.some.config.option", "config-value")  .getOrCreate()

• SparkSession可以用于對數(shù)據(jù)執(zhí)行SQL查詢，將結(jié)果返回為DataFrame。

  sparkSession.sql("select * from person").show

• SparkSession可以用于設(shè)置運行時的配置項，這些配置項可以在SQL中使用變量替換。

  sparkSession.conf.set("spark.some.config", "abcd")sparkSession.conf.get("spark.some.config")sparkSession.sql("select ${spark.some.config}")

• SparkSession包括一個“catalog”方法，其中包含使用Metastore（即數(shù)據(jù)目錄）的方法。方法返回值為數(shù)據(jù)集，可以使用相同的Dataset API來運行。

  val tables = sparkSession.catalog.listTables()val columns = sparkSession.catalog.listColumns("myTable")

• 底層SparkContext可以通過SparkSession的SparkContext API訪問。

  val sparkContext = sparkSession.sparkContext

Structured Streaming原理

Structured Streaming是構(gòu)建在Spark SQL引擎上的流式數(shù)據(jù)處理引擎，用戶可以使用Scala、Java、Python或R中的Dataset/DataFrame API進(jìn)行流數(shù)據(jù)聚合運算、按事件時間窗口計算、流流Join等操作。當(dāng)流數(shù)據(jù)連續(xù)不斷地產(chǎn)生時，Spark SQL將會增量的、持續(xù)不斷地處理這些數(shù)據(jù)并將結(jié)果更新到結(jié)果集中。同時，系統(tǒng)通過checkpoint和Write Ahead Logs確保端到端的完全一次性容錯保證。

Structured Streaming的核心是將流式的數(shù)據(jù)看成一張不斷增加的數(shù)據(jù)庫表，這種流式的數(shù)據(jù)處理模型類似于數(shù)據(jù)塊處理模型，可以把靜態(tài)數(shù)據(jù)庫表的一些查詢操作應(yīng)用在流式計算中，Spark執(zhí)行標(biāo)準(zhǔn)的SQL查詢，從不斷增加的無邊界表中獲取數(shù)據(jù)。

圖8 Structured Streaming無邊界表

每一條查詢的操作都會產(chǎn)生一個結(jié)果集Result Table。每一個觸發(fā)間隔，當(dāng)新的數(shù)據(jù)新增到表中，都會最終更新Result Table。無論何時結(jié)果集發(fā)生了更新，都能將變化的結(jié)果寫入一個外部的存儲系統(tǒng)。

圖9 Structured Streaming數(shù)據(jù)處理模型

Structured Streaming在OutPut階段可以定義不同的存儲方式，有如下3種：

• Complete Mode：整個更新的結(jié)果集都會寫入外部存儲。整張表的寫入操作將由外部存儲系統(tǒng)的連接器完成。
• Append Mode：當(dāng)時間間隔觸發(fā)時，只有在Result Table中新增加的數(shù)據(jù)行會被寫入外部存儲。這種方式只適用于結(jié)果集中已經(jīng)存在的內(nèi)容不希望發(fā)生改變的情況下，如果已經(jīng)存在的數(shù)據(jù)會被更新，不適合使用此種方式。
• Update Mode：當(dāng)時間間隔觸發(fā)時，只有在Result Table中被更新的數(shù)據(jù)才會被寫入外部存儲系統(tǒng)。注意，和Complete Mode方式的不同之處是不更新的結(jié)果集不會寫入外部存儲。

基本概念

• RDD

  即彈性分布數(shù)據(jù)集（Resilient Distributed Dataset），是Spark的核心概念。指的是一個只讀的，可分區(qū)的分布式數(shù)據(jù)集，這個數(shù)據(jù)集的全部或部分可以緩存在內(nèi)存中，在多次計算間重用。

  RDD的生成：

  • 從HDFS輸入創(chuàng)建，或從與Hadoop兼容的其他存儲系統(tǒng)中輸入創(chuàng)建。
  • 從父RDD轉(zhuǎn)換得到新RDD。
  • 從數(shù)據(jù)集合轉(zhuǎn)換而來，通過編碼實現(xiàn)。

  RDD的存儲：

  • 用戶可以選擇不同的存儲級別緩存RDD以便重用（RDD有11種存儲級別）。
  • 當(dāng)前RDD默認(rèn)是存儲于內(nèi)存，但當(dāng)內(nèi)存不足時，RDD會溢出到磁盤中。
• Dependency（RDD的依賴）

  RDD的依賴分別為：窄依賴和寬依賴。

  圖10 RDD的依賴
  
  • 窄依賴：指父RDD的每一個分區(qū)最多被一個子RDD的分區(qū)所用。
  • 寬依賴：指子RDD的分區(qū)依賴于父RDD的所有分區(qū)。

  窄依賴對優(yōu)化很有利。邏輯上，每個RDD的算子都是一個fork/join（此join非上文的join算子，而是指同步多個并行任務(wù)的barrier）：把計算fork到每個分區(qū)，算完后join，然后fork/join下一個RDD的算子。如果直接翻譯到物理實現(xiàn)，是很不經(jīng)濟(jì)的：一是每一個RDD（即使是中間結(jié)果）都需要物化到內(nèi)存或存儲中，費時費空間；二是join作為全局的barrier，是很昂貴的，會被最慢的那個節(jié)點拖死。如果子RDD的分區(qū)到父RDD的分區(qū)是窄依賴，就可以實施經(jīng)典的fusion優(yōu)化，把兩個fork/join合為一個；如果連續(xù)的變換算子序列都是窄依賴，就可以把很多個fork/join并為一個，不但減少了大量的全局barrier，而且無需物化很多中間結(jié)果RDD，這將極大地提升性能。Spark把這個叫做流水線（pipeline）優(yōu)化。

• Transformation和Action（RDD的操作）

  對RDD的操作包含Transformation（返回值還是一個RDD）和Action（返回值不是一個RDD）兩種。RDD的操作流程如圖11所示。其中Transformation操作是Lazy的，也就是說從一個RDD轉(zhuǎn)換生成另一個RDD的操作不是馬上執(zhí)行，Spark在遇到Transformations操作時只會記錄需要這樣的操作，并不會去執(zhí)行，需要等到有Actions操作的時候才會真正啟動計算過程進(jìn)行計算。Actions操作會返回結(jié)果或把RDD數(shù)據(jù)寫到存儲系統(tǒng)中。Actions是觸發(fā)Spark啟動計算的動因。

  圖11 RDD操作示例
  

  RDD看起來與Scala集合類型沒有太大差別，但數(shù)據(jù)和運行模型大相迥異。

  val file = sc.textFile("hdfs://...")val errors = file.filter(_.contains("ERROR"))errors.cache()errors.count()

  1. textFile算子從HDFS讀取日志文件，返回file（作為RDD）。
  2. filter算子篩出帶“ERROR”的行，賦給errors（新RDD）。filter算子是一個Transformation操作。
  3. cache算子緩存下來以備未來使用。
  4. count算子返回errors的行數(shù)。count算子是一個Action操作。
  Transformation操作可以分為如下幾種類型：

  • 視RDD的元素為簡單元素。

    輸入輸出一對一，且結(jié)果RDD的分區(qū)結(jié)構(gòu)不變，主要是map。

    輸入輸出一對多，且結(jié)果RDD的分區(qū)結(jié)構(gòu)不變，如flatMap（map后由一個元素變?yōu)橐粋€包含多個元素的序列，然后展平為一個個的元素）。

    輸入輸出一對一，但結(jié)果RDD的分區(qū)結(jié)構(gòu)發(fā)生了變化，如union（兩個RDD合為一個，分區(qū)數(shù)變?yōu)閮蓚€RDD分區(qū)數(shù)之和）、coalesce（分區(qū)減少）。

    從輸入中選擇部分元素的算子，如filter、distinct（去除重復(fù)元素）、subtract（本RDD有、其他RDD無的元素留下來）和sample（采樣）。

  • 視RDD的元素為Key-Value對。

    對單個RDD做一對一運算，如mapValues（保持源RDD的分區(qū)方式，這與map不同）；

    對單個RDD重排，如sort、partitionBy（實現(xiàn)一致性的分區(qū)劃分，這個對數(shù)據(jù)本地性優(yōu)化很重要）；

    對單個RDD基于key進(jìn)行重組和reduce，如groupByKey、reduceByKey；

    對兩個RDD基于key進(jìn)行join和重組，如join、cogroup。

    

    后三種操作都涉及重排，稱為shuffle類操作。

  Action操作可以分為如下幾種：

  • 生成標(biāo)量，如count（返回RDD中元素的個數(shù)）、reduce、fold/aggregate（返回幾個標(biāo)量）、take（返回前幾個元素）。
  • 生成Scala集合類型，如collect（把RDD中的所有元素導(dǎo)入Scala集合類型）、lookup（查找對應(yīng)key的所有值）。
  • 寫入存儲，如與前文textFile對應(yīng)的saveAsTextFile。
  • 還有一個檢查點算子checkpoint。當(dāng)Lineage特別長時（這在圖計算中時常發(fā)生），出錯時重新執(zhí)行整個序列要很長時間，可以主動調(diào)用checkpoint把當(dāng)前數(shù)據(jù)寫入穩(wěn)定存儲，作為檢查點。
• Shuffle

  Shuffle是MapReduce框架中的一個特定的phase，介于Map phase和Reduce phase之間，當(dāng)Map的輸出結(jié)果要被Reduce使用時，每一條輸出結(jié)果需要按key哈希，并且分發(fā)到對應(yīng)的Reducer上去，這個過程就是shuffle。由于shuffle涉及到了磁盤的讀寫和網(wǎng)絡(luò)的傳輸，因此shuffle性能的高低直接影響到了整個程序的運行效率。

  下圖清晰地描述了MapReduce算法的整個流程。

  圖12 算法流程
  

  概念上shuffle就是一個溝通數(shù)據(jù)連接的橋梁，實際上shuffle這一部分是如何實現(xiàn)的呢，下面就以Spark為例講解shuffle在Spark中的實現(xiàn)。

  Shuffle操作將一個Spark的Job分成多個Stage，前面的stages會包括一個或多個ShuffleMapTasks，最后一個stage會包括一個或多個ResultTask。

• Spark Application的結(jié)構(gòu)

  Spark Application的結(jié)構(gòu)可分為兩部分：初始化SparkContext和主體程序。

  • 初始化SparkContext：構(gòu)建Spark Application的運行環(huán)境。

    構(gòu)建SparkContext對象，如：

    new SparkContext(master, appName, [SparkHome], [jars])

    參數(shù)介紹：

    master：連接字符串，連接方式有l(wèi)ocal、yarn-cluster、yarn-client等。

    appName：構(gòu)建的Application名稱。

    SparkHome：集群中安裝Spark的目錄。

    jars：應(yīng)用程序代碼和依賴包。

  • 主體程序：處理數(shù)據(jù)

  提交Application的描述請參見：https://archive.apache.org/dist/spark/docs/3.1.1/submitting-applications.html

• Spark shell命令

  Spark基本shell命令，支持提交Spark應(yīng)用。命令為：

  ./bin/spark-submit \  --class <main-class> \  --master <master-url> \  ... # other options  <application-jar> \  [application-arguments]

  參數(shù)解釋：

  --class：Spark應(yīng)用的類名。

  --master：Spark用于所連接的master，如yarn-client，yarn-cluster等。

  application-jar：Spark應(yīng)用的jar包的路徑。

  application-arguments：提交Spark應(yīng)用的所需要的參數(shù)（可以為空）。

• Spark JobHistory Server

  用于監(jiān)控正在運行的或者歷史的Spark作業(yè)在Spark框架各個階段的細(xì)節(jié)以及提供日志顯示，幫助用戶更細(xì)粒度地去開發(fā)、配置和調(diào)優(yōu)作業(yè)。