Chapter16-廈門大學-林子雨-大數(shù)據(jù)技術(shù)原理與應用-第十六章-Spark(2016年4月20日版本)

廈門大學計算機科學系2016年版林子雨廈門大學計算機科學系E-mail:ziyulin@主頁：/linziyu

第十講Spark

（PPT版本號：2016年4月20日版本）

《大數(shù)據(jù)技術(shù)原理與應用》/post/bigdata溫馨提示：編輯幻燈片母版，可以修改每頁PPT的廈大?；蘸偷撞课淖终n堂內(nèi)容與教材對應關(guān)系說明廈門大學林子雨編著《大數(shù)據(jù)技術(shù)原理與應用》2015年8月1日人民郵電出版社出版發(fā)行第1版教材共包含13章內(nèi)容第一章大數(shù)據(jù)概述第二章大數(shù)據(jù)處理架構(gòu)Hadoop第三章分布式文件系統(tǒng)HDFS第四章分布式數(shù)據(jù)庫HBase第五章NoSQL數(shù)據(jù)庫第六章云數(shù)據(jù)庫第七章MapReduce第八章流計算第九章圖計算第十章數(shù)據(jù)可視化第十一章大數(shù)據(jù)在互聯(lián)網(wǎng)領(lǐng)域的應用第十二章大數(shù)據(jù)在生物醫(yī)學領(lǐng)域的應用（自學）第十三章大數(shù)據(jù)的其他應用（自學）2016年新增章節(jié)（將加入到第2版教材中）第14章基于Hadoop的數(shù)據(jù)倉庫Hive第15章Hadoop架構(gòu)再探討第16章Spark課堂內(nèi)容與教材對應關(guān)系說明課堂章節(jié)對應的《大數(shù)據(jù)技術(shù)原理與應用》（第1版）教材章節(jié)第1講-大數(shù)據(jù)概述第1章-大數(shù)據(jù)概述第2講-大數(shù)據(jù)處理架構(gòu)Hadoop第2章-大數(shù)據(jù)處理架構(gòu)Hadoop第3講-分布式文件系統(tǒng)HDFS第3章-分布式文件系統(tǒng)HDFS第4講-分布式數(shù)據(jù)庫HBase第4章-分布式數(shù)據(jù)庫HBase第5講-NoSQL數(shù)據(jù)庫第5章-NoSQL數(shù)據(jù)庫第6講-云數(shù)據(jù)庫第6章-云數(shù)據(jù)庫第7講-MapReduce第7章-MapReduce第8講-基于Hadoop的數(shù)據(jù)倉庫Hive新增第14章，不在當前第1版教材中，將放入第2版教材第9講-Hadoop架構(gòu)再探討新增第15章，不在當前第1版教材中，將放入第2版教材第10講-Spark新增第16章，不在當前第1版教材中，將放入第2版教材第11講-流計算第8章-流計算第12講-圖計算第9章-圖計算第13講-數(shù)據(jù)可視化第10章-數(shù)據(jù)可視化第14講-大數(shù)據(jù)在互聯(lián)網(wǎng)領(lǐng)域的應用第11章-大數(shù)據(jù)在互聯(lián)網(wǎng)領(lǐng)域的應用備注：教材的第12章大數(shù)據(jù)在生物醫(yī)學領(lǐng)域的應用和第13章大數(shù)據(jù)在其他領(lǐng)域的應用，為自學章節(jié)，不錄制視頻

廈門大學計算機科學系2016年版林子雨廈門大學計算機科學系E-mail:ziyulin@主頁：/linziyu

第十六章Spark

（第1版教材出版后的2016年新增章節(jié)）《大數(shù)據(jù)技術(shù)原理與應用》/post/bigdata溫馨提示：編輯幻燈片母版，可以修改每頁PPT的廈大?；蘸偷撞课淖郑≒PT版本號：2016年4月20日版本）

