大數(shù)據(jù)技術(shù)原理與應(yīng)用 第十章 Spark

第十講Spark提綱16.1Spark概述16.2Spark生態(tài)系統(tǒng)16.3Spark運(yùn)行架構(gòu)16.4SparkSQL16.5Spark的部署和應(yīng)用方式16.6Spark編程實(shí)踐16.1Spark概述16.1.1Spark簡(jiǎn)介16.1.2Scala簡(jiǎn)介16.1.3Spark與Hadoop的比較16.1.1Spark簡(jiǎn)介Spark最初由美國加州伯克利大學(xué)（UCBerkeley）的AMP實(shí)驗(yàn)室于2009年開發(fā)，是基于內(nèi)存計(jì)算的大數(shù)據(jù)并行計(jì)算框架，可用于構(gòu)建大型的、低延遲的數(shù)據(jù)分析應(yīng)用程序2013年Spark加入Apache孵化器項(xiàng)目后發(fā)展迅猛，如今已成為Apache軟件基金會(huì)最重要的三大分布式計(jì)算系統(tǒng)開源項(xiàng)目之一（Hadoop、Spark、Storm）Spark在2014年打破了Hadoop保持的基準(zhǔn)排序紀(jì)錄Spark/206個(gè)節(jié)點(diǎn)/23分鐘/100TB數(shù)據(jù)Hadoop/2000個(gè)節(jié)點(diǎn)/72分鐘/100TB數(shù)據(jù)Spark用十分之一的計(jì)算資源，獲得了比Hadoop快3倍的速度16.1.1Spark簡(jiǎn)介Spark具有如下幾個(gè)主要特點(diǎn)：運(yùn)行速度快：使用DAG執(zhí)行引擎以支持循環(huán)數(shù)據(jù)流與內(nèi)存計(jì)算容易使用：支持使用Scala、Java、Python和R語言進(jìn)行編程，可以通過SparkShell進(jìn)行交互式編程通用性：Spark提供了完整而強(qiáng)大的技術(shù)棧，包括SQL查詢、流式計(jì)算、機(jī)器學(xué)習(xí)和圖算法組件運(yùn)行模式多樣：可運(yùn)行于獨(dú)立的集群模式中，可運(yùn)行于Hadoop中，也可運(yùn)行于AmazonEC2等云環(huán)境中，并且可以訪問HDFS、Cassandra、HBase、Hive等多種數(shù)據(jù)源16.1.1Spark簡(jiǎn)介圖16-1谷歌趨勢(shì)：Spark與Hadoop對(duì)比Spark如今已吸引了國內(nèi)外各大公司的注意，如騰訊、淘寶、百度、亞馬遜等公司均不同程度地使用了Spark來構(gòu)建大數(shù)據(jù)分析應(yīng)用，并應(yīng)用到實(shí)際的生產(chǎn)環(huán)境中16.1.2Scala簡(jiǎn)介Scala是一門現(xiàn)代的多范式編程語言，運(yùn)行于Java平臺(tái)（JVM，Java虛擬機(jī)），并兼容現(xiàn)有的Java程序Scala的特性：Scala具備強(qiáng)大的并發(fā)性，支持函數(shù)式編程，可以更好地支持分布式系統(tǒng)Scala語法簡(jiǎn)潔，能提供優(yōu)雅的APIScala兼容Java，運(yùn)行速度快，且能融合到Hadoop生態(tài)圈中Scala是Spark的主要編程語言，但Spark還支持Java、Python、R作為編程語言Scala的優(yōu)勢(shì)是提供了REPL（Read-Eval-PrintLoop，交互式解釋器），提高程序開發(fā)效率16.1.3Spark與Hadoop的對(duì)比Hadoop存在如下一些缺點(diǎn)：表達(dá)能力有限磁盤IO開銷大延遲高任務(wù)之間的銜接涉及IO開銷在前一個(gè)任務(wù)執(zhí)行完成之前，其他任務(wù)就無法開始，難以勝任復(fù)雜、多階段的計(jì)算任務(wù)16.1.3Spark與Hadoop的對(duì)比Spark在借鑒HadoopMapReduce優(yōu)點(diǎn)的同時(shí)，很好地解決了MapReduce所面臨的問題相比于HadoopMapReduce，Spark主要具有如下優(yōu)點(diǎn)：Spark的計(jì)算模式也屬于MapReduce，但不局限于Map和Reduce操作，還提供了多種數(shù)據(jù)集操作類型，編程模型比HadoopMapReduce更靈活Spark提供了內(nèi)存計(jì)算，可將中間結(jié)果放到內(nèi)存中，對(duì)于迭代運(yùn)算效率更高Spark基于DAG的任務(wù)調(diào)度執(zhí)行機(jī)制，要優(yōu)于HadoopMapReduce的迭代執(zhí)行機(jī)制16.1.3Spark與Hadoop的對(duì)比圖16-2Hadoop與Spark的執(zhí)行流程對(duì)比16.1.3Spark與Hadoop的對(duì)比圖16-3Hadoop與Spark執(zhí)行邏輯回歸的時(shí)間對(duì)比使用Hadoop進(jìn)行迭代計(jì)算非常耗資源Spark將數(shù)據(jù)載入內(nèi)存后，之后的迭代計(jì)算都可以直接使用內(nèi)存中的中間結(jié)果作運(yùn)算，避免了從磁盤中頻繁讀取數(shù)據(jù)16.2Spark生態(tài)系統(tǒng)在實(shí)際應(yīng)用中，大數(shù)據(jù)處理主要包括以下三個(gè)類型：復(fù)雜的批量數(shù)據(jù)處理：通常時(shí)間跨度在數(shù)十分鐘到數(shù)小時(shí)之間基于歷史數(shù)據(jù)的交互式查詢：通常時(shí)間跨度在數(shù)十秒到數(shù)分鐘之間基于實(shí)時(shí)數(shù)據(jù)流的數(shù)據(jù)處理：通常時(shí)間跨度在數(shù)百毫秒到數(shù)秒之間當(dāng)同時(shí)存在以上三種場(chǎng)景時(shí)，就需要同時(shí)部署三種不同的軟件比如:MapReduce/Impala/Storm這樣做難免會(huì)帶來一些問題：不同場(chǎng)景之間輸入輸出數(shù)據(jù)無法做到無縫共享，通常需要進(jìn)行數(shù)據(jù)格式的轉(zhuǎn)換不同的軟件需要不同的開發(fā)和維護(hù)團(tuán)隊(duì)，帶來了較高的使用成本比較難以對(duì)同一個(gè)集群中的各個(gè)系統(tǒng)進(jìn)行統(tǒng)一的資源協(xié)調(diào)和分配16.2Spark生態(tài)系統(tǒng)Spark的設(shè)計(jì)遵循“一個(gè)軟件棧滿足不同應(yīng)用場(chǎng)景”的理念，逐漸形成了一套完整的生態(tài)系統(tǒng)既能夠提供內(nèi)存計(jì)算框架，也可以支持SQL即席查詢、實(shí)時(shí)流式計(jì)算、機(jī)器學(xué)習(xí)和圖計(jì)算等Spark可以部署在資源管理器YARN之上，提供一站式的大數(shù)據(jù)解決方案因此，Spark所提供的生態(tài)系統(tǒng)足以應(yīng)對(duì)上述三種場(chǎng)景，即同時(shí)支持批處理、交互式查詢和流數(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)應(yīng)用場(chǎng)景時(shí)間跨度其他框架Spark生態(tài)系統(tǒng)中的組件復(fù)雜的批量數(shù)據(jù)處理小時(shí)級(jí)MapReduce、HiveSpark基于歷史數(shù)據(jù)的交互式查詢分鐘級(jí)、秒級(jí)Impala、Dremel、DrillSparkSQL基于實(shí)時(shí)數(shù)據(jù)流的數(shù)據(jù)處理毫秒、秒級(jí)Storm、S4SparkStreaming基于歷史數(shù)據(jù)的數(shù)據(jù)挖掘-MahoutMLlib圖結(jié)構(gòu)數(shù)據(jù)的處理-Pregel、HamaGraphX表1Spark生態(tài)系統(tǒng)組件的應(yīng)用場(chǎng)景16.3Spark運(yùn)行架構(gòu)16.3.1基本概念16.3.2架構(gòu)設(shè)計(jì)16.3.3Spark運(yùn)行基本流程16.3.4Spark運(yùn)行原理16.3.1基本概念RDD：是ResillientDistributedDataset（彈性分布式數(shù)據(jù)集）的簡(jiǎn)稱，是分布式內(nèi)存的一個(gè)抽象概念，提供了一種高度受限的共享內(nèi)存模型DAG：是DirectedAcyclicGraph（有向無環(huán)圖）的簡(jiǎn)稱，反映RDD之間的依賴關(guān)系Executor：是運(yùn)行在工作節(jié)點(diǎn)（WorkerNode）的一個(gè)進(jìn)程，負(fù)責(zé)運(yùn)行TaskApplication：用戶編寫的Spark應(yīng)用程序Task：運(yùn)行在Executor上的工作單元Job：一個(gè)Job包含多個(gè)RDD及作用于相應(yīng)RDD上的各種操作Stage：是Job的基本調(diào)度單位，一個(gè)Job會(huì)分為多組Task，每組Task被稱為Stage，或者也被稱為TaskSet，代表了一組關(guān)聯(lián)的、相互之間沒有Shuffle依賴關(guān)系的任務(wù)組成的任務(wù)集16.3.2架構(gòu)設(shè)計(jì)Spark運(yùn)行架構(gòu)包括集群資源管理器（ClusterManager）、運(yùn)行作業(yè)任務(wù)的工作節(jié)點(diǎn)（WorkerNode）、每個(gè)應(yīng)用的任務(wù)控制節(jié)點(diǎn)（Driver）和每個(gè)工作節(jié)點(diǎn)上負(fù)責(zé)具體任務(wù)的執(zhí)行進(jìn)程（Executor）資源管理器可以自帶或Mesos或YARN與HadoopMapReduce計(jì)算框架相比，Spark所采用的Executor有兩個(gè)優(yōu)點(diǎn)：一是利用多線程來執(zhí)行具體的任務(wù)，減少任務(wù)的啟動(dòng)開銷二是Executor中有一個(gè)BlockManager存儲(chǔ)模塊，會(huì)將內(nèi)存和磁盤共同作為存儲(chǔ)設(shè)備，有效減少IO開銷16.3.2架構(gòu)設(shè)計(jì)圖16-5Spark運(yùn)行架構(gòu)16.3.2架構(gòu)設(shè)計(jì)一個(gè)Application由一個(gè)Driver和若干個(gè)Job構(gòu)成，一個(gè)Job由多個(gè)Stage構(gòu)成，一個(gè)Stage由多個(gè)沒有Shuffle關(guān)系的Task組成當(dāng)執(zhí)行一個(gè)Application時(shí)，Driver會(huì)向集群管理器申請(qǐng)資源，啟動(dòng)Executor，并向Executor發(fā)送應(yīng)用程序代碼和文件，然后在Executor上執(zhí)行Task，運(yùn)行結(jié)束后，執(zhí)行結(jié)果會(huì)返回給Driver，或者寫到HDFS或者其他數(shù)據(jù)庫中16.3.2架構(gòu)設(shè)計(jì)圖16-6Spark中各種概念之間的相互關(guān)系16.3.3Spark運(yùn)行基本流程（1）首先為應(yīng)用構(gòu)建起基本的運(yùn)行環(huán)境，即由Driver創(chuàng)建一個(gè)SparkContext，進(jìn)行資源的申請(qǐng)、任務(wù)的分配和監(jiān)控（2）資源管理器為Executor分配資源，并啟動(dòng)Executor進(jìn)程（3）SparkContext根據(jù)RDD的依賴關(guān)系構(gòu)建DAG圖，DAG圖提交給DAGScheduler解析成Stage，然后把一個(gè)個(gè)TaskSet提交給底層調(diào)度器TaskScheduler處理；Executor向SparkContext申請(qǐng)Task，TaskScheduler將Task發(fā)放給Executor運(yùn)行，并提供應(yīng)用程序代碼（4）Task在Executor上運(yùn)行，把執(zhí)行結(jié)果反饋給TaskScheduler，然后反饋給DAGScheduler，運(yùn)行完畢后寫入數(shù)據(jù)并釋放所有資源

