課程資源云計(jì)算ccspark

南京大學(xué)軟件學(xué)院李傳藝

《云計(jì)算》——Spark簡介目錄簡介概覽、功能、集群管理、存儲(chǔ)、生態(tài)系統(tǒng)本課程的實(shí)踐目標(biāo)系統(tǒng)搭建原理理解運(yùn)行模式運(yùn)行過程RDD資源調(diào)度項(xiàng)目實(shí)踐Spark

StreamingSparkGraphXSparkMLlib課程作業(yè)簡介(1)——概覽官網(wǎng)：ApacheSpark?

isafastandgeneralengineforlarge-scaledataprocessing.快基于內(nèi)存的MapReduce計(jì)算比Hadoop快100x倍，基于硬盤的則快10x倍易用支持Scala、Java、Python和R語言開發(fā)功能強(qiáng)SparkSQL、SparkStreaming、SparkGraphX、SparkMLlib更加通用適用于多種不同的集群管理框架Standaloneclustermode、ApacheMesos、HadoopYARN、intheCloud（EC2）適用于多種不同的數(shù)據(jù)存儲(chǔ)方式：數(shù)據(jù)讀取接口HDFS、HBase、MongoDB、Cassendra簡介(2)——功能SparkSQL前身是Shark，基于Hive的SparkSQL，代碼量大、復(fù)雜，難優(yōu)化和維護(hù)交互式查詢、標(biāo)準(zhǔn)訪問接口、兼容Hive專門用于處理結(jié)構(gòu)化數(shù)據(jù)：分布式SQL引擎；在Spark程序中調(diào)用APISparkStreaming實(shí)時(shí)對(duì)大量數(shù)據(jù)進(jìn)行快速處理，處理周期短SparkGraphX以圖為基礎(chǔ)數(shù)據(jù)結(jié)構(gòu)的算法實(shí)現(xiàn)和相關(guān)應(yīng)用SparkMLlib為解決機(jī)器學(xué)習(xí)開發(fā)的庫，包括分類、回歸、聚類和協(xié)同過濾等簡介(3)——集群管理框架Standalonemode原生集群管理功能：任務(wù)調(diào)度、資源分配等HadoopYarnHadoopMapReduce的第二個(gè)版本架構(gòu)，把資源管理和任務(wù)管理剝離開；實(shí)現(xiàn)了靜態(tài)資源分配和動(dòng)態(tài)資源分配功能；ApacheMesos從分布式計(jì)算節(jié)點(diǎn)上抽象CPU,memory,storage,andothercomputeresources給其他框架使用，實(shí)現(xiàn)了靜態(tài)資源分配功能EC2AmazonEC2云平臺(tái)，提供一個(gè)安裝了Spark、Shark和HDFS的集群，可直接登錄到集群，把它當(dāng)作你實(shí)驗(yàn)室的集群使用簡介(4)——集群管理框架圖簡介(4)——集群管理框架圖簡介(5)——存儲(chǔ)方式HDFSHadoop分布式文件系統(tǒng)Hbase基于HDFS的非關(guān)系型數(shù)據(jù)庫(NoSQL數(shù)據(jù)庫)MongoDB基于分布式存儲(chǔ)的數(shù)據(jù)庫，介于關(guān)系型和非關(guān)系型數(shù)據(jù)庫之間是NoSQL數(shù)據(jù)庫中最像關(guān)系型數(shù)據(jù)庫的一個(gè)功能非常強(qiáng)大：支持多種開發(fā)語言、支持完全索引、支持查詢等Cassendra基于列的分布式數(shù)據(jù)庫，易擴(kuò)展、模式靈活、按照范圍查詢等它們和Spark的關(guān)系：Spark是一個(gè)計(jì)算程序，需要讀取數(shù)據(jù)和存儲(chǔ)數(shù)據(jù)，它們就是數(shù)據(jù)存儲(chǔ)的地方簡介(6)——Spark生態(tài)系統(tǒng)簡介(7)——從MapReduce說起MapReduceGoogle提出的一種編程范式；最早起源于函數(shù)式編程范式；Lambda演算；聲明式編程vs.命令式編程（非結(jié)構(gòu)化、結(jié)構(gòu)化（過程式、面向?qū)ο螅㎝apReduce范式的重點(diǎn)是編寫兩個(gè)函數(shù)map():過濾和采集數(shù)據(jù)生成key-value對(duì)reduce():根據(jù)map生成的鍵完成規(guī)約、分組和總結(jié)組合key-value對(duì)進(jìn)行計(jì)算什么時(shí)候使用MapReduce范式——什么時(shí)候不適用一個(gè)值的計(jì)算依賴前一個(gè)值數(shù)據(jù)集合小需要用同步來處理共享數(shù)據(jù)一個(gè)操作依賴其它操作基本計(jì)算是處理器敏感操作為什么使用MapReduce——實(shí)現(xiàn)橫向擴(kuò)容編程模型+基礎(chǔ)架構(gòu)能夠編寫在上千臺(tái)機(jī)器上運(yùn)行的程序自動(dòng)并行和分布容錯(cuò)性強(qiáng)調(diào)度、檢查、監(jiān)控簡介(8)——從Hadoop到Spark的實(shí)現(xiàn)Hadoop開源的MapReduce應(yīng)用實(shí)現(xiàn)將工作簡化到編寫map()和reduce()函數(shù)是一個(gè)純粹的MapReduce框架map()combine()reduce()適合單趟計(jì)算：先map再reduce；如果多趟則會(huì)比較慢Spark不是一個(gè)專門的MapReduce框架，但是支持MapReduce框架的功能不限于先map再reduce：任意的DAG(后面會(huì)講到)

