Part 1

1. Spark計算模型

1.1 Spark程序模型

首先通過一個簡單的實例了解Spark的程序模型杨耙。

1）SparkContext中的textFile函數(shù)從HDFS讀取日志文件盆耽，輸出變量file掷贾。

valfile=sc.textFile("hdfs://xxx")

2）RDD中的filter函數(shù)過濾帶“ERROR”的行涩拙，輸出errors（errors也是一個RDD）蚊丐。

valerrors=file.filter(line=>line.contains("ERROR")

3）RDD的count函數(shù)返回“ERROR”的行數(shù)：

errors.count()。

其示意圖如下：

從圖中可以看到馍迄，RDD由很多partition組成福也，一個partition對應到物理模塊上算是一個block，由block

manager統(tǒng)一管理攀圈。

Spark程序模型的主要思想是RDD（Resilient

Distributed Dataset）暴凑，把所有計算的數(shù)據(jù)保存在分布式的內存中。在迭代計算中赘来，通常情況下现喳，都是對同一的數(shù)據(jù)集做反復的迭代計算，數(shù)據(jù)保存在內存中犬辰，將大大提高性能嗦篱。RDD就是數(shù)據(jù)partition方式保存在cluster的內存中。操作有兩種：transformation和action忧风，transform就是把一種RDD轉換為另一個RDD默色，和Hadoop的map操作很類似，只是定義operator比較豐富（map狮腿，join腿宰，filter，groupByKey等操作）缘厢，action就類似于hadoop的reduce吃度，其輸出是一個aggregation函數(shù)的值如count，或者是一個集合（collection）贴硫。

1.2 彈性分布式數(shù)據(jù)集（RDD）

RDD是Spark的核心數(shù)據(jù)結構椿每，通過RDD的依賴關系形成Spark的調度順序。通過對RDD的操作形成整個Spark程序英遭。

1.2.1?RDD的四種創(chuàng)建方式

1）從Hadoop文件系統(tǒng)（或與Hadoop兼容的其他持久化存儲系統(tǒng)间护，如Hive、Cassandra挖诸、Hbase）輸入（如HDFS）創(chuàng)建汁尺。

2）從父RDD轉換得到新的RDD。

3）調用SparkContext方法的parallelize多律，將Driver上的數(shù)據(jù)集并行化痴突，轉化為分布式的RDD搂蜓。

4）更改RDD的持久性（persistence），例如cache()函數(shù)辽装，默認RDD計算后會在內存中清除帮碰。通過cache函數(shù)將計算后的RDD緩存在內存中。

1.2.2?RDD的兩種操作算子

對于RDD可以有兩種計算操作算子：Transformation（變換）與Action（行動）拾积。

1）Transformation（變換）殉挽。Transformation操作是延遲計算的，也就是說從一個RDD轉換生成另一個RDD的轉換操作不是馬上執(zhí)行殷勘，需要等到有Actions操作時此再，才真正觸發(fā)運算。

2）Action（行動）Action算子會觸發(fā)Spark提交作業(yè)（Job）玲销，并將數(shù)據(jù)輸出到Spark系統(tǒng)输拇。

1.2.3?RDD的重要內部屬性

1）分區(qū)列表。

2）計算每個分片的函數(shù)贤斜。

3）對父RDD的依賴列表策吠。

4）對Key-Value對數(shù)據(jù)類型RDD的分區(qū)器，控制分區(qū)策略和分區(qū)數(shù)瘩绒。

5）每個數(shù)據(jù)分區(qū)的地址列表（如HDFS上的數(shù)據(jù)塊的地址）猴抹。

1.2.4?RDD與DSM的對比

通過上述表格可以看到，RDD有更好的容錯性锁荔，采用血統(tǒng)機制后蟀给，可以不用回滾程序實現(xiàn)容錯。

RDD和DSM對比主要有如下兩個優(yōu)勢：

1）對于RDD中的批量操作阳堕，運行時將根據(jù)數(shù)據(jù)存放的位置來調度任務跋理，從而提高性能。

2）對于掃描類型操作恬总，如果內存不足以緩存整個RDD前普，就進行部分緩存，將內存容納不下的分區(qū)存儲到磁盤上壹堰。

1.3 Spark的數(shù)據(jù)存儲

Spark數(shù)據(jù)存儲的核心是彈性分布式數(shù)據(jù)集（RDD）拭卿。RDD可以被抽象地理解為一個大的數(shù)組（Array），但是這個數(shù)組是分布在集群上的贱纠。邏輯上RDD的每個分區(qū)叫一個Partition峻厚。

在Spark的執(zhí)行過程中，RDD經(jīng)歷一個個的Transfomation算子之后谆焊，最后通過Action算子進行觸發(fā)操作目木。

RDD之間通過Lineage產(chǎn)生依賴關系，這個關系在容錯中有很重要的作用。

RDD會被劃分成很多的分區(qū)分布到集群的多個節(jié)點中刽射。分區(qū)是個邏輯概念，變換前后的新舊分區(qū)在物理上可能是同一塊內存存儲剃执。這是很重要的優(yōu)化誓禁，以防止函數(shù)式數(shù)據(jù)不變性導致的內存需求無限擴張。

在上圖中肾档，在物理上摹恰，RDD對象實質上是一個元數(shù)據(jù)結構，存儲著Block怒见、Node等的映射關系俗慈，以及其他的元數(shù)據(jù)信息。

每個Block中存儲著RDD所有數(shù)據(jù)項的一個子集遣耍。

1.4 Spark的算子作用與分類

1.4.1 算子的作用

算子是RDD中定義的函數(shù)闺阱，可以對RDD中的數(shù)據(jù)進行轉換和操作。

1）輸入：在Spark程序運行中舵变，數(shù)據(jù)從外部數(shù)據(jù)空間（如分布式存儲：textFile讀取HDFS等酣溃，parallelize方法輸入Scala集合或數(shù)據(jù)）輸入Spark，數(shù)據(jù)進入Spark運行時數(shù)據(jù)空間纪隙，轉化為Spark中的數(shù)據(jù)塊赊豌，通過BlockManager進行管理。

2）運行：在Spark數(shù)據(jù)輸入形成RDD后便可以通過變換算子绵咱，如fliter等碘饼，對數(shù)據(jù)進行操作并將RDD轉化為新的RDD，通過Action算子悲伶，觸發(fā)Spark提交作業(yè)艾恼。如果數(shù)據(jù)需要復用，可以通過Cache算子拢切，將數(shù)據(jù)緩存到內存蒂萎。

3）輸出：程序運行結束數(shù)據(jù)會輸出Spark運行時空間，存儲到分布式存儲中（如saveAsTextFile輸出到HDFS）淮椰，或Scala數(shù)據(jù)或集合中（collect輸出到Scala集合五慈，count返回Scala int型數(shù)據(jù)）。

1.4.2 算子的分類

大致可以分為三大類算子主穗。

1）Value數(shù)據(jù)類型的Transformation算子泻拦，這種變換并不觸發(fā)提交作業(yè)，針對處理的數(shù)據(jù)項是Value型的數(shù)據(jù)忽媒。

2）Key-Value數(shù)據(jù)類型的Transfromation算子争拐，這種變換并不觸發(fā)提交作業(yè)，針對處理的數(shù)據(jù)項是Key-Value型的數(shù)據(jù)對晦雨。

3）Action算子架曹，這類算子會觸發(fā)SparkContext提交Job作業(yè)隘冲。

1.4.3 Value型的Transformation算子

這種算子可以根據(jù)RDD變換算子的輸入分區(qū)與輸出分區(qū)關系分為以下幾種類型：

1）輸入分區(qū)與輸出分區(qū)一對一型。

2）輸入分區(qū)與輸出分區(qū)多對一型绑雄。

3）輸入分區(qū)與輸出分區(qū)多對多型展辞。

4）輸出分區(qū)為輸入分區(qū)子集型。

5）還有一種特殊的輸入與輸出分區(qū)一對一的算子類型：Cache型万牺。Cache算子對RDD分區(qū)進行緩存罗珍。

1.4.3.1 輸入分區(qū)與輸出分區(qū)一對一型：

(1)Map

源碼中的map算子相當于初始化一個RDD，新RDD叫作

MappedRDD(this, sc.clean(f))脚粟。

(2)flatMap

將原來RDD中的每個元素通過函數(shù)f轉換為新的元素覆旱，并將生成的RDD的每個集合中的元素合并為一個集合。內部創(chuàng)建FlatMappedRDD(this, sc.clean(f))核无。

(3)mapPartitions

mapPartitions函數(shù)獲取到每個分區(qū)的迭代器扣唱，在函數(shù)中通過這個分區(qū)整體的迭代器對整個分區(qū)的元素進行操作。內部實現(xiàn)是生成MapPartitionsRDD厕宗。

(4)glom

glom函數(shù)將每個分區(qū)形成一個數(shù)組画舌，內部實現(xiàn)是返回的GlommedRDD。

圖3-7中的每個方框代表一個RDD分區(qū)已慢。

1.4.3.2 輸入分區(qū)與輸出分區(qū)多對一型：

(1)union

使用union函數(shù)時需要保證兩個RDD元素的數(shù)據(jù)類型相同曲聂，返回的RDD數(shù)據(jù)類型和被合并的RDD元素數(shù)據(jù)類型相同，并不進行去重操作佑惠，保存所有元素朋腋。如果想去重，可以使用distinct()膜楷。++符號相當于union函數(shù)操作旭咽。

(2)cartesian

對兩個RDD內的所有元素進行笛卡爾積操作。操作后赌厅，內部實現(xiàn)返回CartesianRDD穷绵。

1.4.3.3 輸入分區(qū)與輸出分區(qū)多對多型：

groupBy：

將元素通過函數(shù)生成相應的Key，數(shù)據(jù)就轉化為Key-Value格式特愿，之后將Key相同的元素分為一組仲墨。

實現(xiàn)：

②sc.clean( )函數(shù)將用戶函數(shù)預處理：valcleanF = sc.clean(f)

②對數(shù)據(jù)map進行函數(shù)操作，最后再對groupByKey進行分組操作揍障。this.map(t => (cleanF(t), t)).groupByKey(p)

1.4.3.4 輸出分區(qū)為輸入分區(qū)子集型：

(1)filter

filter的功能是對元素進行過濾目养，對每個元素應用f函數(shù)，返回值為true的元素在RDD中保留毒嫡，返回為false的將過濾掉癌蚁。內部實現(xiàn)相當于生成FilteredRDD(this，sc.clean(f))。

deffilter(f:T=>Boolean):RDD[T]=new FilteredRDD(this,sc.clean(f))

(2)distinct

distinct將RDD中的元素進行去重操作努释。

(3)subtract

subtract相當于進行集合的差操作碘梢，RDD 1去除RDD 1和RDD 2交集中的所有元素。

(4)sample

sample將RDD這個集合內的元素進行采樣伐蒂，獲取所有元素的子集痘系。用戶可以設定是否有放回的抽樣、百分比饿自、隨機種子，進而決定采樣方式龄坪。內部實現(xiàn)是生成

SampledRDD(withReplacement,fraction, seed)

函數(shù)參數(shù)設置如下:

withReplacement=true昭雌，表示有放回的抽樣

withReplacement=false，表示無放回的抽樣

(5)takesample

takeSample()函數(shù)和上面的sample函數(shù)是一個原理健田，但是不使用相對比例采樣烛卧，而是按設定的采樣個數(shù)進行采樣，同時返回結果不再是RDD妓局，而是相當于對采樣后的數(shù)據(jù)進行Collect()总放，返回結果的集合為單機的數(shù)組。

1.4.3.5 Cache型:

(1)cache

cache將RDD元素從磁盤緩存到內存好爬，相當于persist(MEMORY_ONLY)函數(shù)的功能局雄。

(2)persist

persist函數(shù)對RDD進行緩存操作。數(shù)據(jù)緩存在哪里由StorageLevel枚舉類型確定存炮。

persist(newLevel:StorageLevel)

part 2

2. Spark I/O機制

2.1 序列化

2.1.1 序列化的含義和目的

含義：序列化是將對象轉換為字節(jié)流炬搭，本質上可以理解為將鏈表存儲的非連續(xù)空間的數(shù)據(jù)存儲轉化為連續(xù)空間存儲的數(shù)組中。這樣就可以將數(shù)據(jù)進行流式傳輸或者塊存儲穆桂。相反宫盔，反序列化就是將字節(jié)流轉化為對象。

目的：進程間通信：不同節(jié)點之間進行數(shù)據(jù)傳輸享完。

數(shù)據(jù)持久化存儲到磁盤：本地節(jié)點將對象寫入磁盤灼芭。

2.1.2 兩種序列化方式對比

Java序列化：在默認情況下，Spark采用Java的ObjectOutputStream序列化一個對象般又。該方式適用于所有實現(xiàn)了java.io.Serializable的類彼绷。Java序列化非常靈活，但是速度較慢倒源，在某些情況下序列化的結果也比較大苛预。

Kryo序列化：Spark也能使用Kryo（版本2）序列化對象。Kryo不但速度極快笋熬，而且產(chǎn)生的結果更為緊湊（通常能提高10倍）热某。Kryo的缺點是不支持所有類型，為了更好的性能，你需要提前注冊程序中所使用的類（class）昔馋。

2.2 壓縮

當大片連續(xù)區(qū)域進行數(shù)據(jù)存儲并且存儲區(qū)域中數(shù)據(jù)重復性高的狀況下筹吐，數(shù)據(jù)適合進行壓縮。

序列化后的數(shù)據(jù)可以壓縮秘遏，使數(shù)據(jù)緊縮丘薛，減少空間開銷。

2.2.1 兩種壓縮方式對比

Snappy的目標是在合理的壓縮量的情況下邦危，提高壓縮速度洋侨，因此壓縮比并不是很高。根據(jù)數(shù)據(jù)集的不同倦蚪，壓縮比能達到20%~100%希坚。Snappy通常在達到相當壓縮的情況下，要比同類的LZO陵且、LZF裁僧、FastLZ和QuickLZ等快速的壓縮算法快，LZF提供了更高的壓縮比慕购。

2.2.2 序列化與壓縮

在分布式計算中聊疲，序列化和壓縮是兩個重要的手段。Spark通過序列化將鏈式分布的數(shù)據(jù)轉化為連續(xù)分布的數(shù)據(jù)沪悲，這樣就能夠進行分布式的進程間數(shù)據(jù)通信获洲，或者在內存進行數(shù)據(jù)壓縮等操作，提升Spark的應用性能可训。通過壓縮昌妹，能夠減少數(shù)據(jù)的內存占用，以及IO和網(wǎng)絡數(shù)據(jù)傳輸開銷握截。

2.3 Spark塊管理

整體的I/O管理分為以下兩個層次：

1）通信層：I/O模塊也是采用Master-Slave結構來實現(xiàn)通信層的架構飞崖，Master和Slave之間傳輸控制信息、狀態(tài)信息谨胞。

2）存儲層：Spark的塊數(shù)據(jù)需要存儲到內存或者磁盤固歪，有可能還需傳輸?shù)竭h端機器，這些是由存儲層完成的胯努。

