講稿4編程模型上

1、Spark大數(shù)據(jù)平臺主講人：楊2.4、Spark編程模型(上)概要01從Hadoop MR到Spark02Spark編程模型03初識RDD從HadoopMR到Spark回顧hadoopmapreduce計算過程MR VS SparkClientLoop outside the systemStepStepStepMR慢的，序列化和磁盤IO開銷。：頻繁寫磁盤導(dǎo)致額外的Loop outside the systemClientSpark為何快：內(nèi)存計算+DAG(delay scheduling)StepStepStepStepStepStepStepSpark編程模型概念注意：對比mr里的概念來學(xué)

2、習(xí)術(shù)語解釋Application基于Spark的用戶程序，包含了driver程序和集群上運行的executorDriver Program運main函數(shù)并且新建SparkContext的程序Executor是在一個worker node上為某應(yīng)用啟動的一個進程，該進程負責(zé)運行任務(wù)，并且負責(zé)將數(shù)據(jù)存在內(nèi)存或者磁盤上。每個應(yīng)用都有各自的executors（多個）Cluster Mmanager在集群上獲取的外部服務(wù)(例如:standalone,Mesos,Yarn )Worker Node集群中任何可以運行應(yīng)用代碼的節(jié)點Task被送到某個executor上的工作單元Job包含很多任務(wù)的并行計算，可

3、以看做和Spark的action對應(yīng)Stage一個Job會被拆分很多組任務(wù)，每組任務(wù)被稱為Stage(就像Mapreduce分map任務(wù)和reduce任務(wù)一樣)Spark Application的組成Spark應(yīng)用程序的組成 Driver Executor注意：對照helloworld來思考Spark Application基本概念.Stage1StageNtask1.taskNtask1.taskNjob1DriverProgramStage1StageNtask1.taskNtask1.taskNApplicationjobNSpark Application編程模型Spark 應(yīng)用程序編

4、程模型 Driver Program （ SparkContext ） Executor （ RDD 操作） 輸入Base-> RDD Transformation RDD->RDD Action RDD->driver or Base 共享變量 broadcast variables(廣播變量) accumulators(累加器)回顧Spark Hello World部分就是對RDD的操作創(chuàng)建RDD轉(zhuǎn)換操作action初識RDD什么是RDD定義:在集群背后，有一個非常重要的分布式數(shù)據(jù)架構(gòu)，即彈性分布式數(shù)據(jù)集（resilientdistributed dataset，RDD）

5、，它是邏輯集中的實體，在集群中的多臺上進行了數(shù)據(jù)分區(qū)。 RDD是Spark的形成整個Spark程序。數(shù)據(jù)結(jié)構(gòu)，通過RDD的依賴關(guān)系形成Spark的調(diào)度順序。 通過對RDD的操作RDD 是只讀的。每次操作會轉(zhuǎn)換成一個新的RDD，之前的RDD保持不變。RDD 是分區(qū)的集合。每個RDD包含多個分區(qū)partition，實現(xiàn)并行計算。比如每個map操作一個partion，可以有多個map并行操作。RDD 是容錯的。RDD實現(xiàn)了基于Lineage（血統(tǒng)）的容錯機制。RDD的轉(zhuǎn)換關(guān)系，了compute chain，可以把這個compute chain認為是RDD之間演化的Lineage。在部分計算結(jié)果丟失的

6、時候，只需要根據(jù)這個Lineage重算即可。RDD 是高效的。RDD 不需要物化。-物化：進行實際的變換并最終寫入穩(wěn)定的器上RDD 可以緩存的。-可指定緩存級別（可以緩存內(nèi)存或者緩存磁盤）RDD是spark的，也是整個spark的架構(gòu)基礎(chǔ)，RDD是彈性分布式集合（ResilientDistributed Datasets）的簡稱，是分布式只讀且已分區(qū)集合對象。這些集合是彈性的，如果數(shù)據(jù)集一部分丟失，則可以對它們進行重建。RDD接口接口/屬性說明partitions分區(qū)，一個rdd有一個或多個分區(qū)組成，是數(shù)據(jù)的邏輯分區(qū)preferredLocations(p)對于分區(qū)p來說，返回數(shù)據(jù)本地化計算的

7、節(jié)點dependencies()所依賴的父RDD的關(guān)系computer(p,context)對分區(qū)p來說，進行具體的迭代計算partitioner()Rdd 的分區(qū)函數(shù)：hash rangeRDD的本質(zhì)特征RDD-partitionsSpark中將1100的數(shù)組轉(zhuǎn)換為rdd通過第15行的size獲得rdd的partition的個數(shù)， 此處創(chuàng)建rdd顯式指定定分區(qū)個數(shù)2，默認數(shù)值是這個程序所分配到的的cpu核的個數(shù)RDD-preferredLocations返回此RDD的一個partition的數(shù)據(jù)塊信息，如果一個數(shù)據(jù)塊（block）有多個備份再返回所有備份的location地址信息返回地址：主