vs.Hadoop是有向的一個(gè)DAG可以包含多個(gè)階段vs.Hadoop單趟“不共享”范式，but共享不可變數(shù)據(jù)結(jié)構(gòu)（HadoopConfiguration傳遞）Spark還支持只寫累加器總結(jié)：Hadoop和Spark好處為大數(shù)據(jù)處理提供可讀性容錯(cuò)：任何節(jié)點(diǎn)宕機(jī)不會(huì)丟失所需計(jì)算結(jié)果可伸縮性支持龐大的服務(wù)器集群分布式處理輸入數(shù)據(jù)和處理是分布式的，可以全面支持大數(shù)據(jù)并行化可以在節(jié)點(diǎn)集群上并行地執(zhí)行應(yīng)用查詢?nèi)罩咎幚恚蛔ト?、索引和搜索；分析、文本處理、情感分析機(jī)器學(xué)習(xí)；推薦系統(tǒng)；文檔聚類和分類；生物信息學(xué)、基因組分析……目錄簡介概覽、功能、集群管理、存儲(chǔ)、生態(tài)系統(tǒng)本課程實(shí)踐目標(biāo)系統(tǒng)搭建原理理解運(yùn)行模式運(yùn)行過程RDD資源調(diào)度項(xiàng)目實(shí)踐Spark

StreamingSparkGraphXSparkMLlib課程作業(yè)本課程的實(shí)踐目標(biāo)SparkStandalone+HDFS+MongoDB1.能夠搭建一個(gè)左圖混合系統(tǒng)；2.利用Spark進(jìn)行數(shù)據(jù)處理：2.1理解Spark原理2.2使用Spark2.2.1結(jié)合HDFS和MongoDB讀取和存儲(chǔ)數(shù)據(jù)2.2.2使用SparkStreaming2.2.3使用SparkGraphX2.2.4使用SparkMLlib目標(biāo)1：系統(tǒng)搭建安裝HDFS下載Hadoop：配置Hadoop集群：HDFS自然就可以使用了安裝Spark下載Spark：配置SparkStandalone集群：安裝MongoDB下載MongoDB：(頁面有安裝指南)下載MongoSparkConnector

目標(biāo)1：補(bǔ)充步驟確定物理機(jī)并安裝操作系統(tǒng)如果沒有網(wǎng)絡(luò)要事先下載所有安裝包設(shè)置SSH無密碼登陸安裝JDK、Scala助教——目前沒有目標(biāo)2：利用Spark處理數(shù)據(jù)理解Spark原理開發(fā)Spark程序：開發(fā)環(huán)境、程序提交、運(yùn)行模式內(nèi)核講解：RDD工作機(jī)制：任務(wù)調(diào)度、資源分配使用SparkSpark讀取、存儲(chǔ)HDFS、MongoDBSparkStreamingSparkGraphXSparkMLlib開發(fā)環(huán)境搭建——IDEAScala語言