中國高校大數(shù)據(jù)課程公共服務平臺/post/bigdata-teaching-platform/百度搜索“廈門大學數(shù)據(jù)庫實驗室”訪問平臺主頁課程教材講義PPT學習指南免費提供備課指南上機習題授課視頻技術(shù)資料全方位、一站式服務提綱本PPT是如下教材的配套講義：21世紀高等教育計算機規(guī)劃教材《大數(shù)據(jù)技術(shù)原理與應用——概念、存儲、處理、分析與應用》（2015年8月第1版）廈門大學林子雨編著，人民郵電出版社ISBN:978-7-115-39287-9歡迎訪問《大數(shù)據(jù)技術(shù)原理與應用》教材官方網(wǎng)站：/post/bigdata歡迎訪問“中國高校大數(shù)據(jù)課程公共服務平臺”旗下子欄目“大數(shù)據(jù)課程學生服務站”，為學生學習大數(shù)據(jù)課程提供全方位、一站式免費服務：/post/4331/16.1Spark概述16.2Spark生態(tài)系統(tǒng)16.3Spark運行架構(gòu)16.4SparkSQL16.5Spark的部署和應用方式16.6Spark編程實踐16.1Spark概述16.1.1Spark簡介16.1.2Scala簡介16.1.3Spark與Hadoop的比較16.1.1Spark簡介Spark最初由美國加州伯克利大學（UCBerkeley）的AMP實驗室于2009年開發(fā)，是基于內(nèi)存計算的大數(shù)據(jù)并行計算框架，可用于構(gòu)建大型的、低延遲的數(shù)據(jù)分析應用程序2013年Spark加入Apache孵化器項目后發(fā)展迅猛，如今已成為Apache軟件基金會最重要的三大分布式計算系統(tǒng)開源項目之一（Hadoop、Spark、Storm）Spark在2014年打破了Hadoop保持的基準排序紀錄Spark/206個節(jié)點/23分鐘/100TB數(shù)據(jù)Hadoop/2000個節(jié)點/72分鐘/100TB數(shù)據(jù)Spark用十分之一的計算資源，獲得了比Hadoop快3倍的速度16.1.1Spark簡介Spark具有如下幾個主要特點：運行速度快：使用DAG執(zhí)行引擎以支持循環(huán)數(shù)據(jù)流與內(nèi)存計算容易使用：支持使用Scala、Java、Python和R語言進行編程，可以通過SparkShell進行交互式編程通用性：Spark提供了完整而強大的技術(shù)棧，包括SQL查詢、流式計算、機器學習和圖算法組件運行模式多樣：可運行于獨立的集群模式中，可運行于Hadoop中，也可運行于AmazonEC2等云環(huán)境中，并且可以訪問HDFS、Cassandra、HBase、Hive等多種數(shù)據(jù)源

16.1.1Spark簡介圖16-1谷歌趨勢：Spark與Hadoop對比Spark如今已吸引了國內(nèi)外各大公司的注意，如騰訊、淘寶、百度、亞馬遜等公司均不同程度地使用了Spark來構(gòu)建大數(shù)據(jù)分析應用，并應用到實際的生產(chǎn)環(huán)境中16.1.2Scala簡介Scala是一門現(xiàn)代的多范式編程語言，運行于Java平臺（JVM，Java虛擬機），并兼容現(xiàn)有的Java程序Scala的特性：Scala具備強大的并發(fā)性，支持函數(shù)式編程，可以更好地支持分布式系統(tǒng)Scala語法簡潔，能提供優(yōu)雅的APIScala兼容Java，運行速度快，且能融合到Hadoop生態(tài)圈中

Scala是Spark的主要編程語言，但Spark還支持Java、Python、R作為編程語言Scala的優(yōu)勢是提供了REPL（Read-Eval-PrintLoop，交互式解釋器），提高程序開發(fā)效率16.1.3Spark與Hadoop的對比Hadoop存在如下一些缺點：表達能力有限磁盤IO開銷大延遲高任務之間的銜接涉及IO開銷在前一個任務執(zhí)行完成之前，其他任務就無法開始，難以勝任復雜、多階段的計算任務

16.1.3Spark與Hadoop的對比Spark在借鑒HadoopMapReduce優(yōu)點的同時，很好地解決了MapReduce所面臨的問題相比于HadoopMapReduce，Spark主要具有如下優(yōu)點：Spark的計算模式也屬于MapReduce，但不局限于Map和Reduce操作，還提供了多種數(shù)據(jù)集操作類型，編程模型比HadoopMapReduce更靈活Spark提供了內(nèi)存計算，可將中間結(jié)果放到內(nèi)存中，對于迭代運算效率更高Spark基于DAG的任務調(diào)度執(zhí)行機制，要優(yōu)于HadoopMapReduce的迭代執(zhí)行機制