16.3.3Spark運(yùn)行基本流程圖16-7Spark運(yùn)行基本流程圖16.3.3Spark運(yùn)行基本流程總體而言，Spark運(yùn)行架構(gòu)具有以下特點(diǎn)：（1）每個(gè)Application都有自己專屬的Executor進(jìn)程，并且該進(jìn)程在Application運(yùn)行期間一直駐留。Executor進(jìn)程以多線程的方式運(yùn)行Task（2）Spark運(yùn)行過程與資源管理器無關(guān)，只要能夠獲取Executor進(jìn)程并保持通信即可（3）Task采用了數(shù)據(jù)本地性和推測(cè)執(zhí)行等優(yōu)化機(jī)制16.3.4RDD運(yùn)行原理1.設(shè)計(jì)背景2.RDD概念3.RDD特性4.RDD之間的依賴關(guān)系5.Stage的劃分6.RDD運(yùn)行過程16.3.4RDD運(yùn)行原理1.設(shè)計(jì)背景許多迭代式算法（比如機(jī)器學(xué)習(xí)、圖算法等）和交互式數(shù)據(jù)挖掘工具，共同之處是，不同計(jì)算階段之間會(huì)重用中間結(jié)果目前的MapReduce框架都是把中間結(jié)果寫入到HDFS中，帶來了大量的數(shù)據(jù)復(fù)制、磁盤IO和序列化開銷RDD就是為了滿足這種需求而出現(xiàn)的，它提供了一個(gè)抽象的數(shù)據(jù)架構(gòu)，我們不必?fù)?dān)心底層數(shù)據(jù)的分布式特性，只需將具體的應(yīng)用邏輯表達(dá)為一系列轉(zhuǎn)換處理，不同RDD之間的轉(zhuǎn)換操作形成依賴關(guān)系，可以實(shí)現(xiàn)管道化，避免中間數(shù)據(jù)存儲(chǔ)16.3.4RDD運(yùn)行原理2.RDD概念一個(gè)RDD就是一個(gè)分布式對(duì)象集合，本質(zhì)上是一個(gè)只讀的分區(qū)記錄集合，每個(gè)RDD可分成多個(gè)分區(qū)，每個(gè)分區(qū)就是一個(gè)數(shù)據(jù)集片段，并且一個(gè)RDD的不同分區(qū)可以被保存到集群中不同的節(jié)點(diǎn)上，從而可以在集群中的不同節(jié)點(diǎn)上進(jìn)行并行計(jì)算RDD提供了一種高度受限的共享內(nèi)存模型，即RDD是只讀的記錄分區(qū)的集合，不能直接修改，只能基于穩(wěn)定的物理存儲(chǔ)中的數(shù)據(jù)集創(chuàng)建RDD，或者通過在其他RDD上執(zhí)行確定的轉(zhuǎn)換操作（如map、join和groupby）而創(chuàng)建得到新的RDD16.3.4RDD運(yùn)行原理RDD提供了一組豐富的操作以支持常見的數(shù)據(jù)運(yùn)算，分為“動(dòng)作”（Action）和“轉(zhuǎn)換”（Transformation）兩種類型RDD提供的轉(zhuǎn)換接口都非常簡(jiǎn)單，都是類似map、filter、groupBy、join等粗粒度的數(shù)據(jù)轉(zhuǎn)換操作，而不是針對(duì)某個(gè)數(shù)據(jù)項(xiàng)的細(xì)粒度修改（不適合網(wǎng)頁爬蟲）表面上RDD的功能很受限、不夠強(qiáng)大，實(shí)際上RDD已經(jīng)被實(shí)踐證明可以高效地表達(dá)許多框架的編程模型（比如MapReduce、SQL、Pregel）Spark用Scala語言實(shí)現(xiàn)了RDD的API，程序員可以通過調(diào)用API實(shí)現(xiàn)對(duì)RDD的各種操作16.3.4RDD運(yùn)行原理RDD典型的執(zhí)行過程如下：RDD讀入外部數(shù)據(jù)源進(jìn)行創(chuàng)建RDD經(jīng)過一系列的轉(zhuǎn)換（Transformation）操作，每一次都會(huì)產(chǎn)生不同的RDD，供給下一個(gè)轉(zhuǎn)換操作使用最后一個(gè)RDD經(jīng)過“動(dòng)作”操作進(jìn)行轉(zhuǎn)換，并輸出到外部數(shù)據(jù)源這一系列處理稱為一個(gè)Lineage（血緣關(guān)系），即DAG拓?fù)渑判虻慕Y(jié)果優(yōu)點(diǎn)：惰性調(diào)用、管道化、避免同步等待、不需要保存中間結(jié)果、每次操作變得簡(jiǎn)單16.3.4RDD運(yùn)行原理圖16-8RDD執(zhí)行過程的一個(gè)實(shí)例動(dòng)作轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換轉(zhuǎn)換創(chuàng)建創(chuàng)建16.3.4RDD運(yùn)行原理Spark采用RDD以后能夠?qū)崿F(xiàn)高效計(jì)算的原因主要在于：（1）高效的容錯(cuò)性現(xiàn)有容錯(cuò)機(jī)制：數(shù)據(jù)復(fù)制或者記錄日志RDD：血緣關(guān)系、重新計(jì)算丟失分區(qū)、無需回滾系統(tǒng)、重算過程在不同節(jié)點(diǎn)之間并行、只記錄粗粒度的操作（2）中間結(jié)果持久化到內(nèi)存，數(shù)據(jù)在內(nèi)存中的多個(gè)RDD操作之間進(jìn)行傳遞，避免了不必要的讀寫磁盤開銷（3）存放的數(shù)據(jù)可以是Java對(duì)象，避免了不必要的對(duì)象序列化和反序列化3.RDD特性16.3.4RDD運(yùn)行原理窄依賴表現(xiàn)為一個(gè)父RDD的分區(qū)對(duì)應(yīng)于一個(gè)子RDD的分區(qū)或多個(gè)父RDD的分區(qū)對(duì)應(yīng)于一個(gè)子RDD的分區(qū)寬依賴則表現(xiàn)為存在一個(gè)父RDD的一個(gè)分區(qū)對(duì)應(yīng)一個(gè)子RDD的多個(gè)分區(qū)圖16-9窄依賴與寬依賴的區(qū)別4.RDD之間的依賴關(guān)系16.3.4RDD運(yùn)行原理Spark通過分析各個(gè)RDD的依賴關(guān)系生成了DAG，再通過分析各個(gè)RDD中的分區(qū)之間的依賴關(guān)系來決定如何劃分Stage，具體劃分方法是：在DAG中進(jìn)行反向解析，遇到寬依賴就斷開遇到窄依賴就把當(dāng)前的RDD加入到Stage中將窄依賴盡量劃分在同一個(gè)Stage中，可以實(shí)現(xiàn)流水線計(jì)算5.Stage的劃分16.3.4RDD運(yùn)行原理圖16-10根據(jù)RDD分區(qū)的依賴關(guān)系劃分Stage5.Stage的劃分被分成三個(gè)Stage，在Stage2中，從map到union都是窄依賴，這兩步操作可以形成一個(gè)流水線操作流水線操作實(shí)例分區(qū)7通過map操作生成的分區(qū)9，可以不用等待分區(qū)8到分區(qū)10這個(gè)map操作的計(jì)算結(jié)束，而是繼續(xù)進(jìn)行union操作，得到分區(qū)13，這樣流水線執(zhí)行大大提高了計(jì)算的效率16.3.4RDD運(yùn)行原理Stage的類型包括兩種：ShuffleMapStage和ResultStage，具體如下：（1）ShuffleMapStage：不是最終的Stage，在它之后還有其他Stage，所以，它的輸出一定需要經(jīng)過Shuffle過程，并作為后續(xù)Stage的輸入；這種Stage是以Shuffle為輸出邊界，其輸入邊界可以是從外部獲取數(shù)據(jù)，也可以是另一個(gè)ShuffleMapStage的輸出，其輸出可以是另一個(gè)Stage的開始；在一個(gè)Job里可能有該類型的Stage，也可能沒有該類型Stage；（2）ResultStage：最終的Stage，沒有輸出，而是直接產(chǎn)生結(jié)果或存儲(chǔ)。這種Stage是直接輸出結(jié)果，其輸入邊界可以是從外部獲取數(shù)據(jù)，也可以是另一個(gè)ShuffleMapStage的輸出。在一個(gè)Job里必定有該類型Stage。因此，一個(gè)Job含有一個(gè)或多個(gè)Stage，其中至少含有一個(gè)ResultStage。5.Stage的劃分16.3.4RDD運(yùn)行原理通過上述對(duì)RDD概念、依賴關(guān)系和Stage劃分的介紹，結(jié)合之前介紹的Spark運(yùn)行基本流程，再總結(jié)一下RDD在Spark架構(gòu)中的運(yùn)行過程：（1）創(chuàng)建RDD對(duì)象；（2）SparkContext負(fù)責(zé)計(jì)算RDD之間的依賴關(guān)系，構(gòu)建DAG；（3）DAGScheduler負(fù)責(zé)把DAG圖分解成多個(gè)Stage，每個(gè)Stage中包含了多個(gè)Task，每個(gè)Task會(huì)被TaskScheduler分發(fā)給各個(gè)WorkerNode上的Executor去執(zhí)行。16.3.4RDD運(yùn)行原理圖16-11RDD在Spark中的運(yùn)行過程16.4SparkSQL16.4.1從Shark說起16.4.2SparkSQL設(shè)計(jì)16.4.1從Shark說起Shark即HiveonSpark，為了實(shí)現(xiàn)與Hive兼容，Shark在HiveQL方面重用了Hive中HiveQL的解析、邏輯執(zhí)行計(jì)劃翻譯、執(zhí)行計(jì)劃優(yōu)化等邏輯，可以近似認(rèn)為僅將物理執(zhí)行計(jì)劃從MapReduce作業(yè)替換成了Spark作業(yè)，通過Hive的HiveQL解析，把HiveQL翻譯成Spark上的RDD操作。Shark的設(shè)計(jì)導(dǎo)致了兩個(gè)問題：一是執(zhí)行計(jì)劃優(yōu)化完全依賴于Hive，不方便添加新的優(yōu)化策略；二是因?yàn)镾park是線程級(jí)并行，而MapReduce是進(jìn)程級(jí)并行，因此，Spark在兼容Hive的實(shí)現(xiàn)上存在線程安全問題，導(dǎo)致Shark不得不使用另外一套獨(dú)立維護(hù)的打了補(bǔ)丁的Hive源碼分支Hive中SQL查詢的MapReduce作業(yè)轉(zhuǎn)化過程16.4.2SparkSQL設(shè)計(jì)SparkSQL在Hive兼容層面僅依賴HiveQL解析、Hive元數(shù)據(jù)，也就是說，從HQL被解析成抽象語法樹（AST）起，就全部由SparkSQL接管了。SparkSQL執(zhí)行計(jì)劃生成和優(yōu)化都由Catalyst（函數(shù)式關(guān)系查詢優(yōu)化框架）負(fù)責(zé)圖16-12SparkSQL架構(gòu)16.4.2SparkSQL設(shè)計(jì)圖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的部署和應(yīng)用方式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è)案例這種架構(gòu)部署較為繁瑣16.5.2從Hadoop+Storm架構(gòu)轉(zhuǎn)向Spark架構(gòu)用Spark架構(gòu)具有如下優(yōu)點(diǎn)：實(shí)現(xiàn)一鍵式安裝和配置、線程級(jí)別的任務(wù)監(jiān)控和告警降低硬件集群、軟件維護(hù)、任務(wù)監(jiān)控和應(yīng)用開發(fā)的難度便于做成統(tǒng)一的硬件、計(jì)算平臺(tái)資源池需要說明的是，SparkStreaming無法實(shí)現(xiàn)毫秒級(jí)的流計(jì)算，因此，對(duì)于需要毫秒級(jí)實(shí)時(shí)響應(yīng)的企業(yè)應(yīng)用而言，仍然需要采用流計(jì)算框架（如Storm）圖16-19用Spark架構(gòu)滿足批處理和流處理需求16.5.3Hadoop和Spark的統(tǒng)一部署不同的計(jì)算框架統(tǒng)一運(yùn)行在YARN中，可以帶來如下好處：計(jì)算資源按需伸縮不用負(fù)載應(yīng)用混搭，集群利用率高共享底層存儲(chǔ)，避免數(shù)據(jù)跨集群遷移圖16-20Hadoop和Spark的統(tǒng)一部署由于Hadoop生態(tài)系統(tǒng)中的一些組件所實(shí)現(xiàn)的功能，目前還是無法由Spark取代的，比如，Storm現(xiàn)有的Hadoop組件開發(fā)的應(yīng)用，完全轉(zhuǎn)移到Spark上需要一定的成本16.6Spark編程實(shí)踐16.6.1Spark安裝16.6.2啟動(dòng)SparkShell16.6.3SparkRDD基本操作16.6.4Spark應(yīng)用程序

