摘要：原文基本概念解析寫在前面本系列是綜合了自己在學(xué)習(xí)過程中的理解記錄對(duì)參考文章中的一些理解個(gè)人實(shí)踐過程中的一些心得而來。是項(xiàng)目組設(shè)計(jì)用來表示數(shù)據(jù)集的一種數(shù)據(jù)結(jié)構(gòu)。

原文：『 Spark 』2. spark 基本概念解析

本系列是綜合了自己在學(xué)習(xí)spark過程中的理解記錄 ＋ 對(duì)參考文章中的一些理解 ＋ 個(gè)人實(shí)踐spark過程中的一些心得而來。寫這樣一個(gè)系列僅僅是為了梳理個(gè)人學(xué)習(xí)spark的筆記記錄，并非為了做什么教程，所以一切以個(gè)人理解梳理為主，沒有必要的細(xì)節(jié)就不會(huì)記錄了。若想深入了解，最好閱讀參考文章和官方文檔。

其次，本系列是基于目前最新的 spark 1.6.0 系列開始的，spark 目前的更新速度很快，記錄一下版本好還是必要的。
最后，如果各位覺得內(nèi)容有誤，歡迎留言備注，所有留言 24 小時(shí)內(nèi)必定回復(fù)，非常感謝。
Tips: 如果插圖看起來不明顯，可以：1. 放大網(wǎng)頁(yè)；2. 新標(biāo)簽中打開圖片，查看原圖哦。

用戶在 spark 上構(gòu)建的程序，包含了 driver 程序以及在集群上運(yùn)行的程序代碼，物理機(jī)器上涉及了 driver，master，worker 三個(gè)節(jié)點(diǎn).

創(chuàng)建 sc ，定義 udf 函數(shù)，定義一個(gè) spark 應(yīng)用程序所需要的三大步驟的邏輯：加載數(shù)據(jù)集，處理數(shù)據(jù)，結(jié)果展示。

集群的資源管理器，在集群上獲取資源的外部服務(wù)。
拿 Yarn 舉例，客戶端程序會(huì)向 Yarn 申請(qǐng)計(jì)算我這個(gè)任務(wù)需要多少的 memory，多少 CPU，etc。
然后 Cluster Manager 會(huì)通過調(diào)度告訴客戶端可以使用，然后客戶端就可以把程序送到每個(gè) Worker Node 上面去執(zhí)行了。

集群中任何一個(gè)可以運(yùn)行spark應(yīng)用代碼的節(jié)點(diǎn)。Worker Node就是物理節(jié)點(diǎn)，可以在上面啟動(dòng)Executor進(jìn)程。

在每個(gè) Worker Node 上為某應(yīng)用啟動(dòng)的一個(gè)進(jìn)程，該進(jìn)程負(fù)責(zé)運(yùn)行任務(wù)，并且負(fù)責(zé)將數(shù)據(jù)存在內(nèi)存或者磁盤上，每個(gè)任務(wù)都有各自獨(dú)立的 Executor。
Executor 是一個(gè)執(zhí)行 Task 的容器。它的主要職責(zé)是：

初始化程序要執(zhí)行的上下文 SparkEnv，解決應(yīng)用程序需要運(yùn)行時(shí)的 jar 包的依賴，加載類。

同時(shí)還有一個(gè) ExecutorBackend 向 cluster manager 匯報(bào)當(dāng)前的任務(wù)狀態(tài)，這一方面有點(diǎn)類似 hadoop的 tasktracker 和 task。

總結(jié)：Executor 是一個(gè)應(yīng)用程序運(yùn)行的監(jiān)控和執(zhí)行容器。

包含很多 task 的并行計(jì)算，可以認(rèn)為是 Spark RDD 里面的 action，每個(gè) action 的觸發(fā)會(huì)生成一個(gè)job。
用戶提交的 Job 會(huì)提交給 DAGScheduler，Job 會(huì)被分解成 Stage，Stage 會(huì)被細(xì)化成 Task，Task 簡(jiǎn)單的說就是在一個(gè)數(shù)據(jù) partition 上的單個(gè)數(shù)據(jù)處理流程。關(guān)于 job，stage，task，詳細(xì)可以參考這篇文章：『 Spark 』6. 深入研究 spark 運(yùn)行原理之 job, stage, task

A job is triggered by an action, like count() or saveAsTextFile(), click on a job to see info about the stages of tasks inside it.

一個(gè) Job 會(huì)被拆分為多組 Task，每組任務(wù)被稱為一個(gè) Stage 就像 Map Stage， Reduce Stage。

Stage 的劃分在 RDD 的論文中有詳細(xì)的介紹，簡(jiǎn)單的說是以 shuffle 和 result 這兩種類型來劃分。
在 Spark 中有兩類 task:

shuffleMapTask

輸出是shuffle所需數(shù)據(jù), stage的劃分也以此為依據(jù)，shuffle之前的所有變換是一個(gè)stage，shuffle之后的操作是另一個(gè)stage。

resultTask，

輸出是result，比如 rdd.parallize(1 to 10).foreach(println) 這個(gè)操作沒有shuffle，直接就輸出了，那么只有它的task是resultTask，stage也只有一個(gè)；如果是rdd.map(x => (x, 1)).reduceByKey(_ + _).foreach(println), 這個(gè)job因?yàn)橛衦educe，所以有一個(gè)shuffle過程，那么reduceByKey之前的是一個(gè)stage，執(zhí)行shuffleMapTask，輸出shuffle所需的數(shù)據(jù)，reduceByKey到最后是一個(gè)stage，直接就輸出結(jié)果了。如果job中有多次shuffle，那么每個(gè)shuffle之前都是一個(gè)stage。

被送到 executor 上的工作單元。

Partition 類似 hadoop 的 Split，計(jì)算是以 partition 為單位進(jìn)行的，當(dāng)然 partition 的劃分依據(jù)有很多，這是可以自己定義的，像 HDFS 文件，劃分的方式就和 MapReduce 一樣，以文件的 block 來劃分不同的 partition。總而言之，Spark 的 partition 在概念上與 hadoop 中的 split 是相似的，提供了一種劃分?jǐn)?shù)據(jù)的方式。

先看看原文 [Resilient Distributed Datasets: A Fault-Tolerant Abstraction for
In-Memory Cluster Computing](http://litaotao.github.io/files/spark-rd... 是怎么介紹 RDD 的。

a distributed memory abstraction that lets programmers perform in-memory computations on large clusters in a fault-tolerant manner.

RDDs are motivated by two types of applications that current computing frameworks handle inefficiently:

iterative algorithms;

interactive data mining tools;

In both cases, keeping data in memory can improve performance by an order of magnitude.

To achieve fault tolerance efficiently, RDDs provide a restricted form of shared memory, based on coarsegrained transformations rather than fine-grained updates to shared state. However, we show that RDDs are expressive enough to capture a wide class of computations, including recent specialized programming models for iterative jobs, such as Pregel, and new applications that these models do not capture. We have implemented RDDs in a system called Spark, which we evaluate through a variety of user applications and benchmarks.

每個(gè)RDD有5個(gè)主要的屬性：

一組分片（partition），即數(shù)據(jù)集的基本組成單位

一個(gè)計(jì)算每個(gè)分片的函數(shù)

對(duì)parent RDD的依賴，這個(gè)依賴描述了RDD之間的lineage

對(duì)于key-value的RDD，一個(gè)Partitioner，這是可選擇的

一個(gè)列表，存儲(chǔ)存取每個(gè)partition的preferred位置。對(duì)于一個(gè)HDFS文件來說，存儲(chǔ)每個(gè)partition所在的塊的位置。這也是可選擇的

　　把上面這5個(gè)主要的屬性總結(jié)一下，可以得出RDD的大致概念。首先要知道，RDD大概是這樣一種表示數(shù)據(jù)集的東西，它具有以上列出的一些屬性。是spark項(xiàng)目組設(shè)計(jì)用來表示數(shù)據(jù)集的一種數(shù)據(jù)結(jié)構(gòu)。而spark項(xiàng)目組為了讓RDD能handle更多的問題，又規(guī)定RDD應(yīng)該是只讀的，分區(qū)記錄的一種數(shù)據(jù)集合中。可以通過兩種方式來創(chuàng)建RDD：一種是基于物理存儲(chǔ)中的數(shù)據(jù)，比如說磁盤上的文件；另一種，也是大多數(shù)創(chuàng)建RDD的方式，即通過其他RDD來創(chuàng)建【以后叫做轉(zhuǎn)換】而成。而正因?yàn)镽DD滿足了這么多特性，所以spark把RDD叫做Resilient Distributed Datasets，中文叫做彈性分布式數(shù)據(jù)集。很多文章都是先講RDD的定義，概念，再來說RDD的特性。我覺得其實(shí)也可以倒過來，通過RDD的特性反過來理解RDD的定義和概念，通過這種由果溯因的方式來理解RDD也未嘗不可。反正對(duì)我個(gè)人而言這種方式是挺好的。

RDD是Spark的核心，也是整個(gè)Spark的架構(gòu)基礎(chǔ)，可以總下出幾個(gè)它的特性來：

它是不變的數(shù)據(jù)結(jié)構(gòu)存儲(chǔ)