16.1.3Spark與Hadoop的對比圖16-2Hadoop與Spark的執(zhí)行流程對比16.1.3Spark與Hadoop的對比圖16-3Hadoop與Spark執(zhí)行邏輯回歸的時間對比使用Hadoop進行迭代計算非常耗資源Spark將數(shù)據(jù)載入內(nèi)存后，之后的迭代計算都可以直接使用內(nèi)存中的中間結(jié)果作運算，避免了從磁盤中頻繁讀取數(shù)據(jù)16.2Spark生態(tài)系統(tǒng)在實際應用中，大數(shù)據(jù)處理主要包括以下三個類型：復雜的批量數(shù)據(jù)處理：通常時間跨度在數(shù)十分鐘到數(shù)小時之間基于歷史數(shù)據(jù)的交互式查詢：通常時間跨度在數(shù)十秒到數(shù)分鐘之間基于實時數(shù)據(jù)流的數(shù)據(jù)處理：通常時間跨度在數(shù)百毫秒到數(shù)秒之間當同時存在以上三種場景時，就需要同時部署三種不同的軟件比如:MapReduce/Impala/Storm這樣做難免會帶來一些問題：不同場景之間輸入輸出數(shù)據(jù)無法做到無縫共享，通常需要進行數(shù)據(jù)格式的轉(zhuǎn)換不同的軟件需要不同的開發(fā)和維護團隊，帶來了較高的使用成本比較難以對同一個集群中的各個系統(tǒng)進行統(tǒng)一的資源協(xié)調(diào)和分配16.2Spark生態(tài)系統(tǒng)Spark的設計遵循“一個軟件棧滿足不同應用場景”的理念，逐漸形成了一套完整的生態(tài)系統(tǒng)既能夠提供內(nèi)存計算框架，也可以支持SQL即席查詢、實時流式計算、機器學習和圖計算等Spark可以部署在資源管理器YARN之上，提供一站式的大數(shù)據(jù)解決方案因此，Spark所提供的生態(tài)系統(tǒng)足以應對上述三種場景，即同時支持批處理、交互式查詢和流數(shù)據(jù)處理16.2Spark生態(tài)系統(tǒng)Spark的生態(tài)系統(tǒng)主要包含了SparkCore、SparkSQL、SparkStreaming、MLLib和GraphX等組件圖16-4BDAS架構(gòu)Spark生態(tài)系統(tǒng)已經(jīng)成為伯克利數(shù)據(jù)分析軟件棧BDAS（BerkeleyDataAnalyticsStack）的重要組成部分16.2Spark生態(tài)系統(tǒng)應用場景時間跨度其他框架Spark生態(tài)系統(tǒng)中的組件復雜的批量數(shù)據(jù)處理小時級MapReduce、HiveSpark基于歷史數(shù)據(jù)的交互式查詢分鐘級、秒級Impala、Dremel、DrillSparkSQL基于實時數(shù)據(jù)流的數(shù)據(jù)處理毫秒、秒級Storm、S4SparkStreaming基于歷史數(shù)據(jù)的數(shù)據(jù)挖掘-MahoutMLlib圖結(jié)構(gòu)數(shù)據(jù)的處理-Pregel、HamaGraphX表1Spark生態(tài)系統(tǒng)組件的應用場景16.3Spark運行架構(gòu)16.3.1基本概念16.3.2架構(gòu)設計16.3.3Spark運行基本流程16.3.4Spark運行原理16.3.1基本概念RDD：是ResillientDistributedDataset（彈性分布式數(shù)據(jù)集）的簡稱，是分布式內(nèi)存的一個抽象概念，提供了一種高度受限的共享內(nèi)存模型DAG：是DirectedAcyclicGraph（有向無環(huán)圖）的簡稱，反映RDD之間的依賴關(guān)系Executor：是運行在工作節(jié)點（WorkerNode）的一個進程，負責運行TaskApplication：用戶編寫的Spark應用程序Task：運行在Executor上的工作單元

Job：一個Job包含多個RDD及作用于相應RDD上的各種操作Stage：是Job的基本調(diào)度單位，一個Job會分為多組Task，每組Task被稱為Stage，或者也被稱為TaskSet，代表了一組關(guān)聯(lián)的、相互之間沒有Shuffle依賴關(guān)系的任務組成的任務集16.3.2架構(gòu)設計圖16-5Spark運行架構(gòu)Spark運行架構(gòu)包括集群資源管理器（ClusterManager）、運行作業(yè)任務的工作節(jié)點（WorkerNode）、每個應用的任務控制節(jié)點（Driver）和每個工作節(jié)點上負責具體任務的執(zhí)行進程（Executor）資源管理器可以自帶或Mesos或YARN與HadoopMapReduce計算框架相比，Spark所采用的Executor有兩個優(yōu)點：一是利用多線程來執(zhí)行具體的任務，減少任務的啟動開銷二是Executor中有一個BlockManager存儲模塊，會將內(nèi)存和磁盤共同作為存儲設備，有效減少IO開銷16.3.2架構(gòu)設計圖16-6Spark中各種概念之間的相互關(guān)系一個Application由一個Driver和若干個Job構(gòu)成，一個Job由多個Stage構(gòu)成，一個Stage由多個沒有Shuffle關(guān)系的Task組成當執(zhí)行一個Application時，Driver會向集群管理器申請資源，啟動Executor，并向Executor發(fā)送應用程序代碼和文件，然后在Executor上執(zhí)行Task，運行結(jié)束后，執(zhí)行結(jié)果會返回給Driver，或者寫到HDFS或者其他數(shù)據(jù)庫中16.3.3Spark運行基本流程圖16-7Spark運行基本流程圖（1）首先為應用構(gòu)建起基本的運行環(huán)境，即由Driver創(chuàng)建一個SparkContext，進行資源的申請、任務的分配和監(jiān)控（2）資源管理器為Executor分配資源，并啟動Executor進程（3）SparkContext根據(jù)RDD的依賴關(guān)系構(gòu)建DAG圖，DAG圖提交給DAGScheduler解析成Stage，然后把一個個TaskSet提交給底層調(diào)度器TaskScheduler處理；Executor向SparkContext申請Task，TaskScheduler將Task發(fā)放給Executor運行，并提供應用程序代碼（4）Task在Executor上運行，把執(zhí)行結(jié)果反饋給TaskScheduler，然后反饋給DAGScheduler，運行完畢后寫入數(shù)據(jù)并釋放所有資源