16.6.1Spark安裝安裝Spark之前需要安裝Java環(huán)境和Hadoop環(huán)境。下載地址：

進(jìn)入下載頁面后，點(diǎn)擊主頁右側(cè)的“DownloadSpark”按鈕進(jìn)入下載頁面，下載頁面中提供了幾個(gè)下載選項(xiàng)，主要是Sparkrelease及Packagetype的選擇，如下圖所示。第1項(xiàng)Sparkrelease一般默認(rèn)選擇最新的發(fā)行版本，如截止至2016年3月份的最新版本為1.6.0。第2項(xiàng)packagetype則選擇“Pre-buildwithuser-providedHadoop[canusewithmostHadoopdistributions]”，可適用于多數(shù)Hadoop版本。選擇好之后，再點(diǎn)擊第4項(xiàng)給出的鏈接就可以下載Spark了。圖Spark下載選項(xiàng)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)容：

保存配置文件后，就可以啟動(dòng)、運(yùn)行Spark了。Spark包含多種運(yùn)行模式：?jiǎn)螜C(jī)模式、偽分布式模式、完全分布式模式。本章使用單機(jī)模式運(yùn)行Spark。若需要使用HDFS中的文件，則在使用Spark前需要啟動(dòng)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啟動(dòng)SparkShellSparkShell提供了簡(jiǎn)單的方式來學(xué)習(xí)SparkAPISparkShell可以以實(shí)時(shí)、交互的方式來分析數(shù)據(jù)SparkShell支持Scala和Python

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

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

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

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

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

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

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

