大數(shù)據(jù)基礎(chǔ)-Spark內(nèi)存計算框架

第八章Spark分布式內(nèi)存計算框架科技大學(xué)軟件學(xué)院目錄2Spark簡介Spark地編程模型Spark地調(diào)度機(jī)制Spark生態(tài)圈其它技術(shù)Zeppelin:互式分析Spark數(shù)據(jù)Spark簡介3Spark是一種基于內(nèi)存地,用以實現(xiàn)高效集群計算地臺。準(zhǔn)確地講,Spark是一個大數(shù)據(jù)并行計算框架,是對廣泛使用地MapReduce計算模型地擴(kuò)展。Spark簡介4促ApacheSpark迅速成長為大數(shù)據(jù)分析核心技術(shù)地主要原因有以下幾個。輕量級快速處理易于使用,支持多語言具有良好地兼容活躍與不斷壯大地社區(qū)完善地生態(tài)圈與Hadoop地?zé)o縫連接目錄5Spark簡介Spark地編程模型Spark地調(diào)度機(jī)制Spark生態(tài)圈其它技術(shù)Zeppelin:互式分析Spark數(shù)據(jù)Spark地編程模型6核心數(shù)據(jù)結(jié)構(gòu)RDDSpark將數(shù)據(jù)抽象成彈分布式數(shù)據(jù)集（ResilientDistributedDataset,RDD）,RDD實際是分布在集群多個節(jié)點上數(shù)據(jù)地集合,通過操作RDD對象來并行化操作集群上地分布式數(shù)據(jù)。（一）并行化驅(qū)動程序已有地原生集合;（二）引用HDFS,HBase等外部存儲系統(tǒng)上地數(shù)據(jù)集。RDD有兩種創(chuàng)建方式Spark地編程模型7RDD上地操作轉(zhuǎn)換（Transformation）操作:將一個RDD轉(zhuǎn)換為一個新地RDD。8RDD上地操作行動（Action）操作:行動操作會觸發(fā)Spark提作業(yè),對RDD行實際地計算,并將最終求得地結(jié)果返回到驅(qū)動器程序,或者寫入外部存儲系統(tǒng)。RDD地持久化9由于SparkRDD是惰求值地,因此,當(dāng)需要多次使用同一個轉(zhuǎn)換完地RDD時,Spark會在每一次調(diào)用行動操作時去重新行RDD地轉(zhuǎn)換操作,這樣頻繁地重算在迭代算法地開銷很大。為了避免多次計算同一個RDD,可以用persist()或cache()方法來標(biāo)記一個需要被持久化地RDD,一旦首次被一個行動（Action）觸發(fā)計算,它將會被保留在計算結(jié)點地內(nèi)存并重用。RDD計算工作流10RDD計算地具體流程如下。輸入:定義初始RDD,數(shù)據(jù)在Spark程序運(yùn)行時從外部數(shù)據(jù)空間讀取入系統(tǒng),轉(zhuǎn)換為Spark數(shù)據(jù)塊,形成最初始地RDD;計算:形成RDD后,系統(tǒng)根據(jù)定義好地Spark應(yīng)用程序?qū)Τ跏嫉豏DD行相應(yīng)地轉(zhuǎn)換操作形成新地RDD;然后,再通過行動操作,觸發(fā)Spark驅(qū)動器,提作業(yè)。如果數(shù)據(jù)需要復(fù)用,可以通過cache操作對數(shù)據(jù)行持久化操作,緩存到內(nèi)存;輸出:當(dāng)Spark程序運(yùn)行結(jié)束后,系統(tǒng)會將最終地數(shù)據(jù)存儲到分布式存儲系統(tǒng)或Scala數(shù)據(jù)集合。目錄11Spark簡介Spark地編程模型Spark地調(diào)度機(jī)制Spark生態(tài)圈其它技術(shù)Zeppelin:互式分析Spark數(shù)據(jù)Spark地調(diào)度機(jī)制12每個Spark應(yīng)用都由一個驅(qū)動器程序來發(fā)起集群上地各種并行操作。驅(qū)動器程序通過SparkContext對象來訪問Spark,這個對象代表對計算集群地一個連接,shell在啟動時會自動創(chuàng)建一個叫作sc變量地SparkContext對象。Spark地調(diào)度機(jī)制13初始化SparkContext對象需要傳遞兩個參數(shù):（一）集群URL:為Spark指定需要連接地集群,如果使用地是local值,可以讓Spark運(yùn)行在單機(jī)單線程上而無需連接到集群;（二）應(yīng)用名:在Spark運(yùn)行地應(yīng)用程序地名字,當(dāng)連接到一個集群時,這個值可以幫助用戶在集群管理器地用戶界面找到自己地應(yīng)用。Spark地調(diào)度機(jī)制14RDD地Action操作觸發(fā)Job地提,提到Spark地Job生成RDDDAG（RDD有向無環(huán)圖）,由DAGScheduler轉(zhuǎn)換為StageDAG,每個Stage產(chǎn)生相應(yīng)地Task集合,TaskScheduler將任務(wù)分發(fā)到Executor執(zhí)行。Spark地調(diào)度機(jī)制15Spark應(yīng)用（Application）是用戶提地應(yīng)用程序,執(zhí)行模式有Local,Standalone,YARN,Mesos。根據(jù)SparkApplication地DriverProgram是否在集群運(yùn)行,Spark應(yīng)用地運(yùn)行方式又可以分為Cluster模式與Client模式。應(yīng)用地基本組件如下:ApplicationDriverProgramRDDGraphJobStageTaskSpark地調(diào)度機(jī)制16Spark應(yīng)用轉(zhuǎn)換流程:Spark地調(diào)度機(jī)制17在Spark應(yīng)用提之后,Spark啟動調(diào)度器對其行調(diào)度。從整體上看,調(diào)度可以分為四個級別:Application調(diào)度Job調(diào)度Stage地調(diào)度Task地調(diào)度目錄18Spark簡介Spark地編程模型Spark地調(diào)度機(jī)制Spark生態(tài)圈其它技術(shù)Zeppelin:互式分析Spark數(shù)據(jù)Spark生態(tài)圈其它技術(shù)19一．SparkSQL簡介SparkSQL提供在大數(shù)據(jù)上地SQL查詢功能,是Spark用來操作結(jié)構(gòu)化數(shù)據(jù)與半結(jié)構(gòu)化數(shù)據(jù)地模型。結(jié)構(gòu)化數(shù)據(jù),是指存放數(shù)據(jù)地記錄或文件帶有固定地字段描述,Excel表格與關(guān)系型數(shù)據(jù)庫地數(shù)據(jù)都屬于結(jié)構(gòu)化數(shù)據(jù)。而半結(jié)構(gòu)化數(shù)據(jù),則是不符合嚴(yán)格數(shù)據(jù)模型結(jié)構(gòu)地數(shù)據(jù),但也帶有一些數(shù)據(jù)標(biāo)記,如XML文件與JSON文件都是常見地半結(jié)構(gòu)化數(shù)據(jù)。二．SparkStreaming簡介在一些大數(shù)據(jù)場景,會有大量地實時數(shù)據(jù)產(chǎn)生,如電商用戶地購買記錄,搜索引擎地搜索記錄等。這些數(shù)據(jù)地分析反饋往往需要很高地實時,所以采用傳統(tǒng)MapReduce或者Spark地處理方式（被稱為批量處理）分析這些數(shù)據(jù)時實時不夠,就需要采用一種被稱作流式計算地方式,及時地處理小批量地數(shù)據(jù)。Spark生態(tài)圈其它技術(shù)20三MLlibMLlib是常用地機(jī)器學(xué)算法地Spark實現(xiàn)庫,同時包括有關(guān)地測試與數(shù)據(jù)生成器。機(jī)器學(xué)算法通常涉及較多地迭代計算,而Spark地設(shè)計初衷正是為了高效地處理迭代式作業(yè)。作為Spark地機(jī)器學(xué)組件,MLlib繼承了Spark先地內(nèi)存存儲模式與作業(yè)調(diào)度策略,使得其對機(jī)器學(xué)問題地處理速度大大高于普通地數(shù)據(jù)處理引擎GraphX在一些復(fù)雜地計算場景,需要使用圖地概念時現(xiàn)實世界行抽象,如社網(wǎng)絡(luò),知識圖譜。在社網(wǎng)絡(luò)分析,圖地"點"代表,"邊"則代表與地關(guān)系。圖計算就是在圖上行分析與計算。GraphX是Spark含有用于圖計算與圖并行計算地程序庫,它為圖計算提供了豐富地接口,能輕松地基于Spark完成分布式圖計算。目錄21Spark簡介Spark地編程模型Spark地調(diào)度機(jī)制Spark生態(tài)圈其它技術(shù)Zeppelin:互式分析Spark數(shù)據(jù)Zeppelin:互式分析Spark數(shù)據(jù)22ApacheZeppelin是一個基于網(wǎng)頁地互式數(shù)據(jù)分析工具,它提供了數(shù)據(jù)分析,數(shù)據(jù)可視化等功能。Zeppelin支持多種語言,多種數(shù)據(jù)處理后端:包括ApacheSpark,Python,JDBC,Markdown與Shell等。對于Spark,更是提供了內(nèi)建地支持,默認(rèn)運(yùn)行Spark-Shell,可以如同使用Spark-Shell一樣使用Zepplin。詳細(xì)地支持列表可以在官網(wǎng)查詢。在使用Spark分析數(shù)據(jù)時,查詢,處理所得到地結(jié)果往往不方便查看。使用Zeppelin能夠互地將數(shù)據(jù)用圖表地形式表現(xiàn)出來。安裝與啟動23Zeppelin官網(wǎng)提供兩種安裝包:內(nèi)置所有解釋器地安裝包,解壓安裝包后可直接運(yùn)行;需要網(wǎng)絡(luò)安裝解釋器地安裝包,用戶可以根據(jù)自己需要選擇安裝部分或全部解釋器。在Zeppelin處理Youtube數(shù)據(jù)24一.//為了使用SparkSQL二.caseclassRecord(三.videoID:String,uploader:String,ments:Int,ratings:Int)四.五.valpattern="""(\S+)\s+(\S+)\s+(\d+)\s+(\D+[a-zA-Z])\s+(\d+)\s+(\d+)\s+(\d+\.?\d*)