16.3.3Spark運行基本流程總體而言，Spark運行架構(gòu)具有以下特點：（1）每個Application都有自己專屬的Executor進程，并且該進程在Application運行期間一直駐留。Executor進程以多線程的方式運行Task（2）Spark運行過程與資源管理器無關(guān)，只要能夠獲取Executor進程并保持通信即可（3）Task采用了數(shù)據(jù)本地性和推測執(zhí)行等優(yōu)化機制16.3.4RDD運行原理1.設計背景2.RDD概念3.RDD特性4.RDD之間的依賴關(guān)系5.Stage的劃分6.RDD運行過程16.3.4RDD運行原理1.設計背景許多迭代式算法（比如機器學習、圖算法等）和交互式數(shù)據(jù)挖掘工具，共同之處是，不同計算階段之間會重用中間結(jié)果目前的MapReduce框架都是把中間結(jié)果寫入到HDFS中，帶來了大量的數(shù)據(jù)復制、磁盤IO和序列化開銷RDD就是為了滿足這種需求而出現(xiàn)的，它提供了一個抽象的數(shù)據(jù)架構(gòu)，我們不必擔心底層數(shù)據(jù)的分布式特性，只需將具體的應用邏輯表達為一系列轉(zhuǎn)換處理，不同RDD之間的轉(zhuǎn)換操作形成依賴關(guān)系，可以實現(xiàn)管道化，避免中間數(shù)據(jù)存儲16.3.4RDD運行原理2.RDD概念一個RDD就是一個分布式對象集合，本質(zhì)上是一個只讀的分區(qū)記錄集合，每個RDD可分成多個分區(qū)，每個分區(qū)就是一個數(shù)據(jù)集片段，并且一個RDD的不同分區(qū)可以被保存到集群中不同的節(jié)點上，從而可以在集群中的不同節(jié)點上進行并行計算RDD提供了一種高度受限的共享內(nèi)存模型，即RDD是只讀的記錄分區(qū)的集合，不能直接修改，只能基于穩(wěn)定的物理存儲中的數(shù)據(jù)集創(chuàng)建RDD，或者通過在其他RDD上執(zhí)行確定的轉(zhuǎn)換操作（如map、join和groupby）而創(chuàng)建得到新的RDD16.3.4RDD運行原理RDD提供了一組豐富的操作以支持常見的數(shù)據(jù)運算，分為“動作”（Action）和“轉(zhuǎn)換”（Transformation）兩種類型RDD提供的轉(zhuǎn)換接口都非常簡單，都是類似map、filter、groupBy、join等粗粒度的數(shù)據(jù)轉(zhuǎn)換操作，而不是針對某個數(shù)據(jù)項的細粒度修改（不適合網(wǎng)頁爬蟲）表面上RDD的功能很受限、不夠強大，實際上RDD已經(jīng)被實踐證明可以高效地表達許多框架的編程模型（比如MapReduce、SQL、Pregel）Spark用Scala語言實現(xiàn)了RDD的API，程序員可以通過調(diào)用API實現(xiàn)對RDD的各種操作16.3.4RDD運行原理RDD典型的執(zhí)行過程如下：RDD讀入外部數(shù)據(jù)源進行創(chuàng)建RDD經(jīng)過一系列的轉(zhuǎn)換（Transformation）操作，每一次都會產(chǎn)生不同的RDD，供給下一個轉(zhuǎn)換操作使用最后一個RDD經(jīng)過“動作”操作進行轉(zhuǎn)換，并輸出到外部數(shù)據(jù)源