2.3.1 實體與類

如圖中所示牢裳，在Storage模塊中，根據(jù)層次劃分有如下模塊：

（1）管理和接口

BlockManager：當其他模塊要和storage模塊進行交互時叶沛，storage模塊提供了統(tǒng)一的操作類BlockManager蒲讯，外部類與storage模塊打交道都需要調用BlockManager相應接口來實現(xiàn)。

（2）通信層

·BlockManagerMasterActor：從主節(jié)點創(chuàng)建灰署，從節(jié)點通過這個Actor的引用向主節(jié)點傳遞消息和狀態(tài)判帮。

·BlockManagerSlaveActor：在從節(jié)點創(chuàng)建局嘁，主節(jié)點通過這個Actor的引用向從節(jié)點傳遞命令，控制從節(jié)點的塊讀寫晦墙。

·BlockManagerMaster：對Actor通信進行管理悦昵。

（3）數(shù)據(jù)讀寫層

·DiskStore：提供Block在磁盤上以文件形式讀寫的功能邑蒋。

·MemoryStore：提供Block在內存中的Block讀寫功能函匕。

·ConnectionManager：提供本地機器和遠端節(jié)點進行網(wǎng)絡傳輸Block的功能。

·BlockManagerWorker：對遠端數(shù)據(jù)的異步傳輸進行管理觉鼻。