下載安裝IDEA安裝JDK安裝Scala插件實(shí)戰(zhàn)例子Java語言

Python

到Spark、IDEA官方網(wǎng)站找資料Spark原理(1)——Spark提交應(yīng)用程序main方法所在的類的路徑集群管理器的地址應(yīng)用程序部署的模式，是交互式還是集群式Spark的配置參數(shù)及值應(yīng)用程序及所有依賴的包所在位置Spark原理(2)——Driver程序應(yīng)用執(zhí)行過程Spark應(yīng)用的入口程序：Driver在集群模式下用戶開發(fā)的Spark程序稱為Driver應(yīng)用的執(zhí)行位置Local集群應(yīng)用的部署模式集群模式：如果在遠(yuǎn)離（網(wǎng)絡(luò)遠(yuǎn)離）計(jì)算節(jié)點(diǎn)的本地機(jī)器上提交Spark程序到集群內(nèi)部執(zhí)行，稱為集群模式，當(dāng)程序提交后，本地機(jī)器就結(jié)束了和集群的交互，程序的運(yùn)行結(jié)果也不會(huì)返回到本地，只保存在集群內(nèi)部?？蛻舳四Ｊ剑喝绻诳拷?jì)算節(jié)點(diǎn)的機(jī)器上執(zhí)行（Master或者就是Worker）程序，可以選擇客戶端模式，就是通過Spark提供的Shell執(zhí)行，程序會(huì)和集群產(chǎn)生持續(xù)的通信，集群的任務(wù)執(zhí)行完成后，會(huì)把執(zhí)行結(jié)果返回到客戶端。Spark原理(3)——SparkContextSparkContext對(duì)象使得應(yīng)用程序能夠和集群進(jìn)行溝通，實(shí)現(xiàn)CPU、內(nèi)存等資源的分配每個(gè)Driver程序都有一個(gè)SC對(duì)象如果是交互式編程，則Spark

Shell會(huì)自動(dòng)創(chuàng)建一個(gè)SC對(duì)象程序啟動(dòng)后，SC會(huì)告訴集群管理器在WorkerNode上創(chuàng)建執(zhí)行器執(zhí)行器是每一個(gè)Spark程序在計(jì)算節(jié)點(diǎn)上專屬的進(jìn)程程序代碼會(huì)發(fā)送到對(duì)應(yīng)的Worker

Node上SC分發(fā)任務(wù)(Task)到各個(gè)執(zhí)行器Spark原理(4)——其它相關(guān)名詞Application獨(dú)立的Spark程序Master（Cluster

Manager），Worker

執(zhí)行器（Executor）Spark程序在對(duì)應(yīng)計(jì)算節(jié)點(diǎn)上啟動(dòng)的專屬線程，負(fù)責(zé)執(zhí)行TaskJob一次RDD

Action稱為一個(gè)Job，是一個(gè)概念Stage介于Job和Task之間，是一個(gè)Task集合Task在執(zhí)行器上執(zhí)行的最小單元，例如在某個(gè)計(jì)算節(jié)點(diǎn)對(duì)一個(gè)RDD的分區(qū)進(jìn)行的Transformation操作Spark原理(5)——具體執(zhí)行過程分為兩部分Driver程序客戶端模式下，Driver在本地，不會(huì)把Driver提交給Master，而直接提交任務(wù)集群模式下，首先Driver被提交給Master，Master首先找一個(gè)Worker運(yùn)行Driver集群節(jié)點(diǎn)RDDObjects構(gòu)建DAG圖DAG

SchedulerDAG劃分StageTask