圖16-8RDD執(zhí)行過程的一個實例這一系列處理稱為一個Lineage（血緣關(guān)系），即DAG拓撲排序的結(jié)果優(yōu)點：惰性調(diào)用、管道化、避免同步等待、不需要保存中間結(jié)果、每次操作變得簡單動作轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換創(chuàng)建創(chuàng)建16.3.4RDD運行原理Spark采用RDD以后能夠?qū)崿F(xiàn)高效計算的原因主要在于：（1）高效的容錯性現(xiàn)有容錯機制：數(shù)據(jù)復制或者記錄日志RDD：血緣關(guān)系、重新計算丟失分區(qū)、無需回滾系統(tǒng)、重算過程在不同節(jié)點之間并行、只記錄粗粒度的操作（2）中間結(jié)果持久化到內(nèi)存，數(shù)據(jù)在內(nèi)存中的多個RDD操作之間進行傳遞，避免了不必要的讀寫磁盤開銷（3）存放的數(shù)據(jù)可以是Java對象，避免了不必要的對象序列化和反序列化3.RDD特性16.3.4RDD運行原理窄依賴表現(xiàn)為一個父RDD的分區(qū)對應于一個子RDD的分區(qū)或多個父RDD的分區(qū)對應于一個子RDD的分區(qū)寬依賴則表現(xiàn)為存在一個父RDD的一個分區(qū)對應一個子RDD的多個分區(qū)圖16-9窄依賴與寬依賴的區(qū)別4.RDD之間的依賴關(guān)系16.3.4RDD運行原理Spark通過分析各個RDD的依賴關(guān)系生成了DAG，再通過分析各個RDD中的分區(qū)之間的依賴關(guān)系來決定如何劃分Stage，具體劃分方法是：在DAG中進行反向解析，遇到寬依賴就斷開遇到窄依賴就把當前的RDD加入到Stage中將窄依賴盡量劃分在同一個Stage中，可以實現(xiàn)流水線計算5.Stage的劃分16.3.4RDD運行原理圖16-10根據(jù)RDD分區(qū)的依賴關(guān)系劃分Stage5.Stage的劃分被分成三個Stage，在Stage2中，從map到union都是窄依賴，這兩步操作可以形成一個流水線操作流水線操作實例分區(qū)7通過map操作生成的分區(qū)9，可以不用等待分區(qū)8到分區(qū)10這個map操作的計算結(jié)束，而是繼續(xù)進行union操作，得到分區(qū)13，這樣流水線執(zhí)行大大提高了計算的效率16.3.4RDD運行原理Stage的類型包括兩種：ShuffleMapStage和ResultStage，具體如下：（1）ShuffleMapStage：不是最終的Stage，在它之后還有其他Stage，所以，它的輸出一定需要經(jīng)過Shuffle過程，并作為后續(xù)Stage的輸入；這種Stage是以Shuffle為輸出邊界，其輸入邊界可以是從外部獲取數(shù)據(jù)，也可以是另一個ShuffleMapStage的輸出，其輸出可以是另一個Stage的開始；在一個Job里可能有該類型的Stage，也可能沒有該類型Stage；（2）ResultStage：最終的Stage，沒有輸出，而是直接產(chǎn)生結(jié)果或存儲。這種Stage是直接輸出結(jié)果，其輸入邊界可以是從外部獲取數(shù)據(jù)，也可以是另一個ShuffleMapStage的輸出。在一個Job里必定有該類型Stage。因此，一個Job含有一個或多個Stage，其中至少含有一個ResultStage。5.Stage的劃分16.3.4RDD運行原理通過上述對RDD概念、依賴關(guān)系和Stage劃分的介紹，結(jié)合之前介紹的Spark運行基本流程，再總結(jié)一下RDD在Spark架構(gòu)中的運行過程：（1）創(chuàng)建RDD對象；（2）SparkContext負責計算RDD之間的依賴關(guān)系，構(gòu)建DAG；（3）DAGScheduler負責把DAG圖分解成多個Stage，每個Stage中包含了多個Task，每個Task會被TaskScheduler分發(fā)給各個WorkerNode上的Executor去執(zhí)行。圖16-11RDD在Spark中的運行過程16.4SparkSQL16.4.1從Shark說起16.4.2SparkSQL設計16.4.1從Shark說起Shark即HiveonSpark，為了實現(xiàn)與Hive兼容，Shark在HiveQL方面重用了Hive中HiveQL的解析、邏輯執(zhí)行計劃翻譯、執(zhí)行計劃優(yōu)化等邏輯，可以近似認為僅將物理執(zhí)行計劃從MapReduce作業(yè)替換成了Spark作業(yè)，通過Hive的HiveQL解析，把HiveQL翻譯成Spark上的RDD操作。Shark的設計導致了兩個問題：一是執(zhí)行計劃優(yōu)化完全依賴于Hive，不方便添加新的優(yōu)化策略；二是因為Spark是線程級并行，而MapReduce是進程級并行，因此，Spark在兼容Hive的實現(xiàn)上存在線程安全問題，導致Shark不得不使用另外一套獨立維護的打了補丁的Hive源碼分支Hive中SQL查詢的MapReduce作業(yè)轉(zhuǎn)化過程16.4.2SparkSQL設計SparkSQL在Hive兼容層面僅依賴HiveQL解析、Hive元數(shù)據(jù)，也就是說，從HQL被解析成抽象語法樹（AST）起，就全部由SparkSQL接管了。SparkSQL執(zhí)行計劃生成和優(yōu)化都由Catalyst（函數(shù)式關(guān)系查詢優(yōu)化框架）負責圖16-12SparkSQL架構(gòu)16.4.2SparkSQL設計圖16-13SparkSQL支持的數(shù)據(jù)格式和編程語言SparkSQL增加了SchemaRDD（即帶有Schema信息的RDD），使用戶可以在SparkSQL中執(zhí)行SQL語句，數(shù)據(jù)既可以來自RDD，也可以是Hive、HDFS、Cassandra等外部數(shù)據(jù)源，還可以是JSON格式的數(shù)據(jù)SparkSQL目前支持Scala、Java、Python三種語言，支持SQL-92規(guī)范16.5Spark的部署和應用方式16.5.1Spark三種部署方式16.5.2從Hadoop+Storm架構(gòu)轉(zhuǎn)向Spark架構(gòu)16.5.3Hadoop和Spark的統(tǒng)一部署16.5.1Spark三種部署方式Spark支持三種不同類型的部署方式，包括：Standalone（類似于MapReduce1.0，slot為資源分配單位）SparkonMesos（和Spark有血緣關(guān)系，更好支持Mesos）SparkonYARN圖16-17SparkonYarn架構(gòu)16.5.2從Hadoop+Storm架構(gòu)轉(zhuǎn)向Spark架構(gòu)圖16-18采用Hadoop+Storm部署方式的一個案例這種架構(gòu)部署較為繁瑣16.5.2從Hadoop+Storm架構(gòu)轉(zhuǎn)向Spark架構(gòu)用Spark架構(gòu)具有如下優(yōu)點：實現(xiàn)一鍵式安裝和配置、線程級別的任務監(jiān)控和告警降低硬件集群、軟件維護、任務監(jiān)控和應用開發(fā)的難度便于做成統(tǒng)一的硬件、計算平臺資源池需要說明的是，SparkStreaming無法實現(xiàn)毫秒級的流計算，因此，對于需要毫秒級實時響應的企業(yè)應用而言，仍然需要采用流計算框架（如Storm）圖16-19用Spark架構(gòu)滿足批處理和流處理需求16.5.3Hadoop和Spark的統(tǒng)一部署不同的計算框架統(tǒng)一運行在YARN中，可以帶來如下好處：計算資源按需伸縮不用負載應用混搭，集群利用率高共享底層存儲，避免數(shù)據(jù)跨集群遷移圖16-20Hadoop和Spark的統(tǒng)一部署由于Hadoop生態(tài)系統(tǒng)中的一些組件所實現(xiàn)的功能，目前還是無法由Spark取代的，比如，Storm現(xiàn)有的Hadoop組件開發(fā)的應用，完全轉(zhuǎn)移到Spark上需要一定的成本16.6Spark編程實踐16.6.1Spark安裝16.6.2啟動SparkShell16.6.3SparkRDD基本操作16.6.4Spark應用程序