整體的數(shù)據(jù)存儲通信仍相當于Master-Slave模型抗楔，節(jié)點之間傳遞消息和狀態(tài)棋凳，Master節(jié)點負責總體控制，Slave節(jié)點接收命令连躏、匯報狀態(tài)贫橙。

2.3.2 讀寫流程

（1）數(shù)據(jù)寫入

1）RDD調用compute（）方法進行指定分區(qū)的寫入。

2）CacheManager中調用BlockManager判斷數(shù)據(jù)是否已經(jīng)寫入反粥，如果未寫則寫入。

3）BlockManager中數(shù)據(jù)與其他節(jié)點同步疲迂。

4）BlockManager根據(jù)存儲級別寫入指定的存儲層才顿。

5）BlockManager向主節(jié)點匯報存儲狀態(tài)

（2）數(shù)據(jù)讀取

在RDD類中，通過compute方法調用iterator讀寫某個分區(qū)（Partition）尤蒿，作為數(shù)據(jù)讀取的入口郑气。分區(qū)是邏輯概念，在物理上是一個數(shù)據(jù)塊（block）腰池。

（3）讀取邏輯

通過BlockManager讀取代碼進入讀取邏輯

1）本地讀取尾组。

在本地同步讀取數(shù)據(jù)塊，首先看能否在內存讀取數(shù)據(jù)塊示弓，如果不能讀取讳侨，則看能否從Tachyon讀取數(shù)據(jù)塊，如果仍不能讀取奏属，則看能否從磁盤讀取數(shù)據(jù)塊跨跨。

