第四章批處理系統(tǒng)SparkSpark簡介Spark由美國加州伯克利大學(xué)（UCBerkeley）的AMP實(shí)驗(yàn)室于2009年開發(fā)最初是基于內(nèi)存計算的批處理系統(tǒng)，用于構(gòu)建大型的、低延遲的數(shù)據(jù)分析應(yīng)用程序逐步發(fā)展成為內(nèi)外存同時使用的批處理系統(tǒng)，并增加了SparkStreaming支持實(shí)時流計算，以及StructuredStreaming支持批流融合2Spark簡介2013年Spark加入Apache孵化器項(xiàng)目后發(fā)展迅猛Spark在2014年打破了Hadoop保持的基準(zhǔn)排序紀(jì)錄Spark/206個節(jié)點(diǎn)/23分鐘/100TB數(shù)據(jù)Hadoop/2000個節(jié)點(diǎn)/72分鐘/100TB數(shù)據(jù)Spark用十分之一的計算資源，獲得了比Hadoop快3倍的速度3Spark軟件棧4Spark

CoreStructured

Streaming+Spark

SQL

Spark

StreamingMLlib(machinelearning)

GraphX(graph)大綱5設(shè)計思想MapReduce的局限性數(shù)據(jù)模型計算模型體系架構(gòu)工作原理容錯機(jī)制編程示例MapReduce編程范型編程容易，不需要掌握分布式并行編程細(xì)節(jié)，很容易把程序運(yùn)行在分布式系統(tǒng)上但是基礎(chǔ)算子太少：例如，如何實(shí)現(xiàn)join？ClassX{map(){//map函數(shù)的實(shí)現(xiàn)…}reduce(){//reduce函數(shù)的實(shí)現(xiàn)}main(){Jobjob=…//定義分布式作業(yè)job.config=//作業(yè)參數(shù)設(shè)置}}6單個MapReduce作業(yè)Map端的結(jié)果要先寫入本地磁盤，再由Reduce端來拉取文件A文件BMap+Reduce讀磁盤Shuffle階段Map端的結(jié)果需寫磁盤寫磁盤7多個MapReduce作業(yè)例子：迭代計算過程中每一迭代步結(jié)束時將結(jié)果寫入HDFS，下一步將該結(jié)果再次從HDFS讀出StepStepStepStepStepClient8MapReduce局限性編程框架的表達(dá)能力有限，用戶編程復(fù)雜僅map和reduce函數(shù)，無法直接用join等操作單個作業(yè)中需要Shuffle的數(shù)據(jù)以阻塞方式傳輸，磁盤IO開銷大、延遲高Shuffle數(shù)據(jù)需要先寫磁盤多個作業(yè)之間銜接涉及IO開銷，應(yīng)用程序的延遲高特別是迭代計算，迭代中間結(jié)果的反復(fù)讀寫，使得整個應(yīng)用程序的延遲非常高9大綱10設(shè)計思想MapReduce的局限性數(shù)據(jù)模型計算模型體系架構(gòu)工作原理容錯機(jī)制編程示例RDD概念11Resilient

Distributed

DatasetRDD是一個數(shù)據(jù)集(Dataset)：與MapReduce不同，Spark操作對象是抽象的數(shù)據(jù)集，而不是文件RDD是分布式的(Distributed)：每個RDD可分成多個分區(qū)，每個分區(qū)就是一個數(shù)據(jù)集片段，一個RDD的不同分區(qū)可以存到集群中不同的節(jié)點(diǎn)上RDD具有彈性(Resilient)：即具備可恢復(fù)的容錯特性大綱12設(shè)計思想MapReduce的局限性數(shù)據(jù)模型計算模型體系架構(gòu)工作原理容錯機(jī)制編程示例RDD操作算子13創(chuàng)建：從本地內(nèi)存或外部數(shù)據(jù)源創(chuàng)建RDD，提供了數(shù)據(jù)輸入的功能RDD操作算子14轉(zhuǎn)換(Transformation)：描述RDD的轉(zhuǎn)換邏輯，提供對RDD進(jìn)行變換的功能RDD操作算子15行動(Action)：標(biāo)志轉(zhuǎn)換結(jié)束，觸發(fā)DAG生成邏輯計算模型：OperatorDAG16從算子操作的角度來描述計算的過程邏輯計算模型：RDDLineage17從RDD變換的角度來描述計算過程RDDLineageRDDLineage（DAG拓?fù)浣Y(jié)構(gòu)）RDD讀入外部數(shù)據(jù)源進(jìn)行創(chuàng)建RDD經(jīng)過一系列的轉(zhuǎn)換（Transformation）操作，每一次都會產(chǎn)生不同的RDD，供給下一個轉(zhuǎn)換操作使用最后一個RDD經(jīng)過“動作”操作進(jìn)行轉(zhuǎn)換，并輸出到外部數(shù)據(jù)源Spark系統(tǒng)保留RDDLineage的信息為什么？18非嚴(yán)格情況下兩個概念可以混用RDD只讀、不可變RDD是只讀的記錄分區(qū)的集合本質(zhì)上一個只讀的對象集合RDD經(jīng)創(chuàng)建后，不能進(jìn)行修改RDD不可變（Immutable）通過在其他RDD上執(zhí)行確定的轉(zhuǎn)換操作（如map、join和groupby）而得到新的RDD，而不是改變原有的RDD遵循了函數(shù)式編程的特性變量的值是不可變的19物理計算模型：OperatorDAG20分布式架構(gòu)中，DAG中的操作算子實(shí)際上由若干個實(shí)例任務(wù)(Task)來實(shí)現(xiàn)問題：右圖需要啟動多少個Task？物理計算模型：RDDLineage21每個Task通常負(fù)責(zé)處理RDD的一個分區(qū)大綱22設(shè)計思想體系架構(gòu)架構(gòu)圖應(yīng)用程序執(zhí)行流程工作原理容錯機(jī)制編程示例抽象架構(gòu)圖23DriverClusterManagerExecutorWorkerNodeTaskTaskExecutorWorkerNodeTaskTaskClusterManager24集群管理器：負(fù)責(zé)管理整個系統(tǒng)的資源、監(jiān)控工作節(jié)點(diǎn)根據(jù)Spark部署方式的不同在Standalone方式（即不使用Yarn或Mesos等其它資源管理系統(tǒng)）中，集群管理器包含Master和Worker在Yarn方式中集群管理器包括ResourceManager和NodeManagerExecutor25執(zhí)行器：負(fù)責(zé)任務(wù)執(zhí)行Executor是運(yùn)行在工作節(jié)點(diǎn)上的一個進(jìn)程，它啟動若干個線程Task或線程組TaskSet來進(jìn)行執(zhí)行任務(wù)在Standalone部署方式下，Executor進(jìn)程的名稱為CoarseGrainedExecutorBackend問題：MapReduce中Task是線程還是進(jìn)程？Driver26驅(qū)動器：負(fù)責(zé)啟動應(yīng)用程序的主方法并管理作業(yè)運(yùn)行Spark的架構(gòu)實(shí)現(xiàn)了資源管理和作業(yè)管理兩大功能的分離ClusterManager負(fù)責(zé)集群資源管理Driver負(fù)責(zé)作業(yè)管理Standalone架構(gòu)圖27架構(gòu)圖比較28Standalone架構(gòu)圖抽象架構(gòu)圖SparkSubmit(Client)/MasterClusterManagerWorkerCoarseGrainedExecutorBackendExecutor物理節(jié)點(diǎn)（從節(jié)點(diǎn)）WorkerNode？DriverStandalone中的Driver29邏輯上，Driver獨(dú)立于主節(jié)點(diǎn)、從節(jié)點(diǎn)以及客戶端但根據(jù)應(yīng)用程序的Client或Cluster運(yùn)行方式，Driver會以不同的形式存在Client方式:Driver和客戶端以同一個進(jìn)程存在Cluster方式:系統(tǒng)將由某一Worker啟動一個進(jìn)程作為Driver客戶端提交應(yīng)用程序時可以選擇Client或Cluster方式StandaloneClient模式30Driver和客戶端以同一個進(jìn)程存在StandaloneCluster模式31某一Worker啟動一個名為DriverWrapper的進(jìn)程作為DriverSparkvs.MapReduce32二者架構(gòu)比較MapReduceSpark系統(tǒng)進(jìn)程JobTrackerMasterTaskTrackerWorkerChildCoarseGrainedExecutorBackend工作線程/Task任務(wù)代碼Task基礎(chǔ)接口map/reduceRDDAPI大綱33設(shè)計思想體系架構(gòu)架構(gòu)圖應(yīng)用程序執(zhí)行流程工作原理容錯機(jī)制編程示例抽象執(zhí)行流程34Driver(SparkContext)ExecutorWorkerNodeTaskTaskExecutorWorkerNodeTaskTask(1)(2)(3)(4)(3)(4)(5)(5)ClusterManager應(yīng)用程序執(zhí)行流程35啟動Driver，以Standalone模式為例如果使用Client部署方式，客戶端直接啟動Driver，并向Master注冊如果使用Cluster部署方式，客戶端將應(yīng)用程序提交給Master，由Master選擇一個Worker啟動Driver進(jìn)程(DriverWrapper)構(gòu)建基本運(yùn)行環(huán)境，即由Driver創(chuàng)建SparkContext，向ClusterManager進(jìn)行資源申請，并由Driver進(jìn)行任務(wù)分配和監(jiān)控應(yīng)用程序執(zhí)行流程（續(xù)）36ClusterManager通知工作節(jié)點(diǎn)啟動Executor進(jìn)程，該進(jìn)程內(nèi)部以多線程方式運(yùn)行任務(wù)Executor進(jìn)程向Driver注冊SparkContext構(gòu)建DAG并進(jìn)行任務(wù)劃分，從而交給Executor進(jìn)程中的線程來執(zhí)行任務(wù)大綱37設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例Driver內(nèi)部工作過程38ExecutorWorkerNodeTaskTaskExecutorWorkerNodeTaskTaskClusterManagerDriver(SparkContext)解析器DAGTask調(diào)度器Stage(TaskSet)DAG調(diào)度器RDD對象大綱39設(shè)計思想體系架構(gòu)工作原理Stage劃分Stage內(nèi)部數(shù)據(jù)傳輸Stage之間數(shù)據(jù)傳輸應(yīng)用與作業(yè)容錯機(jī)制編程示例RDD依賴關(guān)系窄依賴表現(xiàn)為一個父RDD的分區(qū)對應(yīng)于一個子RDD的分區(qū)或多個父RDD的分區(qū)對應(yīng)于一個子RDD的分區(qū)寬依賴則表現(xiàn)為存在一個父RDD的一個分區(qū)對應(yīng)一個子RDD的多個分區(qū)40通過依賴關(guān)系進(jìn)行Stage劃分分析各個RDD的偏序關(guān)系生成DAG，再通過分析各個RDD中的分區(qū)之間的依賴關(guān)系來決定如何劃分Stage具體劃分方法：在DAG中進(jìn)行反向解析，遇到寬依賴就斷開遇到窄依賴就把當(dāng)前的RDD加入到Stage中為什么將窄依賴盡可能劃分在同一個Stage？41基于RDDLineage的Stage劃分42寬依賴基于OperatorDAG的Stage劃分43Stage類型ShuffleMapStage輸入/輸出輸入可以是從外部獲取數(shù)據(jù)，也可以是另一個ShuffleMapStage的輸出以Shuffle為輸出，作為另一個Stage開始特點(diǎn)不是最終的Stage，在它之后還有其他Stage它的輸出一定需要經(jīng)過Shuffle過程，并作為后續(xù)Stage的輸入在一個DAG里可能有該類型的Stage，也可能沒有該類型Stage44Stage類型ResultStage輸入/輸出其輸入可以是從外部獲取數(shù)據(jù)，也可以是另一個ShuffleMapStage的輸出輸出直接產(chǎn)生結(jié)果或存儲特點(diǎn)最終的Stage在一個DAG里必定有該類型Stage因此，一個DAG含有一個或多個Stage，其中至少含有一個ResultStage45ShuffleMapStagevs.ResultStage46大綱47設(shè)計思想體系架構(gòu)工作原理Stage劃分Stage內(nèi)部數(shù)據(jù)傳輸Stage之間數(shù)據(jù)傳輸應(yīng)用與作業(yè)容錯機(jī)制編程示例Stage內(nèi)部的特點(diǎn)48所有依賴關(guān)系都是窄依賴，可以實(shí)現(xiàn)pipeline方式進(jìn)行數(shù)據(jù)傳輸流水線（Pipeline）方式49SparkPipelinevs.MapReduceShuffle50與MapReduce中Shuffle方式不同，流水線方式不要求物化前序算子的所有計算結(jié)果分區(qū)7通過map操作生成的分區(qū)9，并不需要物化分區(qū)9，而且可以不用等待分區(qū)8到分區(qū)10這個map操作的計算結(jié)束，繼續(xù)進(jìn)行union操作，得到分區(qū)13如果采用MapReduce中的Shuffle方式，那么意味著分區(qū)7、8經(jīng)map計算得到分區(qū)9、10并將這兩個分區(qū)進(jìn)行物化之后，才可以進(jìn)行unionShuffleMapTaskvs.ResultTask51大綱52設(shè)計思想體系架構(gòu)工作原理Stage劃分Stage內(nèi)部數(shù)據(jù)傳輸Stage之間數(shù)據(jù)傳輸應(yīng)用與作業(yè)容錯機(jī)制編程示例Stage之間的特點(diǎn)53所有依賴關(guān)系都是寬依賴，不可以實(shí)現(xiàn)pipeline方式進(jìn)行數(shù)據(jù)傳輸，只能ShuffleStage之間的Shuffle54Stage之間的數(shù)據(jù)傳輸需要進(jìn)行Shuffle，該過程與MapReduce中的Shuffle類似Shuffle過程可能發(fā)生在兩個ShuffleMapStage之間，或者ShuffleMapStage與ResultStage之間從Task的層面來看，該過程表現(xiàn)為兩組ShuffleMapTask之間，或一組ShuffleMapTask與一組ResultTask之間的數(shù)據(jù)傳輸SparkShuffle55ShuffleWritevs.ShuffleRead56在ShuffleWrite階段，ShuffleMapTask需要將輸出RDD的記錄按照partition函數(shù)劃分到相應(yīng)的bucket當(dāng)中并物化到本地磁盤形成ShuffleblockFile，之后才可以在ShuffleRead階段被拉取在ShuffleRead階段，ShuffleMapTask或ResultTask根據(jù)partiton函數(shù)讀取相應(yīng)的ShuffleblockFile，存入buffer并進(jìn)行繼續(xù)后續(xù)的計算Shufflevs.Pipeline57Stage之間：ShuffleStage內(nèi)部：PipelinePipelineShuffleShufflePipeline大綱58設(shè)計思想體系架構(gòu)工作原理Stage劃分Stage內(nèi)部數(shù)據(jù)傳輸Stage之間數(shù)據(jù)傳輸應(yīng)用與作業(yè)容錯機(jī)制編程示例應(yīng)用與作業(yè)Application：用戶編寫的Spark應(yīng)用程序Job：一個Job包含多個RDD及作用于相應(yīng)RDD轉(zhuǎn)換操作，其中最后一個為actionMapReducevs.SparkMapReduce中一個應(yīng)用就是一個作業(yè)Spark中的一個應(yīng)用可以由多個作業(yè)來構(gòu)成59名稱MapReduceSpark應(yīng)用JobApplication作業(yè)Job/DAG作業(yè)與任務(wù)Stage：一個Job會分為多組Task，每組Task被稱為Stage，或者也被稱為TaskSetJob的基本調(diào)度單位代表了一組關(guān)聯(lián)的、相互之間沒有Shuffle依賴關(guān)系的任務(wù)組成的任務(wù)集Task：運(yùn)行在Executor上的工作單元60應(yīng)用、作業(yè)與任務(wù)邏輯執(zhí)行角度一個Application=一個或多個DAG一個DAG=一個或多個Stage一個Stage=若干窄依賴的RDD操作物理執(zhí)行角度一個Application=一個或多個Job一個Job=一個或多個TaskSet一個TaskSet=多個沒有Shuffle關(guān)系的Task61Job=DAGStage=TaskSet邏輯概念與物理概念邏輯物理ApplicationDAGJobStageTaskSetOperatorTask62大綱63設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例故障類型Master故障：ZooKeeper配置多個MasterWorker故障Executor故障Driver故障：重啟64大綱65設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制RDD持久化故障恢復(fù)檢查點(diǎn)編程示例RDD存儲機(jī)制66由于計算過程中會不斷地產(chǎn)生新的RDD，所以系統(tǒng)不能將所有的RDD都存在內(nèi)存一旦達(dá)到相應(yīng)存儲空間的閾值，Spark會使用置換算法（例如，LRU）將部分RDD的內(nèi)存空間騰出如果不做任何聲明，這些RDD會被直接丟棄。但是，某些RDD在后續(xù)很可能會被再次使用RDD存儲機(jī)制RDD提供的持久化接口persist(StorageLevel)接受StorageLevel類型參數(shù)，可配置各種級別持久化后的RDD將會保留在工作節(jié)點(diǎn)的中，可重復(fù)使用cache()：緩存相當(dāng)于persist(MEMORY_ONLY)67StorageLevelMEMORY_ONLYMEMORY_AND_DISKMEMORY_ONLY_SERMEMORY_AND_DISK_SERDISK_ONLYMEMORY_ONLY_2,MEMORY_AND_DISK_268StorageLevelMEMORY_ONLY：在JVM中緩存Java的對象。如果內(nèi)存不足，直接丟棄某些partitionMEMORY_AND_DISK：在JVM中緩存Java的對象。如果內(nèi)存不足，則將某些partitions寫入到磁盤中MEMORY_ONLY_SER：在內(nèi)存為每個partition存儲一個byte數(shù)組，數(shù)組內(nèi)容為當(dāng)前partition中Java對象的序列化結(jié)果69StorageLevelMEMORY_AND_DISK_SER：與MEMORY_AND_DISK類似，但每個分區(qū)存儲的是Java對象序列化后組成的byte數(shù)組DISK_ONLY:將每個分區(qū)的數(shù)據(jù)序列化到磁盤中70StorageLevel71MEMORY_ONLY_2：與MEMORY_ONLY相同，但是每個分區(qū)備份到兩臺機(jī)器上MEMORY_AND_DISK_2:與MEMORY_AND_DISK相同，但是每個分區(qū)備份到兩臺機(jī)器上大綱72設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制RDD持久化故障恢復(fù)檢查點(diǎn)編程示例Lineage機(jī)制RDDdependencies73Lineage機(jī)制紅色部分丟失，需重算紫色部分74基于RDDLineage恢復(fù)利用RDDLineage的故障恢復(fù)重新計算丟失分區(qū)重算過程在不同節(jié)點(diǎn)之間可以并行與數(shù)據(jù)庫恢復(fù)的比較RDDLineage：記錄粗粒度的操作數(shù)據(jù)復(fù)制或者日志：記錄細(xì)粒度的操作75大綱76設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制RDD持久化故障恢復(fù)檢查點(diǎn)編程示例檢查點(diǎn)機(jī)制前述機(jī)制的不足之處Lineage可能非常長RDD持久化機(jī)制保存到集群內(nèi)機(jī)器的磁盤，并不完全可靠檢查點(diǎn)機(jī)制將RDD寫入外部可靠的（本身具有容錯機(jī)制）分布式文件系統(tǒng)，例如HDFS在實(shí)現(xiàn)層面，寫檢查點(diǎn)的過程是一個獨(dú)立的作業(yè)，在用戶作業(yè)結(jié)束后運(yùn)行77大綱78設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例SparkAPI79SparkCoreSparkSQLDatasetDataFrameRDDStructuredStreamingSparkStreamingDStreamMllib(DataFrame-based)GraphX(DataFrame-based)MLlib(RDD-based)GraphX(RDD-based)SQLRDDAPI編程框架80大綱81設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例詞頻統(tǒng)計關(guān)系表自然連接及其優(yōu)化網(wǎng)頁鏈接排名K均值聚類檢查點(diǎn)詞頻統(tǒng)計82輸入：一個包含大量單詞的文本文件輸出：文件中每個單詞及其出現(xiàn)次數(shù)（頻數(shù)）每個單詞和其頻數(shù)占一行，單詞和頻數(shù)之間有間隔輸入輸出AnAnMyMeAnAnMyHeMyMyAnMyAn5He1Me1My5轉(zhuǎn)換操作算子回顧83map、flatMap、groupByKey“Hadoopisgood”“Sparkisfast”“Sparkisbetter”{“Hadoop”,“is”,“good”}{“Spark”,“is”,“fast”}{“Spark”,“is”,“better”}map(line=>line.split(“”))“Hadoopisgood”“Sparkisfast”“Sparkisbetter”“Hadoop”“is”“good”“Spark”“is”“fast”“Spark”“is”“better”flatMap(line=>line.split(“”))[“Hadoop”,1][“is”,1][“good”,1][“Spark”,1][“is”,1][“fast”,1][“Spark”,1][“is”,1][“better”,1]groupByKey()[“Hadoop”,1][“is”,{1,1,1}][“good”,1][“Spark”,{1,1}][“fast”,1][“better”,1]解決方案84flatMap將每行文本按分隔符拆分成單詞map將每個單詞映射為[單詞,1]鍵值對groupByKey將[單詞,1]鍵值對按單詞進(jìn)行分組map對組內(nèi)每個單詞的頻數(shù)進(jìn)行求和RDDLineage85Spark提供比MapReduce更豐富的算子，更易于編程。代碼86優(yōu)化方案87減少Shuffle數(shù)據(jù)量(a)優(yōu)化前(b)優(yōu)化后優(yōu)化方案的代碼88MapReduce要求用戶自定義combine方法，而Spark內(nèi)置combine機(jī)制，簡化了用戶編程。大綱89設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例詞頻統(tǒng)計關(guān)系表自然連接及其優(yōu)化網(wǎng)頁鏈接排名K均值聚類檢查點(diǎn)關(guān)系表的自然連接90輸入：兩個CSV文件，分別保存雇員表和部門表輸出：雇員表和部門表的自然連接結(jié)果雇員表部門表雇員表?部門表轉(zhuǎn)換操作算子回顧91cogroupcogroup[1,“Spark”][2,“MapReduce”][1,10][1,20][2,30][2,40][1,{{“Spark”},{10,20}}][2,{{“MapReduce”},{30,40}}]解決方案92map將來自部門表的每個元組映射為[DeptName,

Manager]鍵值對將來自雇員表的每個元組映射為[DeptName,

Name

EmpId]鍵值對cogroup將雇員表和部門表按DeptName進(jìn)行協(xié)同分組flatMap對組內(nèi)具有相同DeptName的元組兩兩組合，得到自然連接結(jié)果RDDLineage93MapReduce要求用戶顯示標(biāo)記每個元組來自哪張表，而Spark可由2個算子分別處理2張關(guān)系表得到不同的RDD。代碼94

95假如部門表比較小，雇員表非常大雇員表的Shuffle開銷非常大優(yōu)化方案將“小表”廣播出去，避免“大表”進(jìn)行Shuffle小表大表小表大表(a)Shuffle(b)廣播優(yōu)化方案的代碼96大綱97設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例詞頻統(tǒng)計關(guān)系表自然連接及其優(yōu)化網(wǎng)頁鏈接排名K均值聚類檢查點(diǎn)網(wǎng)頁鏈接排名98輸入：保存在文本文件中，一行為一項(xiàng)網(wǎng)頁信息網(wǎng)頁信息：(網(wǎng)頁名網(wǎng)頁排名值(出站鏈接出站鏈接的權(quán)重...))輸出：網(wǎng)頁名及其排名值輸入輸出A1.0B1.0D1.0B1.0C1.0C1.0A1.0B1.0D1.0B1.0C1.0A0.21436B0.36332C0.40833D0.13027轉(zhuǎn)換操作算子回顧99join、mapValuesjoin[1,“Spark”][2,“MapReduce”][1,10][1,20][2,30][2,40][1,{“Spark”,10}][1,{“Spark”,20}][2,{“MapReduce”,30}][2,{“MapReduce”,40}][1,“Spark”][2,“MapReduce”][1,“Spark2.x”][2,“MapReduce2.x”]mapValues(_+“2.x”)解決方案100

整個迭代是一個Spark應(yīng)用join、flatMapreduceByKey、mapValuesRDDLineage101MapReduce迭代間數(shù)據(jù)的傳遞需要借助讀寫HDFS來完成，而Spark迭代間的RDD數(shù)據(jù)可以直接駐留在內(nèi)存中。代碼102迭代比較103MapReduce每一迭代步驟(Step)結(jié)束時將結(jié)果寫入HDFS，下一迭代步驟將該結(jié)果再次從HDFS讀出。迭代間反復(fù)讀寫HDFS開銷大。Spark每一迭代步驟(Step)結(jié)束時得到的RDD可以駐留在內(nèi)存中直接作為下一迭代步驟的輸入。避免了迭代間反復(fù)讀寫磁盤。大綱104設(shè)計思想體系架構(gòu)工作原理容錯機(jī)制編程示例詞頻統(tǒng)計關(guān)系表自然連接及其優(yōu)化網(wǎng)頁鏈接排名K均值聚類檢查點(diǎn)K均值聚類105輸入：兩個文本文件，分別保存數(shù)據(jù)集和聚類中心集數(shù)據(jù)集：每行為一個二維數(shù)據(jù)點(diǎn)及其類別標(biāo)簽聚類中心集：每行為一個二維數(shù)據(jù)點(diǎn)輸出：數(shù)據(jù)點(diǎn)及其類別標(biāo)簽數(shù)據(jù)集聚類中心集聚類結(jié)果0,0-11,2-13,1-18,8-19,10-110,7-11,23,10,01.01,21.010,72.03,11.08,82.09,102.0Na?ve解決方案106算法執(zhí)行過程設(shè)定聚類中心數(shù)k。

例如，k=2選取聚類中心尋找每個數(shù)據(jù)點(diǎn)距