Spark上機實踐詳細過程，請參考廈門大學數(shù)據(jù)庫實驗室建設的“中國高校大數(shù)據(jù)課程公共服務平臺”中的“大數(shù)據(jù)課程學生服務站”中的“學習指南”欄目：學生服務站地址：/post/4331/學習指南欄目中包含了《第十六章Spark學習指南》/blog/778-2/掃一掃訪問學生服務站16.6.1Spark安裝安裝Spark之前需要安裝Java環(huán)境和Hadoop環(huán)境。下載地址：

進入下載頁面后，點擊主頁右側(cè)的“DownloadSpark”按鈕進入下載頁面，下載頁面中提供了幾個下載選項，主要是Sparkrelease及Packagetype的選擇，如下圖所示。第1項Sparkrelease一般默認選擇最新的發(fā)行版本，如截止至2016年3月份的最新版本為1.6.0。第2項packagetype則選擇“Pre-buildwithuser-providedHadoop[canusewithmostHadoopdistributions]”，可適用于多數(shù)Hadoop版本。選擇好之后，再點擊第4項給出的鏈接就可以下載Spark了。圖Spark下載選項16.6.1Spark安裝解壓安裝包spark-1.6.0-bin-without-hadoop.tgz至路徑/usr/local：

$cd/usr/local$sudomv./spark-1.6.0-bin-without-hadoop/./spark#更改文件夾名$sudochown-Rhadoop./spark#此處的hadoop為系統(tǒng)用戶名