8、機ip或作用：spark在進行任務(wù)調(diào)度室盡可能根據(jù)block的地址做到本地計算RDD-dependenciesRDD之間的依賴關(guān)系分為兩類：窄依賴每個父RDD的分區(qū)都至多被一個子RDD的分區(qū)使用，即為OneToOneDependecies； 寬依賴多個子RDD的分區(qū)依賴一個父RDD的分區(qū)，即為ShuffleDependency 。例如，map操作是一種窄依賴，而join操作是一種寬依賴（除非父RDD已經(jīng)基于Hash策略被劃分過了，co-partitioned）RDD-dependencies相比于寬依賴，窄依賴對優(yōu)化很有利 ，主要基于以下兩點：第一點寬依賴往往對應(yīng)著shuffle操作，需要在運

9、行過程中將同一個父RDD的分區(qū)傳入到不同的子RDD 分區(qū)中，中間可能涉及多個節(jié)點之間的數(shù)據(jù)傳輸；而窄依賴的每個父RDD的分區(qū)只會傳入到一個子RDD分區(qū)中，通常可以在一個節(jié)點內(nèi)完成轉(zhuǎn)換。RDD-dependencies相比于寬依賴，窄依賴對優(yōu)化很有利 ，主要基于以下兩點：第二點（1） 當RDD分區(qū)丟失時（某個節(jié)點故障），spark會對數(shù)據(jù)進行重算。（2） 對于窄依賴，由于父RDD的一個分區(qū)只對應(yīng)一個子RDD分區(qū)，這樣只需要重算和子RDD分區(qū)對應(yīng)的父RDD分區(qū)即可，所以這個重算對數(shù)據(jù)的利用率是100%的；（3） 對于寬依賴，重算的父RDD分區(qū)對應(yīng)多個子RDD分區(qū)，這樣實際上父RDD 中只有一部分的

10、數(shù)據(jù)是被用于恢復(fù)這個丟失的子RDD分區(qū)的，另一部分對應(yīng)子RDD的其它未丟失分區(qū)，這就造成了多余的計算；另外，寬依賴中子RDD分區(qū)通常來自多個父RDD分區(qū)，。情況下，所有的父RDD分區(qū)都要進行重新計算（4）如下圖所示，b1分區(qū)丟失，則需要重新計算a1,a2和a3，這就產(chǎn)生了冗余計算(a1,a2,a3中對應(yīng)b2的數(shù)據(jù))。RDD-dependencies窄依賴相比寬依賴更高效消耗更少1.在單個集群節(jié)點上流水線式執(zhí)行，這個節(jié)點可以計算所有父級分區(qū)。例如，可以逐個元素地依次執(zhí)行filter操作和map操作（多步操作在同一個節(jié)點可以執(zhí)行， 否則第一步和第二步不在一個節(jié)點操作）。而寬依賴則需要首先計算好所有

11、父分區(qū)數(shù)據(jù)，然后在節(jié)點之間進行Shuffle，這與2.MapReduce類似（map都計算完畢，才計算reduce）。在窄依賴中，節(jié)點失敗后的恢復(fù)更加高效。1.2.因為只有丟失的父級分區(qū)需要重新計算，并且這些丟失的父級分區(qū)可以并行地在不同節(jié)點上重新計算。而對于一個寬依賴關(guān)系的Lineage圖，單個節(jié)點失效導(dǎo)致這個RDD丟失部分分區(qū)， 可能所有祖先RDD因此需要整體重新計算。pute分區(qū)計算Spark對RDD的計算是以partition為最小要保存每次的計算結(jié)果的，并且都是對迭代器進行復(fù)合，不需RDD- partitioner分區(qū)函數(shù)：目前spark中提供兩種分區(qū)函數(shù)：HashPatitione

12、r（哈希分區(qū)）RangePatitioner（區(qū)域分區(qū)）且partitioner只存在于（K，V）類型的RDD中，rdd本身決定了分區(qū)的數(shù)量。RDD- lineageval lines = sc.textFile("hdfs:/.")/ transformed RDDsval errors = lines.filter(_.startsWith("ERROR") val messages = errors.map(_.split("t").map(r => r(1)messages.cache()/ action 1 messages.filter(_.contains("mysql").count()/ action 2 messages.filter(_.contains("php").count()RDD- lineageRDD11、RDD經(jīng)過trans或action后產(chǎn)生一個新的RDD，2、RDD之間的通過lineage來表達依賴關(guān)系，lineage 是rdd容錯的重要機