假設(shè)我們只需要得到包含“Spark”的行數(shù)，那么存儲(chǔ)篩選后的文本數(shù)據(jù)是多余的，因?yàn)檫@部分?jǐn)?shù)據(jù)在計(jì)算得到行數(shù)后就不再使用到了。Spark基于整個(gè)操作鏈，僅儲(chǔ)存、計(jì)算所需的數(shù)據(jù)，提升了運(yùn)行效率。Scala>vallinesCountWithSpark=textFile.filter(line=>line.contains("Spark")).count()16.6.3SparkRDD基本操作Spark屬于MapReduce計(jì)算模型，因此也可以實(shí)現(xiàn)MapReduce的計(jì)算流程，如實(shí)現(xiàn)單詞統(tǒng)計(jì)，可以使用如下的命令實(shí)現(xiàn)：首先使用flatMap()將每一行的文本內(nèi)容通過空格進(jìn)行劃分為單詞；再使用map()將單詞映射為(K,V)的鍵值對(duì)，其中K為單詞，V為1；最后使用reduceByKey()將相同單詞的計(jì)數(shù)進(jìn)行相加，最終得到該單詞總的出現(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)計(jì)結(jié)果//Array[(String,Int)]=Array((package,1),(For,2),(Programs,1),(processing.,1),(Because,1),(The,1)...)16.6.4Spark應(yīng)用程序在SparkShell中進(jìn)行編程主要是方便對(duì)代碼進(jìn)行調(diào)試，但需要以逐行代碼的方式運(yùn)行。一般情況下，會(huì)選擇將調(diào)試后代碼打包成獨(dú)立的Spark應(yīng)用程序，提交到Spark中運(yùn)行。采用Scala編寫的程序需要使用sbt（SimpleBuildTool）進(jìn)行打包，sbt的安裝配置步驟如下：下載sbt-launch.jar（下載地址/s/1eRyFddw）將下載后的文件拷貝至安裝目錄/usr/local/sbt中，命令如下：在安裝目錄中創(chuàng)建一個(gè)Shell腳本文件（文件路徑：/usr/local/sbt/sbt）用于啟動(dòng)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)當(dā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