Scheduler分派和監(jiān)控TaskSparkContextDAG圖TaskSetMaster分配并監(jiān)控資源Workers啟動(dòng)執(zhí)行器匯報(bào)資源情況Cluster1.注冊(cè)并申請(qǐng)資源2.分配Worker創(chuàng)建執(zhí)行器3.執(zhí)行器發(fā)送心跳3.執(zhí)行器注冊(cè)并發(fā)送心跳申請(qǐng)任務(wù)4.分配任務(wù)并發(fā)送代碼5.任務(wù)結(jié)束注銷執(zhí)行器啟動(dòng)后運(yùn)行Driver程序的節(jié)點(diǎn)不需要和管理器有太多交互，而與worker間交互密切，因此Driver程序應(yīng)該放在距離Worker節(jié)點(diǎn)較近的地方。RDDDAG

SchedulerTaskSchedulerMaster分配資源詳細(xì)的應(yīng)用執(zhí)行過程展示

結(jié)合源代碼分析的Spark執(zhí)行過程解讀

Spark原理(6)——RDD(1)Job、Task、DAGScheduler的定義均依賴RDDResilientDistributedDataset彈性分布式數(shù)據(jù)集是Spark的核心數(shù)據(jù)結(jié)構(gòu)是一個(gè)數(shù)據(jù)集，就像Array、List和Set等一樣的數(shù)據(jù)結(jié)構(gòu)內(nèi)容是平鋪的，可以順序遍歷，像Array和List一樣RDD是分布式存儲(chǔ)的，支持并行計(jì)算RDD的分布是彈性的，某些操作使不同的數(shù)據(jù)塊重新匯聚和分布RDD是只讀的RDD可以緩存在內(nèi)存中RDD可以通過重復(fù)計(jì)算獲得（實(shí)現(xiàn)高可靠性）Spark原理(6)——RDD(2)RDD定義：包含5個(gè)屬性分區(qū)列表：記錄了數(shù)據(jù)塊所在的分區(qū)位置；一個(gè)RDD對(duì)應(yīng)的數(shù)據(jù)是切分為數(shù)據(jù)塊存儲(chǔ)在集群的不同節(jié)點(diǎn)上的@transientdefgetPartitions_:Array[Partition]=null依賴列表：記錄了當(dāng)前這個(gè)RDD依賴于哪些其它的RDDprivatevardependencies_:Seq[Dependency[_]]=null計(jì)算函數(shù)compute，用于計(jì)算RDD各個(gè)分區(qū)的值defcompute(split:Partition,context:TaskContext):Iterator[T]可選：分區(qū)器，子類可以重新指定新的分區(qū)方式：Hash和Range@transientvalpartitioner:Option[Partitioner]=None可選：計(jì)算各分區(qū)時(shí)優(yōu)先的位置列表例如從HDFS文件生成RDD時(shí)，HDFS文件所在位置就是對(duì)應(yīng)生成的RDD分區(qū)所在位置的優(yōu)先選擇protecteddefgetPreferredLocations(split:Partitiion):Seq[String]=NilSpark原理(6)——RDD(3)RDD的種類：繼承自基礎(chǔ)的RDD類HadoopRDD，newAPIHadoopRDDMapperedRDDFlatMappedRDDFilteredRDDPairedRDD從外部數(shù)據(jù)讀入并初始化為RDD文件系統(tǒng)：單個(gè)文件（textFile）、文件目錄（wholeTextFiles）HadoopInputformat：不同的讀取函數(shù)返回HadoopRDD從Driver數(shù)據(jù)集生成RDDSparkContext的parallelized函數(shù)對(duì)RDD的一些操作后產(chǎn)生新的RDD從程序中已經(jīng)存在的數(shù)據(jù)對(duì)象生成RDDSpark原理(6)——RDD(4)對(duì)RDD的操作Transformation接收一個(gè)RDD作為輸入，返回一個(gè)新的RDD但是并不會(huì)真實(shí)執(zhí)行計(jì)算，只是定義生成的新RDD并且設(shè)置其與前一個(gè)RDD的依賴關(guān)系操作對(duì)RDD依賴關(guān)系的區(qū)分窄依賴：只依賴前一個(gè)RDD的確定的幾個(gè)分區(qū)寬依賴：依賴前一個(gè)RDD的所有分區(qū)Action輸入還是RDD，但是輸出就不是RDD了調(diào)用了Action之后，整個(gè)從輸入RDD到輸出值的Transformation鏈條會(huì)被執(zhí)行Spark原理(6)——RDD(5)TransformationTransformation操作說明map(func)對(duì)源RDD中的每個(gè)元素調(diào)用func，生產(chǎn)新的元素，這些元素構(gòu)成新的RDD并返回flatMap(func)與map類似，但是func的返回是多個(gè)成員filter(func)對(duì)RDD成員進(jìn)行過濾，成員調(diào)用func返回True的才保留，保留的構(gòu)成新RDD返回mapPartitions(func)和map類似，但是func作用于整個(gè)分區(qū)，類型是Iterator<T>=>Iterator<T>mapPartitionsWithIndex(func)…union(otherDataset)…reduceByKey(func,[numTasks])對(duì)<key,value>結(jié)構(gòu)的RDD聚合，相同key的value調(diào)用reduce,func是(v,v)=>vjoin(otherDataset,[numTasks])…Spark原理(6)——RDD(6)Transformation操作對(duì)應(yīng)的依賴關(guān)系（例舉）Spark原理(6)——RDD(5)ActionAction操作說明reduce(func)對(duì)RDD成員使用func進(jìn)行reduce操作，func接收兩個(gè)值只返回一個(gè)值；reduce只有一個(gè)返回值。func會(huì)并發(fā)執(zhí)行collect()將RDD讀取到Driver程序里面，類型是Array，要求RDD不能太大count()返回RDD成員數(shù)量first()返回RDD第一個(gè)成員，等價(jià)于take(1)take(n)…saveAsTextFile(path)把RDD轉(zhuǎn)換成文本內(nèi)容并保存到指定的path路徑下，可能有多個(gè)文件；path可以是本地文件系統(tǒng)路徑，也可以是HDFS路徑，轉(zhuǎn)換方法是RDD成員的toString方法saveAsSequenceFile(path)…foreach(func)對(duì)RDD的每個(gè)成員執(zhí)行func方法，沒有返回值SparkMap-Reduce實(shí)例objectMyFunctions{ deflineSplit(line:String):Array[String]={ line.split(“”) }}valtextFile=sc.textFile(“README.md”)valwords=textFile.flatMap(MyFunctions.lineSplit)#先func再mapvalwordPairs=words.map(word=>(word,1))#先func再mapvalwordCounts=wordPairs.reduceByKey((a,b)=>a+b)wordCounts.collect()Spark原理(7)——DAGScheduler(1)把一個(gè)SparkJob轉(zhuǎn)換成Stage的DAG（DirectedAcyclicGraph有向無環(huán)圖）根據(jù)RDD和Stage之間的關(guān)系找出開銷最小的調(diào)度方法，然后把Stage以TaskSet的形式提交給TaskSchedulerRDDObjects構(gòu)建DAG圖DAG