2）遠程讀取。

遠程獲取調用路徑囱皿，然后getRemote調用doGetRemote勇婴，通過BlockManagerWorker.syncGetBlock從遠程獲取數(shù)據(jù)。

其中Tachyon是一個分布式內存文件系統(tǒng)嘱腥，可以在集群里以訪問內存的速度來訪問存在tachyon里的文件耕渴。把Tachyon是架構在最底層的分布式文件存儲和上層的各種計算框架之間的一種中間件。主要職責是將那些不需要落地到DFS里的文件齿兔，落地到分布式內存文件系統(tǒng)中橱脸，來達到共享內存础米，從而提高效率。同時可以減少內存冗余慰技，GC時間等椭盏。

在BlockManagerWorker中調用syncGetBlock獲取遠端數(shù)據(jù)塊，這里使用了Future模型吻商。Future本身是一種被廣泛運用的并發(fā)設計模式掏颊，可在很大程度上簡化需要數(shù)據(jù)流同步的并發(fā)應用開發(fā)。

該模型是將異步請求和代理模式聯(lián)合的模型產(chǎn)物艾帐。

客戶端發(fā)送一個長時間的請求乌叶，服務端不需等待該數(shù)據(jù)處理完成便立即返回一個偽造的代理數(shù)據(jù)（相當于商品訂單，不是商品本身）柒爸，用戶也無需等待准浴，先去執(zhí)行其他的若干操作后，再去調用服務器已經(jīng)完成組裝的真實數(shù)據(jù)捎稚。該模型充分利用了等待的時間片段乐横。

2.3.3 數(shù)據(jù)塊讀寫管理

數(shù)據(jù)塊的讀寫，如果在本地內存存在所需數(shù)據(jù)塊今野，則先從本地內存讀取葡公，如果不存在，則看本地的磁盤是否有數(shù)據(jù)条霜，如果仍不存在催什，再看網(wǎng)絡中其他節(jié)點上是否有數(shù)據(jù)，即數(shù)據(jù)有3個類別的讀寫來源宰睡。

（1）MemoryStore內存塊讀寫

進行塊讀寫是線程間同步的蒲凶。通過entries.synchronized控制多線程并發(fā)讀寫，防止出現(xiàn)異常拆内。

PutBlock對象用來確保只有一個線程寫入數(shù)據(jù)塊旋圆。這樣確保數(shù)據(jù)讀寫且線程安全的。示例代碼如下：

Private val putLock = new Object()

內存Block塊管理是通過鏈表來實現(xiàn)的

（2）DiskStore磁盤塊寫入

在DiskStore中麸恍，一個Block對應一個文件臂聋。在diskManager中，存儲blockId和一個文件路徑映射或南。數(shù)據(jù)塊的讀寫入相當于讀寫文件流孩等。