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

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

16.6.1Spark安裝安裝Spark之前需要安裝Java環境和Hadoop環境。下載地址：

進入下載頁面后，點擊主頁右側的“DownloadSpark”按鈕進入下載頁面，下載頁面中提供了幾個下載選項，主要是Sparkrelease及Packagetype的選擇，如下圖所示。第1項Sparkrelease一般默認選擇最新的發行版本，如截止至2016年3月份的最新版本為1.6.0。第2項packagetype則選擇“Pre-buildwithuser-providedHadoop[canusewithmostHadoopdistributions]”，可適用于多數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為系統用戶名

配置Spark的Classpath。

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

保存配置文件后，就可以啟動、運行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可以以實時、交互的方式來分析數據SparkShell支持Scala和Python

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

SparkRDD支持兩種類型的操作：動作（action）：在數據集上進行運算，返回計算值轉換（transformation）：基于現有的數據集創建一個新的數據集

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

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

輸出結果Long=95（“Long=95”表示該文件共有95行內容）。

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

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

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

輸出結果Long=95（“Long=95”表示該文件共有95行內容）。

Scala>valwordCounts=textFile.flatMap(line=>line.split("")).map(word=>(word,1)).reduceByKey((a,b)=>a+b)Scala>wordCounts.collect()//輸出單詞統計結果//Array[(String,Int)]=Array((package,1),(For,2),(Programs,1),(processing.,1),(Because,1),(The,1)...)16.6.4Spark應用程序在SparkShell中進行編程主要是方便對代碼進行調試，但需要以逐行代碼的方式運行。一般情況下，會選擇將調試后代碼打包成獨立的Spark應用程序，提交到Spark中運行。采用Scala編寫的程序需要使用sbt（SimpleBuildTool）進行打包，sbt的安裝配置步驟如下：下載sbt-launch.jar（下載地址/s/1eRyFddw）將下載后的文件拷貝至安裝目錄/usr/local/sbt中，命令如下：在安裝目錄中創建一個Shell腳本文件（文件路徑：/usr/local/sbt/sbt）用于啟動sbt，腳本文件中的代碼如下：保存后，還需要為該Shell腳本文件增加可執行權限，命令如下：sudomkdir/usr/local/sbt#創建安裝目錄cp~/下載/sbt-launch.jar/usr/local/sbtsudochown-Rhadoop/usr/local/sbt#此處的hadoop為系統當前用戶名#!/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