SchedulerDAG劃分StageTask

Scheduler分派和監(jiān)控TaskSparkContextDAG圖TaskSetMaster分配并監(jiān)控資源Workers啟動(dòng)執(zhí)行器匯報(bào)資源情況Cluster1.注冊(cè)并申請(qǐng)資源2.分配Worker創(chuàng)建執(zhí)行器3.執(zhí)行器發(fā)送心跳3.執(zhí)行器注冊(cè)并發(fā)送心跳申請(qǐng)任務(wù)4.分配任務(wù)并發(fā)送代碼5.任務(wù)結(jié)束注銷Job：一次Action的所有工作就是一個(gè)Job如何劃分Stage？以寬依賴為分界寬依賴之前的所有Transformation為一個(gè)Stage數(shù)據(jù)是否需要重組一個(gè)Stage包括一個(gè)或多個(gè)Transformation一個(gè)Transformation在一個(gè)RDD分區(qū)上執(zhí)行是一個(gè)TaskSpark原理(7)——DAGScheduler(2)從Action開始構(gòu)造Stage

DAG最后的Action構(gòu)造一個(gè)ResultStage填補(bǔ)Stage需要的輸入RDD，并根據(jù)對(duì)RDD的依賴關(guān)系劃分Stage直到所有的RDD依賴關(guān)系都補(bǔ)充完整開始按照DAG計(jì)算需要的RDD將每一個(gè)Stage封裝程