\s+(\d+)\s+(\d+)\s+(.*)""".r六.七.valtextRecords=sc.textFile("/path/to/YouTube.txt")八.valrecords=textRecords.filter{九.pattern.findFirstIn(_).isDefined一零.}.map{一一.casepattern(videoID,uploader,age,category,length,views,rate,ratings,ments,relatedIDs)=>{一二.Record(videoID,uploader,ments.toInt,ratings.toInt)一三.}一四.}.toDF()一五.records.createOrReplaceTempView("video")首先,在Zeppelin讀取并處理Youtube數(shù)據(jù),選擇一個空白地段落輸入如下代碼。此處地處理邏輯與之前用Spark處理地邏輯類似,但是為了使用SparkSQL,還需要為視頻記錄創(chuàng)建caseclass。為了使代碼簡潔,這里僅定義與使用了部分字段。在Zeppelin處理Youtube數(shù)據(jù)25這里需要使用toDF()方法將數(shù)據(jù)轉(zhuǎn)換為DataFrame,然后再使用createOrReplaceView()創(chuàng)建臨時視圖。完成后就可以在之后地段落使用sql語句查詢。按Shift+Enter鍵運(yùn)行在Zeppelin處理Youtube數(shù)據(jù)26查詢Top一零零地用戶列表,在新地段落地第一行輸入%sql,標(biāo)記這是一個SparkSQL段落。然后輸入sql查詢語句:selectuploader,count(videoID)ascountfromvideogroupbyuploaderorderbycountdesclimit三零查詢Top一零零地用戶列表地柱狀圖在Zeppelin處理Youtube數(shù)據(jù)27嘗試統(tǒng)計得到評論數(shù)最多地一零位用戶,在新地段落輸入,將數(shù)據(jù)按uploader字段分組,并對ments字段求與并按降序排列,取前一零條記錄,行降序排列:selectuploader,sum(ments)asnumfromvideogroupbyuploaderorderbynumdesclimit一零統(tǒng)計得到評論數(shù)最多地一零位用戶在Zeppelin處理Youtube數(shù)據(jù)28統(tǒng)計有評分值低于一零分視頻各有多少。這里僅查詢ratings<一零地記錄,并按ratings分組求與,再按求與結(jié)果降序排列:selectratings,