配置Spark的Classpath。

編輯該配置文件，在文件最后面加上如下一行內(nèi)容：

保存配置文件后，就可以啟動、運行Spark了。Spark包含多種運行模式：單機模式、偽分布式模式、完全分布式模式。本章使用單機模式運行Spark。若需要使用HDFS中的文件，則在使用Spark前需要啟動Hadoop。$sudotar-zxf~/下載/spark-1.6.0-bin-without-hadoop.tgz-C/usr/local/$cd/usr/local/spark$cp./conf/spark-env.sh.template./conf/spark-env.sh#拷貝配置文件exportSPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoopclasspath)16.6.2啟動SparkShellSparkShell提供了簡單的方式來學習SparkAPISparkShell可以以實時、交互的方式來分析數(shù)據(jù)SparkShell支持Scala和Python

啟動SparkShell成功后在輸出信息的末尾可以看到“Scala>”的命令提示符，如下圖所示。圖6可通過Yarn進行資源管理的應用本章節(jié)內(nèi)容選擇使用Scala進行編程實踐，了解Scala有助于更好地掌握Spark。執(zhí)行如下命令啟動SparkShell：$./bin/spark-shell16.6.3SparkRDD基本操作Spark的主要操作對象是RDD，RDD可以通過多種方式靈活創(chuàng)建，可通過導入外部數(shù)據(jù)源建立，或者從其他的RDD轉(zhuǎn)化而來。在Spark程序中必須創(chuàng)建一個SparkContext對象，該對象是Spark程序的入口，負責創(chuàng)建RDD、啟動任務等。在啟動SparkShell后，該對象會自動創(chuàng)建，可以通過變量sc進行訪問。作為示例，我們選擇以Spark安裝目錄中的“README.md”文件作為數(shù)據(jù)源新建一個RDD，代碼如下：

SparkRDD支持兩種類型的操作：動作（action）：在數(shù)據(jù)集上進行運算，返回計算值轉(zhuǎn)換（transformation）：基于現(xiàn)有的數(shù)據(jù)集創(chuàng)建一個新的數(shù)據(jù)集

Scala>valtextFile=sc.textFile("file:///usr/local/spark/README.md")//通過file:前綴指定讀取本地文件16.6.3SparkRDD基本操作Spark提供了非常豐富的API，下面兩表格列出了幾個常用的動作、轉(zhuǎn)換API，更詳細的API及說明可查閱官方文檔。

ActionAPI說明count()返回數(shù)據(jù)集中的元素個數(shù)collect()以數(shù)組的形式返回數(shù)據(jù)集中的所有元素first()返回數(shù)據(jù)集中的第一個元素take(n)以數(shù)組的形式返回數(shù)據(jù)集中的前n個元素reduce(func)通過函數(shù)func（輸入兩個參數(shù)并返回一個值）聚合數(shù)據(jù)集中的元素foreach(func)將數(shù)據(jù)集中的每個元素傳遞到函數(shù)func中運行TransformationAPI說明filter(func)篩選出滿足函數(shù)func的元素，并返回一個新的數(shù)據(jù)集map(func)將每個元素傳遞到函數(shù)func中，并將結(jié)果返回為一個新的數(shù)據(jù)集flatMap(func)與map()相似，但每個輸入元素都可以映射到0或多個輸出結(jié)果groupByKey()應用于(K,V)鍵值對的數(shù)據(jù)集時，返回一個新的(K,Iterable<V>)形式的數(shù)據(jù)集reduceByKey(func)應用于(K,V)鍵值對的數(shù)據(jù)集時，返回一個新的(K,V)形式的數(shù)據(jù)集，其中的每個值是將每個key傳遞到函數(shù)func中進行聚合7-1常用的幾個ActionAPI介紹7-2常用的幾個TransformationAPI介紹16.6.3SparkRDD基本操作使用actionAPI-count()可以統(tǒng)計該文本文件的行數(shù)，命令如下：

輸出結(jié)果Long=95（“Long=95”表示該文件共有95行內(nèi)容）。

Scala>textFile.count()使用transformationAPI-filter()可以篩選出只包含Spark的行，命令如下：

第一條命令會返回一個新的RDD；輸出結(jié)果Long=17（表示該文件中共有17行內(nèi)容包含“Spark”）。Scala>vallinesWithSpark=textFile.filter(line=>line.contains("Spark"))Scala>linesWithSpark.count()也可以在同一條代碼中同時使用多個API，連續(xù)進行運算，稱為鏈式操作。不僅可以使Spark代碼更加簡潔，也優(yōu)化了計算過程。如上述兩條代碼可合并為如下一行代碼：