它是支持跨集群的分布式數(shù)據(jù)結(jié)構(gòu)

可以根據(jù)數(shù)據(jù)記錄的key對(duì)結(jié)構(gòu)進(jìn)行分區(qū)

提供了粗粒度的操作，且這些操作都支持分區(qū)

它將數(shù)據(jù)存儲(chǔ)在內(nèi)存中，從而提供了低延遲性

關(guān)于 rdd 的更多詳情，可以參考這篇文章：『 Spark 』4. spark 之 RDD

先看看 api 文檔里是怎么說的：parallelize

parallelize(c, numSlices=None)

Distribute a local Python collection to form an RDD. Using xrange is recommended if the input represents a range for performance.

簡(jiǎn)單的說，parallelize 就是把 driver 端定義的一個(gè)數(shù)據(jù)集，或者一個(gè)獲取數(shù)據(jù)集的生成器，分發(fā)到 worker 上的 executor 中，以供后續(xù)分析。這種方式在測(cè)試代碼邏輯時(shí)經(jīng)常用到，但在構(gòu)建真正的 spark 應(yīng)用程序時(shí)很少會(huì)用到，一般都是從 hdfs 或者數(shù)據(jù)庫(kù)去讀取數(shù)據(jù)。

提交 spark 應(yīng)用時(shí)，spark 會(huì)把應(yīng)用代碼分發(fā)到所有的 worker 上面，應(yīng)用依賴的包需要在所有的worker上都存在，有兩種解決 worker 上相關(guān)包依賴的問題：

選用一些工具統(tǒng)一部署 spark cluster；

在提交 spark 應(yīng)用的時(shí)候，指定應(yīng)用依賴的相關(guān)包，把 應(yīng)用代碼，應(yīng)用依賴包 一起分發(fā)到 worker；

cache 是否支持 priority，目前不支持，而且 spark 里面對(duì) rdd 的 cache 和我們常見的緩存系統(tǒng)是不一樣的。細(xì)節(jié)可以找我討論。

The number of cores to use on each executor. For YARN and standalone mode only. In standalone mode, setting this parameter allows an application to run multiple executors on the same worker, provided that there are enough cores on that worker. Otherwise, only one executor per application will run on each worker.

每一個(gè) core，相當(dāng)于一個(gè) worker 上的進(jìn)程，這些進(jìn)程會(huì)同時(shí)執(zhí)行分配到這個(gè) worker 上的任務(wù)。簡(jiǎn)單的說，就是 spark manager 把一個(gè) job 切分幾個(gè) task 分發(fā)到 worker 上同步執(zhí)行，而每個(gè) worker 把分配給自己的 task 再切分成幾個(gè) subtask，分配給當(dāng)前 worker 上的不同進(jìn)程。

分配給 spark 應(yīng)用的內(nèi)存是僅僅給 cache 數(shù)據(jù)用嗎？

不是，分配給 spark 應(yīng)用的內(nèi)存有三個(gè)方面的應(yīng)用：

spark 本身

spark 應(yīng)用

spark 應(yīng)用過程中 runtime 使用，比如 UDF 函數(shù)

spark 應(yīng)用中的 cache

RDD 之間的依賴類別［ 或者，創(chuàng)建一個(gè) RDD 的不同方法 ］

所謂本地內(nèi)存，是指在 driver 端的程序所需要的內(nèi)存，由 driver 機(jī)器提供，一般用來生成測(cè)試數(shù)據(jù)，接受運(yùn)算結(jié)果等；
所謂集群內(nèi)存，是指提交到集群的作業(yè)能夠向集群申請(qǐng)的最多內(nèi)存使用量，一般用來存儲(chǔ)關(guān)鍵數(shù)據(jù)；

可以在啟動(dòng) spark 應(yīng)用的時(shí)候申請(qǐng)；完全可控。

FIFO 資源分配方式。

spark app 的起點(diǎn)和入口，一般用來加載數(shù)據(jù)集，生成第一個(gè) rdd。

不會(huì)，只會(huì)cache一次。stackoverflow

下一篇，通過幾個(gè)簡(jiǎn)單的例子來介紹 spark 的基本編程模式。

spark-rdd-paper : Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing

spark python API

『 Spark 』1. spark 簡(jiǎn)介

『 Spark 』2. spark 基本概念解析

『 Spark 』3. spark 編程模式

『 Spark 』4. spark 之 RDD

『 Spark 』5. 這些年，你不能錯(cuò)過的 spark 學(xué)習(xí)資源

『 Spark 』6. 深入研究 spark 運(yùn)行原理之 job, stage, task

『 Spark 』7. 使用 Spark DataFrame 進(jìn)行大數(shù)據(jù)分析