假設我們只需要得到包含“Spark”的行數(shù)，那么存儲篩選后的文本數(shù)據(jù)是多余的，因為這部分數(shù)據(jù)在計算得到行數(shù)后就不再使用到了。Spark基于整個操作鏈，僅儲存、計算所需的數(shù)據(jù)，提升了運行效率。Scala>vallinesCountWithSpark=textFile.filter(line=>line.contains("Spark")).count()16.6.3SparkRDD基本操作Spark屬于MapReduce計算模型，因此也可以實現(xiàn)MapReduce的計算流程，如實現(xiàn)單詞統(tǒng)計，可以使用如下的命令實現(xiàn)：首先使用flatMap()將每一行的文本內(nèi)容通過空格進行劃分為單詞；再使用map()將單詞映射為(K,V)的鍵值對，其中K為單詞，V為1；最后使用reduceByKey()將相同單詞的計數(shù)進行相加，最終得到該單詞總的出現(xiàn)的次數(shù)。

輸出結(jié)果Long=95（“Long=95”表示該文件共有95行內(nèi)容）。

Scala>valwordCounts=textFile.flatMap(line=>line.split("")).map(word=>(word,1)).reduceByKey((a,b)=>a+b)Scala>wordCounts.collect()//輸出單詞統(tǒng)計結(jié)果//Array[(String,Int)]=Array((package,1),(For,2),(Programs,1),(processing.,1),(Because,1),(The,1)...)16.6.4Spark應用程序在SparkShell中進行編程主要是方便對代碼進行調(diào)試，但需要以逐行代碼的方式運行。一般情況下，會選擇將調(diào)試后代碼打包成獨立的Spark應用程序，提交到Spark中運行。采用Scala編寫的程序需要使用sbt（SimpleBuildTool）進行打包，sbt的安裝配置步驟如下：下載sbt-launch.jar（下載地址/s/1eRyFddw）將下載后的文件拷貝至安裝目錄/usr/local/sbt中，命令如下：在安裝目錄中創(chuàng)建一個Shell腳本文件（文件路徑：/usr/local/sbt/sbt）用于啟動sbt，腳本文件中的代碼如下：保存后，還需要為該Shell腳本文件增加可執(zhí)行權(quán)限，命令如下：sudomkdir/usr/local/sbt#創(chuàng)建安裝目錄cp~/下載/sbt-launch.jar/usr/local/sbtsudochown-Rhadoop/usr/local/sbt#此處的hadoop為系統(tǒng)當前用戶名#!/bin/bashSBT_OPTS="-Xms512M-Xmx1536M-Xss1M-XX:+CMSClassUnloadingEnabled-XX:MaxPermSize=256M"java$SBT_OPTS-jar`dirname$0`/sbt-launch.jar"$@"chmodu+x/usr/local/sbt/sbt16.6.4Spark應用程序我們以一個簡單的程序為例，介紹如何打包并運行Spark程序，該程序的功能是統(tǒng)計文本文件中包含字母a和字b的各有多少行，具體步驟如下：

創(chuàng)建程序根目錄，并創(chuàng)建程序所需的文件夾結(jié)構(gòu)，命令如下：創(chuàng)建一個SimpleApp.scala文件（文件路徑：~/sparkapp/src/main/scala/SimpleApp.scala），文件中的代碼內(nèi)容如下：importorg.apache.spark.SparkContextimportorg.apache.spark.SparkContext._importorg.apache.spark.SparkConf

objectSimpleApp{defmain(args:Array[String]){vallogFile="file:///usr/local/spark/README.md"http://用于統(tǒng)計的文本文件valconf=newSparkConf().setAppName("SimpleApplication")valsc=newSparkContext(conf)vallogData=sc.textFile(logFile,2).cache()valnumAs=logData.filter(line=>line.contains("a")).count()valnumBs=logData.filter(line=>line.contains("b")).count()println("Lineswitha:%s,Lineswithb:%s".format(numAs,numBs))}}mkdir~/sparkapp#創(chuàng)建程序根目錄mkdir-p~/sparkapp/src/main/scala#創(chuàng)建程序所需的文件夾結(jié)構(gòu)16.6.4Spark應用程序然后創(chuàng)建一個simple.sbt文件（文件路徑：~/sparkapp/simple.sbt），用于聲明該應用程序的信息以及與Spark的依賴關(guān)系，具體內(nèi)容如下：使用sbt對該應用程序進行打包，命令如下：打包成功后，會輸出程序jar包的位置以