Spark大數(shù)據(jù)分析與實(shí)戰(zhàn)（Python+PySpark）課件 第５章 Spark編程進(jìn)階

第５章Spark編程進(jìn)階Spark大數(shù)據(jù)分析與實(shí)戰(zhàn)01搭建PySpark開(kāi)發(fā)環(huán)境理解RDD0203RDD緩存機(jī)制目錄CONTENTS04廣播變量和累加器05Spark生態(tài)和應(yīng)用程序架構(gòu)06Spark集群和應(yīng)用部署01搭建PySpark開(kāi)發(fā)環(huán)境PySpark編程環(huán)境JupyterNotebook編程環(huán)境PyCharm集成開(kāi)發(fā)環(huán)境PySpark編程環(huán)境在PySpark交互式編程環(huán)境中，只要輸入一條語(yǔ)句，就會(huì)自動(dòng)提交執(zhí)行并顯示運(yùn)行結(jié)果。在Python軟件庫(kù)中也存在一個(gè)名為pyspark的軟件包，可以脫離外部Spark運(yùn)行環(huán)境而獨(dú)立存在，對(duì)于第三方開(kāi)發(fā)工具如PyCharm、VScode等更加方便在第1章配置了pip3命令用于管理python3.8的軟件包，pip則被設(shè)置為管理python3.6的軟件包。也可以使用命令sudopython3.6-mpipinstallpyspark==2.4.8py4j==0.10.7PySpark編程環(huán)境Spark框架內(nèi)置的pyspark（PySparkShell）與Python環(huán)境中安裝的pyspark到底有什么區(qū)別，可以從以下3個(gè)方面理解

運(yùn)行環(huán)境不同。從pip源安裝的pyspark本質(zhì)是一個(gè)Python擴(kuò)展庫(kù)，它可在Python交互式編程環(huán)境或任何支持Python的集成開(kāi)發(fā)環(huán)境中引入和使用。Spark框架內(nèi)置的pyspark只能通過(guò)運(yùn)行pyspark命令腳本來(lái)啟動(dòng)，它實(shí)際是一個(gè)工具命令程序

功能不同。從pip源安裝的pyspark庫(kù)僅限于在Python環(huán)境下使用，需要在代碼中通過(guò)import導(dǎo)入。Spark框架則提供了多語(yǔ)言版本的交互式編程環(huán)境，包括pyspark（Python版）、spark-shell（Scala版，默認(rèn)）和sparkR（R版）等多種交互式編程工具，它們都是Spark框架提供的命令工具，所以統(tǒng)稱(chēng)為SparkShell，而pyspark只是其中之一

使用方式不同。從pip源安裝的pyspark庫(kù)在導(dǎo)入后，需要手動(dòng)創(chuàng)建spark和sc入口對(duì)象變量，而Spark框架自帶的pyspark命令啟動(dòng)后，會(huì)自動(dòng)創(chuàng)建spark和sc入口對(duì)象變量，對(duì)初學(xué)者來(lái)說(shuō)更為方便易用PySpark編程環(huán)境安裝好pyspark庫(kù)，就可以在各種Python的編程環(huán)境中使用這里必須手動(dòng)創(chuàng)建SparkContext對(duì)象才能進(jìn)行后續(xù)操作，其他Python開(kāi)發(fā)工具使用pyspark庫(kù)時(shí)也是如此JupyterNotebook編程環(huán)境JupyterNotebook是一個(gè)類(lèi)似網(wǎng)頁(yè)筆記形式的Web編程工具，支持在網(wǎng)頁(yè)中直接編寫(xiě)和運(yùn)行代碼，并能夠以文本、圖表嵌入等方式輸出運(yùn)行結(jié)果，適用于數(shù)據(jù)清洗、數(shù)據(jù)可視化、機(jī)器學(xué)習(xí)等場(chǎng)合安裝findspark庫(kù)，它的作用是使pyspark庫(kù)在JupyterNotebook網(wǎng)頁(yè)中能夠像一個(gè)普通Python模塊那樣導(dǎo)入和使用JupyterNotebook編程環(huán)境稍候?yàn)g覽器中會(huì)顯示JupyterNotebook的初始界面找到瀏覽器頁(yè)面右上角區(qū)域，選擇New下拉列表中的Python3選項(xiàng)，此時(shí)會(huì)新打開(kāi)一個(gè)JupyterNotebook頁(yè)面，Spark代碼就是在這個(gè)頁(yè)面中來(lái)編寫(xiě)的JupyterNotebook編程環(huán)境在JupyterNotebook頁(yè)面中編寫(xiě)代碼的方式，與在普通的Python編程環(huán)境中基本一樣，唯一不同的是，在執(zhí)行代碼之前，必須先運(yùn)行一次findspark.init()方法，只需在當(dāng)前JupyterNotebook頁(yè)面中執(zhí)行一次即可，不用重復(fù)執(zhí)行JupyterNotebook編程環(huán)境JupyterNotebook頁(yè)面是由一些稱(chēng)為Cell的格子行構(gòu)成的，用戶(hù)可以像在普通文檔中一樣在格子里編寫(xiě)代碼或文本內(nèi)容當(dāng)Notebook頁(yè)面的Cell行代碼在運(yùn)行時(shí)，當(dāng)前Cell行的左端會(huì)有一個(gè)“[*]”提示，瀏覽器背后的Linux終端也會(huì)顯示一些日志信息，在遇到問(wèn)題時(shí)這可以作為一個(gè)參考的線索PyCharm集成開(kāi)發(fā)環(huán)境PyCharm就是一個(gè)功能強(qiáng)大的跨平臺(tái)開(kāi)發(fā)環(huán)境，主要用于Python的開(kāi)發(fā)，支持代碼分析、圖形化調(diào)試，集成測(cè)試器、集成版本控制等特性，分為社區(qū)版和專(zhuān)業(yè)版兩種，這里使用的是免費(fèi)的社區(qū)版PyCharm集成開(kāi)發(fā)環(huán)境選擇PyCharm初始窗體左側(cè)的Projects選項(xiàng)，然后單擊右側(cè)的NewProject按鈕啟動(dòng)新建Python項(xiàng)目，項(xiàng)目創(chuàng)建完畢，PyCharm會(huì)切換至項(xiàng)目開(kāi)發(fā)窗體PyCharm集成開(kāi)發(fā)環(huán)境在打開(kāi)的代碼編輯器中將main.py文件的原有代碼全部清除，輸入下面的Spark測(cè)試代碼與PySparkShell、Python、JupyterNotebook等交互式編程環(huán)境不同的是，PyCharm在運(yùn)行代碼時(shí)不能直接通過(guò)變量名將它們的內(nèi)容打印顯示，必須使用print()函數(shù)PyCharm集成開(kāi)發(fā)環(huán)境代碼準(zhǔn)備完畢，現(xiàn)在可以在PyCharm的源代碼編輯器任意空白位置單擊鼠標(biāo)右鍵，在彈出的快捷菜單中選擇Run'×××'命令（×××代表要運(yùn)行的文件名），就會(huì)啟動(dòng)main.py程序的執(zhí)行02理解RDDRDD基本概念RDD的分區(qū)RDD的依賴(lài)關(guān)系RDD的計(jì)算調(diào)度RDD基本概念RDD彈性分布式數(shù)據(jù)集是一種容錯(cuò)的、可被并行操作的數(shù)據(jù)集合，它允許數(shù)據(jù)同時(shí)跨多個(gè)計(jì)算節(jié)點(diǎn)而存在，但在使用上仍像是一個(gè)普通的集合，Spark底層的計(jì)算過(guò)程都是圍繞RDD進(jìn)行的，RDD計(jì)算流程包括初始化RDD、若干次Transformation轉(zhuǎn)換操作、Action行動(dòng)操作RDD的分區(qū)分區(qū)（Partition）是RDD的一種內(nèi)部數(shù)據(jù)組織單位，就像班級(jí)中的小組一樣。RDD數(shù)據(jù)集是由分布在各個(gè)節(jié)點(diǎn)上的分區(qū)構(gòu)成的，RDD通過(guò)分區(qū)機(jī)制將數(shù)據(jù)分散到不同節(jié)點(diǎn)上進(jìn)行計(jì)算RDD的分區(qū)數(shù)會(huì)影響到Spark計(jì)算任務(wù)Task線程的數(shù)量，每個(gè)Task線程負(fù)責(zé)處理一個(gè)或多個(gè)分區(qū)的數(shù)據(jù)。RDD分區(qū)數(shù)對(duì)Spark應(yīng)用程序的運(yùn)行是有影響的，分區(qū)數(shù)太少或太多都不合適，同時(shí)還可以在運(yùn)行時(shí)動(dòng)態(tài)進(jìn)行調(diào)整，包括coalesce()和repartition()兩種調(diào)整方法RDD的依賴(lài)關(guān)系RDD在每次進(jìn)行轉(zhuǎn)換操作后，都會(huì)生成一個(gè)新的RDD，這樣經(jīng)過(guò)多次轉(zhuǎn)換操作生成的RDD之間就會(huì)形成一條類(lèi)似“流水線”一樣的前后鏈接關(guān)系，后一個(gè)RDD（子）是依賴(lài)于前面的一個(gè)或多個(gè)RDD（父）生成的RDD的前后依賴(lài)關(guān)系被劃分為兩種類(lèi)型，分別是“窄依賴(lài)”（NarrowDependency）和“寬依賴(lài)”（WideDependency，或ShuffleDependency）。RDD之間的依賴(lài)關(guān)系，是以RDD里面的分區(qū)數(shù)據(jù)“前后流向”關(guān)系為依據(jù)的窄依賴(lài)是指父RDD中任意一個(gè)分區(qū)的數(shù)據(jù)，至多“流向”子RDD的一個(gè)分區(qū)，前后RDD的分區(qū)之間是“多對(duì)一”的關(guān)系（n∶1，n≥1）在窄依賴(lài)情形下，如果子RDD在執(zhí)行操作時(shí)某個(gè)分區(qū)計(jì)算失敗（如數(shù)據(jù)丟失），此時(shí)只需重新計(jì)算父RDD的對(duì)應(yīng)分區(qū)即可恢復(fù)數(shù)據(jù)。當(dāng)然，如果父RDD的分區(qū)數(shù)據(jù)也有受損，則再往上追溯計(jì)算即可，像map、filter等轉(zhuǎn)換算子生成的RDD，它們之間都是窄依賴(lài)的關(guān)系RDD的依賴(lài)關(guān)系寬依賴(lài)是指父RDD中分區(qū)的數(shù)據(jù)會(huì)同時(shí)“流向”子RDD的多個(gè)分區(qū)，前后RDD的分區(qū)之間是“一對(duì)多”的關(guān)系（1∶n，n≥2）在執(zhí)行具有寬依賴(lài)關(guān)系的算子時(shí)，來(lái)自父RDD的不同分區(qū)的數(shù)據(jù)要進(jìn)行“洗牌”重組，所以寬依賴(lài)也被稱(chēng)為Shuffle依賴(lài)。當(dāng)前后RDD之間是寬依賴(lài)關(guān)系時(shí)，前面RDD的分區(qū)數(shù)據(jù)需要打散，通過(guò)計(jì)算后再重新組合到后面RDD的分區(qū)中（回顧一下詞頻統(tǒng)計(jì)的例子）。在這種情形下，如果子RDD丟失了分區(qū)數(shù)據(jù)，則需要將父RDD的分區(qū)全部重新計(jì)算一次才能恢復(fù)，像reduceByKey、sortByKey等操作就會(huì)產(chǎn)生寬依賴(lài)關(guān)系RDD的依賴(lài)關(guān)系下面是一個(gè)簡(jiǎn)單的演示RDD寬窄依賴(lài)關(guān)系的示例RDD的依賴(lài)關(guān)系RDD的計(jì)算調(diào)度Spark應(yīng)用程序要落實(shí)到具體計(jì)算節(jié)點(diǎn)上去運(yùn)行，還有很多具體工作要做。當(dāng)用戶(hù)代碼提交給Driver進(jìn)程時(shí)，Spark是如何將這些代碼分配到Executor進(jìn)程上去執(zhí)行的。由于Spark的核心設(shè)計(jì)是RDD，因此數(shù)據(jù)的計(jì)算都是圍繞RDD為中心展開(kāi)的。為此，Spark引入了Job（作業(yè)）、Stage（階段）和Task（任務(wù)）等幾個(gè)與計(jì)算調(diào)度相關(guān)的內(nèi)容（1）Job作業(yè)RDD的轉(zhuǎn)換操作采用延遲機(jī)制，在調(diào)用Transformation算子時(shí)并不會(huì)啟動(dòng)運(yùn)行它，直至遇到Action算子才會(huì)啟動(dòng)運(yùn)行。在這種情形下，Spark會(huì)將Action操作和之前形成的一系列RDD計(jì)算鏈構(gòu)成一個(gè)完整的Job（作業(yè)），然后將這個(gè)Job安排到節(jié)點(diǎn)上運(yùn)行，這也是RDD延遲機(jī)制的基本實(shí)現(xiàn)原理。在一個(gè)Job中，通常包含1～n個(gè)Transformation操作和1個(gè)Action操作Job的劃分只是第1步，每個(gè)Job還要再繼續(xù)進(jìn)行分解，以最終生成一系列能夠并行執(zhí)行的Task線程（Spark集群的最小執(zhí)行單元）。為了提高計(jì)算節(jié)點(diǎn)的利用率和執(zhí)行效率，Spark還會(huì)進(jìn)一步對(duì)這些Task進(jìn)行編組，形成相互獨(dú)立的Stage。一個(gè)Stage包含一組Task線程，即TaskSet。經(jīng)過(guò)這樣的處理，這些被精心安排的Task代碼和RDD分區(qū)數(shù)據(jù)就通過(guò)Driver進(jìn)程分發(fā)到各個(gè)節(jié)點(diǎn)的Executor進(jìn)程中執(zhí)行RDD的計(jì)算調(diào)度（1）Job作業(yè)Spark將Action操作和之前形成的一系列RDD計(jì)算鏈構(gòu)成一個(gè)完整的Job（作業(yè)）RDD的計(jì)算調(diào)度（1）Job作業(yè)從Spark提供的WebUI管理頁(yè)面中可以查看應(yīng)用程序在運(yùn)行過(guò)程中生成的Job（作業(yè)），只需在Ubuntu20.04的瀏覽器地址欄中輸入http://localhost:4040即可（每個(gè)Spark應(yīng)用程序都有一個(gè)這樣的狀態(tài)頁(yè)面，端口號(hào)依次為4040、4041、4042等）RDD的計(jì)算調(diào)度（2）Stage階段當(dāng)用戶(hù)代碼在第1步被劃分成Job后，第2步就是在Job里面劃分Stage。Stage一詞的本意是“階段”，也可將其理解為“片段”，它是Spark計(jì)算調(diào)度的一個(gè)重要環(huán)節(jié)，這是因?yàn)镴ob的運(yùn)行粒度還不夠細(xì)，Job里面的轉(zhuǎn)換算子生成的前后RDD存在不同的寬窄依賴(lài)關(guān)系，這些RDD的依賴(lài)關(guān)系會(huì)影響到RDD的計(jì)算順序在對(duì)Job包含的轉(zhuǎn)換操作劃分Stage時(shí)，Spark是按照倒序并參考RDD的依賴(lài)關(guān)系（寬依賴(lài)/窄依賴(lài)）進(jìn)行的。在Job中，從第1次出現(xiàn)的Action（行動(dòng)）操作往回倒推，遇到窄依賴(lài)關(guān)系的操作將其劃分到同一個(gè)Stage中，遇到寬依賴(lài)關(guān)系的操作則劃分出一個(gè)新的Stage，依次往前類(lèi)推不同Stage之間形成一連串的前后關(guān)系，但因?yàn)槲覀兪前凑盏雇频捻樞蜻M(jìn)行劃分的，所以先劃分出來(lái)的被稱(chēng)為“子Stage”，后劃分出來(lái)的被稱(chēng)為“父Stage”。經(jīng)過(guò)這樣的劃分后，就可以從最前面的父Stage開(kāi)始調(diào)度，依次向后執(zhí)行，“子Stage”需要等所有“父Stage”運(yùn)行完才能執(zhí)行。這種不同Stage之間按照先后順序劃分出來(lái)的依賴(lài)關(guān)系，構(gòu)成了一個(gè)網(wǎng)狀圖形，它就是有向無(wú)環(huán)圖（DirectedAcyclicGraph,DAG）RDD的計(jì)算調(diào)度（2）Stage階段下面通過(guò)實(shí)際的示例代碼來(lái)看一下Spark是如何在Job中劃分Stage的RDD的計(jì)算調(diào)度（2）Stage階段在Spark的WebUI管理頁(yè)面中，點(diǎn)擊Job列表中的相應(yīng)鏈接即可查看其所包含的Stage點(diǎn)擊Stage列表中的相應(yīng)鏈接可查看每個(gè)Stage的有向無(wú)環(huán)圖RDD的計(jì)算調(diào)度（3）Task任務(wù)Spark按照Action（行動(dòng)）操作將用戶(hù)代碼分成若干個(gè)Job，每個(gè)Job再按RDD計(jì)算鏈的寬窄依賴(lài)關(guān)系進(jìn)一步劃分為若干個(gè)Stage，以生成更細(xì)致的執(zhí)行計(jì)劃。經(jīng)過(guò)這種由粗到細(xì)的逐層劃分，最后一個(gè)環(huán)節(jié)就是創(chuàng)建Task（任務(wù)）以執(zhí)行代碼RDD的計(jì)算調(diào)度（3）Task任務(wù)RDD的計(jì)算調(diào)度（3）Task任務(wù)在Spark的WebUI管理頁(yè)面，在Stage對(duì)應(yīng)的有向無(wú)環(huán)圖的頁(yè)面底部可以找到該Stage包含的Task線程列表（Task線程所在的Executor進(jìn)程也在這里）RDD的計(jì)算調(diào)度Spark應(yīng)用程序在運(yùn)行過(guò)程中的計(jì)算調(diào)度實(shí)際是在Driver進(jìn)程中完成的，具體則是由Driver進(jìn)程的SparkContext、DAGScheduler、TaskScheduler等組件來(lái)負(fù)責(zé)用戶(hù)代碼提交后，Spark會(huì)啟動(dòng)一個(gè)Driver進(jìn)程并創(chuàng)建SparkContext，然后以RDD的行動(dòng)操作為邊界將用戶(hù)代碼劃分為若干個(gè)Job。在每個(gè)Job中，DAGScheduler根據(jù)RDD前后寬窄依賴(lài)關(guān)系建立有向無(wú)環(huán)圖，并將有向無(wú)環(huán)圖拆分為多個(gè)Task編組（TaskSet），每組Task就被封裝為一個(gè)Stage，以TaskSet的形式提交給底層的TaskScheduler調(diào)度器執(zhí)行具體的代碼RDD的計(jì)算調(diào)度Spark應(yīng)用程序Application在提交運(yùn)行時(shí)，需要啟動(dòng)Driver和Executor進(jìn)程，在調(diào)度時(shí)還涉及Job、Stage、Task等幾個(gè)概念，它們之間的關(guān)系可以從物理、邏輯這兩個(gè)緯度進(jìn)行理解03RDD緩存機(jī)制RDD緩存機(jī)制RDD緩存機(jī)制在Spark編程時(shí)經(jīng)常會(huì)遇到這樣的情況：經(jīng)過(guò)一系列計(jì)算得到一個(gè)RDD數(shù)據(jù)集，這個(gè)RDD數(shù)據(jù)集在后續(xù)的計(jì)算中還會(huì)被使用到。由于RDD是一個(gè)過(guò)程性的數(shù)據(jù)集，不是一直停留在內(nèi)存中的，如果每次都要重復(fù)計(jì)算生成這個(gè)RDD，則可能要耗費(fèi)大量時(shí)間。為避免這種情形的發(fā)生，可以通過(guò)Spark的緩存機(jī)制來(lái)解決RDD緩存機(jī)制Spark提供了一個(gè)很重要的功能，能將計(jì)算過(guò)程中生成的數(shù)據(jù)進(jìn)行持久化緩存，這樣后面的操作就可以直接訪問(wèn)緩存的數(shù)據(jù)，而不用每次沿著RDD計(jì)算鏈再執(zhí)行一遍。當(dāng)一個(gè)RDD被持久化時(shí)，每個(gè)節(jié)點(diǎn)上的線程都可以直接使用這個(gè)緩存的RDD進(jìn)行計(jì)算RDD的持久化還可以使用不同的存儲(chǔ)級(jí)別，只需傳遞一個(gè)StorageLevel參數(shù)即可。常用的RDD持久化存儲(chǔ)級(jí)別主要包括以下形式RDD緩存機(jī)制Spark還支持使用checkpoint（檢查點(diǎn)）機(jī)制進(jìn)行緩存，類(lèi)似于快照，其作用就是將重要的或需要長(zhǎng)時(shí)間保存的中間數(shù)據(jù)設(shè)置一個(gè)檢查點(diǎn)，保存到高可用的存儲(chǔ)系統(tǒng)上。比如，使用checkpoint保存在HDFS上。不過(guò)，由于checkpoint是將RDD數(shù)據(jù)存儲(chǔ)到磁盤(pán)上，因此執(zhí)行效率相比內(nèi)存緩存來(lái)說(shuō)要低。另外，checkpoint保存的數(shù)據(jù)如果不手動(dòng)刪除會(huì)一直存在，而緩存是隨著Spark應(yīng)用程序結(jié)束就失效了04廣播變量和累加器廣播變量累加器RDD緩存機(jī)制廣播變量（BroadcastVariables）是指將數(shù)據(jù)從一個(gè)節(jié)點(diǎn)廣播發(fā)送到其他節(jié)點(diǎn)上。比如在Driver端有一張內(nèi)存數(shù)據(jù)表，其他節(jié)點(diǎn)運(yùn)行的Executor進(jìn)程需要查詢(xún)這張表中的數(shù)據(jù)，那么Driver會(huì)先把這張表發(fā)送到其他節(jié)點(diǎn)上，這樣Executor進(jìn)程的每個(gè)Task計(jì)算任務(wù)就可以在本地節(jié)點(diǎn)上查表RDD緩存機(jī)制Spark的廣播變量被設(shè)計(jì)為只讀，這樣就能保證每個(gè)計(jì)算任務(wù)使用的變量都是一樣的值。廣播變量可以通過(guò)SparkContext.broadcast(v)方法從一個(gè)本地變量v進(jìn)行創(chuàng)建，廣播變量在使用上就相當(dāng)于對(duì)普通變量v的一個(gè)包裝，它的值可以使用value()方法進(jìn)行訪問(wèn)這個(gè)例子中，map算子需要使用變量a，因此Driver進(jìn)程需要分別將變量a發(fā)送給各Task線程。如果線程有4個(gè)，那么變量a在節(jié)點(diǎn)上就同時(shí)存在4個(gè)副本RDD緩存機(jī)制如果改用廣播變量，則只需要將變量a發(fā)送到節(jié)點(diǎn)的Executor進(jìn)程中即可，這樣Executor進(jìn)程中的各個(gè)線程就可以共享a這個(gè)廣播變量。我們將上面的代碼稍作修改上面的代碼，rdd1被劃分為4個(gè)分區(qū)，假定有兩個(gè)節(jié)點(diǎn)，每個(gè)節(jié)點(diǎn)運(yùn)行兩個(gè)Task線程，那么總共有4個(gè)Task分別處理rdd1的4個(gè)分區(qū)數(shù)據(jù)。如果改成rdd2=rdd1.map(lambdax:x-ba.value)（粗體標(biāo)識(shí)的ba是在Driver進(jìn)程中定義的一個(gè)廣播變量），此時(shí)Driver進(jìn)程只需將ba的數(shù)據(jù)分發(fā)給兩個(gè)節(jié)點(diǎn)的Executor進(jìn)程，每個(gè)節(jié)點(diǎn)的Task計(jì)算任務(wù)共享同一份ba數(shù)據(jù)，總共只需發(fā)送兩份ba變量的內(nèi)容即可累加器我們來(lái)看一個(gè)簡(jiǎn)單的例子，這個(gè)例子是按照常規(guī)程序運(yùn)行的思路編寫(xiě)從運(yùn)行結(jié)果容易看出，num變量的累加結(jié)果仍是初始值10，并未把rdd的數(shù)據(jù)元素加進(jìn)來(lái)，這也是分布式計(jì)算引起的一個(gè)典型負(fù)面效果累加器Spark的累加器（Accumulators）是為了支持在集群的不同節(jié)點(diǎn)中進(jìn)行累加計(jì)算的一種手段，比如計(jì)數(shù)、求和，或者集合元素匯總等，它能實(shí)現(xiàn)在Driver端和Executor端的共享變量“寫(xiě)”的功能Spark內(nèi)置了3種類(lèi)型的累加器，分別是LongAccumulator、DoubleAccumulator和CollectionAccumulator。其中，LongAccumulator用來(lái)累加整數(shù)，DoubleAccumulator用來(lái)累加浮點(diǎn)數(shù)，CollectionAccumulator用來(lái)合并集合數(shù)據(jù)。如果Spark內(nèi)置的累加器無(wú)法滿足需求，用戶(hù)還可以自定義累加器累加器累加器變量可以調(diào)用SparkContext.accumulator(v)方法設(shè)置，其中，參數(shù)v是來(lái)自Driver進(jìn)程的普通變量，它可以是整數(shù)、浮點(diǎn)數(shù)或集合，或者符合累加器要求的其他自定義類(lèi)型。累加器變量有一個(gè)名為value的屬性，它存儲(chǔ)了所有節(jié)點(diǎn)返回Driver進(jìn)程的結(jié)果匯總數(shù)據(jù)。另外，累加器的value屬性值獲取操作只能在Driver端的代碼中執(zhí)行05Spark生態(tài)和應(yīng)用程序架構(gòu)Spark生態(tài)架構(gòu)Spark應(yīng)用程序架構(gòu)Spark生態(tài)架構(gòu)Spark致力于成為一個(gè)開(kāi)源的大數(shù)據(jù)處理、數(shù)據(jù)科學(xué)、機(jī)器學(xué)習(xí)和數(shù)據(jù)分析工作的統(tǒng)一引擎，以SparkCore為核心，支持HDFS等多種分布式存儲(chǔ)系統(tǒng)，通過(guò)SparkStandalone、YARN等集群管理器實(shí)現(xiàn)分布式的計(jì)算。Spark生態(tài)包含SparkCore、SparkSQL、SparkStreaming、MLlib、GraphX四大組件SparkCore是Spark框架的核心模塊，實(shí)現(xiàn)了Spark的基本功能，包含任務(wù)調(diào)度、內(nèi)存管理、錯(cuò)誤恢復(fù)、與存儲(chǔ)系統(tǒng)的交互等。SparkCore的實(shí)現(xiàn)依賴(lài)于RDD（彈性分布式數(shù)據(jù)集）這一抽象概念，定義了RDD的相關(guān)API接口Spark生態(tài)架構(gòu)SparkSQL是Spark用來(lái)處理結(jié)構(gòu)化數(shù)據(jù)的模塊，借助SparkSQL可以構(gòu)造SQL語(yǔ)句直接查詢(xún)數(shù)據(jù)，在底層會(huì)將SQL變成一系列分布執(zhí)行的RDD計(jì)算任務(wù)SparkStreaming模塊主要是針對(duì)實(shí)時(shí)流數(shù)據(jù)的處理，SparkStreaming把輸入數(shù)據(jù)按照設(shè)定的時(shí)間片切分為一段一段的數(shù)據(jù)，每段數(shù)據(jù)都被轉(zhuǎn)換成RDD數(shù)據(jù)集，將DStream的轉(zhuǎn)換操作轉(zhuǎn)變?yōu)閷?duì)RDD的轉(zhuǎn)換操作，生成的中間結(jié)果保存在內(nèi)存中MLlib模塊主要用于機(jī)器學(xué)習(xí)領(lǐng)域，它實(shí)現(xiàn)了一系列常用的機(jī)器學(xué)習(xí)和統(tǒng)計(jì)算法，如分類(lèi)、回歸、聚類(lèi)、主成分分析等，還提供了模型評(píng)估、數(shù)據(jù)導(dǎo)入等額外的功能GraphX模塊主要支持圖數(shù)據(jù)的分析和計(jì)算，包含了許多廣泛應(yīng)用的圖計(jì)算算法，比如最短路徑、網(wǎng)頁(yè)排名、最小切割、連通分支等。與其他分布式圖計(jì)算框架相比，GraphX的最大貢獻(xiàn)是在Spark之上提供了一站式的數(shù)據(jù)解決方案，可以方便高效地完成圖計(jì)算的一整套流水作業(yè)Spark應(yīng)用程序架構(gòu)Spark本身主要是由Scala開(kāi)發(fā)的，也包含部分使用Java開(kāi)發(fā)的模塊，運(yùn)行在JVM虛擬機(jī)上。Spark對(duì)Python的支持是通過(guò)一個(gè)Py4j庫(kù)實(shí)現(xiàn)的，這是一個(gè)功能強(qiáng)大的RPC（RemoteProcedureCall，遠(yuǎn)程過(guò)程調(diào)用）庫(kù)，其實(shí)現(xiàn)了Python和JVM之間的交互，可以讓Python自由操縱JavaPySpark不僅可以編寫(xiě)Spark封裝的API接口代碼，還可以編寫(xiě)Python庫(kù)的代碼，不過(guò)Python代碼只能在Python自己的運(yùn)行環(huán)境中運(yùn)行。借助Pipe管道機(jī)制，Spark會(huì)將Python庫(kù)的代碼發(fā)送到Python運(yùn)行環(huán)境中運(yùn)行并獲取返回結(jié)果，整個(gè)過(guò)程就像調(diào)用一個(gè)Spark本地函數(shù)一樣Spark應(yīng)用程序架構(gòu)在一個(gè)集群環(huán)境中，可以同時(shí)運(yùn)行多個(gè)Spark應(yīng)用程序。所謂Spark應(yīng)用程序，是指包括用戶(hù)代碼、Driver和Executor在內(nèi)的若干相關(guān)進(jìn)程，它是一組動(dòng)態(tài)的概念。一個(gè)集群環(huán)境可以同時(shí)運(yùn)行多個(gè)Spark應(yīng)用，每個(gè)應(yīng)用都有各自的一套進(jìn)程，不同應(yīng)用之間的進(jìn)程互相隔離Spark應(yīng)用程序架構(gòu)Spark的應(yīng)用程序架構(gòu)涉及很多專(zhuān)業(yè)術(shù)語(yǔ)，初期理解起來(lái)可能會(huì)存在一些困難，但以后掌握它們的話將有助于加深對(duì)Spark框架體系的理解Spark應(yīng)用程序架構(gòu)Spark的應(yīng)用程序架構(gòu)涉及很多專(zhuān)業(yè)術(shù)語(yǔ)，初期理解起來(lái)可能會(huì)存在一些困難，但以后掌握它們的話將有助于加深對(duì)Spark框架體系的理解Spark應(yīng)用程序架構(gòu)Spark應(yīng)用程序的運(yùn)行可在本地或集群上運(yùn)行。Spark本地運(yùn)行方式比較簡(jiǎn)單，一般只在開(kāi)發(fā)測(cè)試下使用，它通過(guò)在本地計(jì)算機(jī)的一個(gè)Java進(jìn)程中同時(shí)啟動(dòng)Driver和Executor以實(shí)現(xiàn)計(jì)算任務(wù)的本地運(yùn)行。Spark集群運(yùn)行方式支持SparkStandalone、SparkonYARN、SparkonMesos等集群環(huán)境，每個(gè)Spark應(yīng)用程序都包括一個(gè)Driver進(jìn)程、一個(gè)或多個(gè)Executor進(jìn)程當(dāng)在集群節(jié)點(diǎn)上時(shí)，Driver進(jìn)程運(yùn)行在集群節(jié)點(diǎn)上被稱(chēng)為cluster模式，Driver進(jìn)程運(yùn)行在集群之外的計(jì)算機(jī)上時(shí)被稱(chēng)為client模式，此時(shí)提交任務(wù)的客戶(hù)端計(jì)算機(jī)相當(dāng)于臨時(shí)加入集群環(huán)境中Spark生態(tài)架構(gòu)SparkStreaming模塊組件05Spark集群和應(yīng)用部署Spark偽分布集群的搭建Spark應(yīng)用程序部署模式Spark應(yīng)用程序部署實(shí)例Spark偽分布集群的搭建我們已經(jīng)配置了單機(jī)偽分布的Hadoop環(huán)境，下面繼續(xù)在此基礎(chǔ)上配置Spark的偽分布集群，使用的是自帶的SparkStandalone集群管理器。在后續(xù)內(nèi)容中，還會(huì)介紹使用3臺(tái)Linux虛擬機(jī)來(lái)搭建完全分布式的Spark集群（2）因?yàn)楦牧酥鳈C(jī)名，所以還需要修改/etc/hosts文件，在這個(gè)文件中設(shè)定了IP地址與主機(jī)名的對(duì)應(yīng)關(guān)系，類(lèi)似DNS域名服務(wù)器的功能（1）首先查看當(dāng)前虛擬機(jī)的默認(rèn)名稱(chēng)，然后將其修改為vm01Spark偽分布集群的搭建（3）修改Spark的相關(guān)配置文件，包括spark-env.sh和slave這兩個(gè)文件Spark偽分布集群的搭建（3）修改Spark的相關(guān)配置文件，包括spark-env.sh和slave這兩個(gè)文件Spark偽分布集群的搭建（4）使用start-all.sh腳本啟動(dòng)Spark集群的Master主節(jié)點(diǎn)和Worker工作節(jié)點(diǎn)，并通過(guò)jps命令檢查Master和Worker兩個(gè)進(jìn)程是否存在Spark偽分布集群的搭建（5）在Ubuntu20.04虛擬機(jī)中打開(kāi)瀏覽器訪問(wèn)Spark集群的WebUI管理頁(yè)面，地址為localhost:8080Spark偽分布集群的搭建（6）為了驗(yàn)證集群運(yùn)行是否正常，下面嘗試在終端窗體中提交一個(gè)Spark應(yīng)用程序到當(dāng)前集群環(huán)境中運(yùn)行Spark偽分布集群的搭建除了Linux終端窗體顯示的運(yùn)行結(jié)果，在Spark集群的WebUI管理頁(yè)面中也可以查到這個(gè)pi.py示例應(yīng)用程序的具體運(yùn)行信息當(dāng)提交應(yīng)用程序到Spark集群中運(yùn)行時(shí)，要指定“spark://vm01:7077”參數(shù)。集群的WebUI管理頁(yè)面可以通過(guò)http://vm01:8080訪問(wèn)，其中，vm01是集群的Master節(jié)點(diǎn)主機(jī)名。不過(guò)，如果Spark集群服務(wù)經(jīng)過(guò)重啟，之前運(yùn)行過(guò)的應(yīng)用程序就不會(huì)出現(xiàn)在WebUI管理頁(yè)面中，不方便事后的分析。因此，為了能夠保留Spark應(yīng)用程序的運(yùn)行記錄，這里需要配置Spark集群的History歷史服務(wù)器Spark偽分布集群的搭建（7）在spark-defaults.conf文件中增加歷史服務(wù)器的配置信息Spark偽分布集群的搭建（7）在spark-defaults.conf文件中增加歷史服務(wù)器的配置信息Spark偽分布集群的搭建（7）在spark-defaults.conf文件中增加歷史服務(wù)器的配置信息從jps命令列出的進(jìn)程列表中可以看出，HistoryServer已經(jīng)在運(yùn)行，同時(shí)Master和Worker進(jìn)程也在運(yùn)行，HDFS服務(wù)也是正常的Spark偽分布集群的搭建（8）再次在終端窗體中提交Spark應(yīng)用程序到集群環(huán)境中運(yùn)行，使用虛擬機(jī)的瀏覽器訪問(wèn)歷史服務(wù)器的管理頁(yè)面，就可以查看應(yīng)用程序運(yùn)行的歷史信息Spark應(yīng)用程序部署模式Spark應(yīng)用程序的運(yùn)行途徑有很多種，在單機(jī)部署時(shí)以本地模式運(yùn)行，在集群部署時(shí)可以選擇Spark集群或外部YARN資源調(diào)度框架等。在使用spark-submit提交計(jì)算任務(wù)時(shí)，可以在命令行指定--master參數(shù)選擇所運(yùn)行的集群環(huán)境（1）Local模式在Local模式下，Spark應(yīng)用程序是在本地計(jì)算機(jī)的Java進(jìn)程中以線程的方式運(yùn)行的，如圖5-51所示。使用Local模式運(yùn)行應(yīng)用程序，也可以指定CPU核數(shù)，CPU核數(shù)決定了開(kāi)啟的工作線程數(shù)量Spark應(yīng)用程序部署模式（2）SparkStandalone模式SparkStandalone實(shí)現(xiàn)的資源調(diào)度框架主要包括Master主節(jié)點(diǎn)和Worker工作節(jié)點(diǎn)，運(yùn)行的進(jìn)程分別是Master和WorkerSparkStandalone是最簡(jiǎn)單和最容易部署的一種模式，尤其適用于開(kāi)發(fā)測(cè)試階段，在提交任務(wù)時(shí)指定“spark://<IP地址>:7077”格式即可Spark應(yīng)用程序部署模式（3）SparkonMesos模式SparkonMesos模式是指Spark應(yīng)用程序運(yùn)行在ApacheMesos資源管理框架之上，該模式將集群的資源管理功能交給Mesos統(tǒng)一處理，Spark只負(fù)責(zé)任務(wù)調(diào)度和計(jì)算工作Mesos是與HadoopMapReduce兼容良好的一種資源調(diào)度框架，不過(guò)在國(guó)內(nèi)的企業(yè)中使用不多，這里僅作了解即可Spark應(yīng)用程序部署模式（4）SparkonYARN模式SparkonYARN模式是Spark應(yīng)用程序運(yùn)行在HadoopYARN框架上的一種方式，集群的資源管理由YARN處理，Spark只負(fù)責(zé)任務(wù)調(diào)度和計(jì)算工作YARN集群與SparkStandalone集群非常相似，其中ResourceManager對(duì)應(yīng)Master進(jìn)程，負(fù)責(zé)集群所有資源的統(tǒng)一管理和分配；NodeManager對(duì)應(yīng)Worker進(jìn)程，負(fù)責(zé)單個(gè)節(jié)點(diǎn)資源的管理；ApplicationMaster則相當(dāng)于Driver進(jìn)程，負(fù)責(zé)管理提交運(yùn)行的應(yīng)用程序，比如任務(wù)調(diào)度、任務(wù)監(jiān)控和容錯(cuò)等Spark應(yīng)用程序部署模式（4）SparkonYARN模式根據(jù)Spark提交計(jì)算任務(wù)的Driver進(jìn)程在YARN集群中的不同位置，又分為yarn-client和yarn-cluster兩種運(yùn)行模式SparkonYARN模式允許Spark與Hadoop生態(tài)圈完美搭配，組成強(qiáng)大的集群環(huán)境，因此得到了比較廣泛的應(yīng)用。除了本地模式和集群模式這兩大類(lèi)部署模式，Spark還支持云服務(wù)部署模式，比如Kubernetes（K8s）就是一個(gè)典型代表，Spark2.3開(kāi)始支持將應(yīng)用程序部署到K8s云容器中運(yùn)行，像亞馬遜云AWS、阿里云等云服務(wù)企業(yè)都推出了EMR（彈性MapReduce計(jì)算）的相關(guān)技術(shù)產(chǎn)品Spark應(yīng)用程序部署實(shí)例（1）Local模式Local模式是在本地計(jì)算機(jī)上以單進(jìn)程、多線程的方式運(yùn)行，不依賴(lài)任何集群環(huán)境，也不需要什么特別的配置，特別適合在開(kāi)發(fā)調(diào)試或簡(jiǎn)單數(shù)據(jù)計(jì)算任務(wù)的場(chǎng)合下使用這里的命令等價(jià)于pyspark--masterlocal[1]，即開(kāi)啟一個(gè)線程在一個(gè)CPU核上運(yùn)行，還可以指定類(lèi)似local[2]或local[*]這樣的參數(shù)，但不要超過(guò)實(shí)際的物理CPU核數(shù)。如果pyspark命令不帶任何參數(shù)，則默認(rèn)使用全部的CPU核，等價(jià)于pyspark--masterlocal[*]Spark應(yīng)用程序部署實(shí)例（1）Local模式通過(guò)spark-submit命令提交應(yīng)用程序時(shí)，也可以指定其以Local模式運(yùn)行，以pi.py示例應(yīng)用程序?yàn)槔齋park應(yīng)用程序部署實(shí)例（2）SparkStandalone模式在提交SparkStandalone集群之前，先確保Master進(jìn)程和Worker進(jìn)程正常運(yùn)行。還有一個(gè)要特別注意的地方是，在同一節(jié)點(diǎn)中無(wú)論是使用SparkStandalone還是YARN集群，這兩個(gè)集群環(huán)境都只能運(yùn)行其中之一，不能同時(shí)使用首先檢查當(dāng)前節(jié)點(diǎn)配置的YARN集群服務(wù)是否處于停止?fàn)顟B(tài)，然后確認(rèn)HDFS服務(wù)是否已啟動(dòng)，以及SparkStandalone集群服務(wù)是否正在運(yùn)行Spark應(yīng)用程序部署實(shí)例（2）SparkStandalone模式繼續(xù)輸入下面的命令，以集群模式啟動(dòng)PySparkShell交互式編程環(huán)境同樣地，可以輸入幾條簡(jiǎn)單的代碼進(jìn)行測(cè)試。測(cè)試完畢后，直接退出PySparkShell交互式編程界面Spark應(yīng)用程序部署實(shí)例（2）SparkStandalone模式如果將這里的命令改為cluster模式，則在啟動(dòng)命令時(shí)會(huì)出現(xiàn)錯(cuò)誤，提示SparkStandalone集群環(huán)境不支持Python應(yīng)用程序的cluster模式Spark應(yīng)用程序部署實(shí)例（2）SparkStandalone模式將示例應(yīng)用程序pi.py提交到SparkStandalone集群中，并以client模式運(yùn)行Spark應(yīng)用程序部署實(shí)例（2）SparkStandalone模式如果提交的pi.py到Standalone集群中以cluster模式運(yùn)行，會(huì)提示“SparkStandalone集群環(huán)境不支持Python應(yīng)用程序的cluster模式”錯(cuò)誤信息Spark應(yīng)用程序部署實(shí)例（2）SparkStandalone模式這里歸納一下SparkStandalone模式的使用規(guī)律

在SparkStandalone集群環(huán)境下，只能以client模式運(yùn)行pyspark交互命令，或以client模式提交PySpark應(yīng)用程序，比如pi.py。

在SparkStandalone集群環(huán)境下，既不支持pyspark這類(lèi)交互命令以cluster模式運(yùn)行，也不支持PySpark應(yīng)用程序以cluster模式提交。也就是說(shuō)，SparkStandalone集群環(huán)境不支持以cluster模式運(yùn)行任何PySpark應(yīng)用程序。在SparkStandalone集群環(huán)境下，由Scala/Java編寫(xiě)的原生應(yīng)用程序，既能以client模式提交，也能以cluster模式提交。但spark-shell這類(lèi)原生交互命令只能以client模式運(yùn)行。

Standalone集群環(huán)境下提交的應(yīng)用程序，或運(yùn)行的交互命令，都可以通過(guò)集群管理WebUI頁(yè)面（http://localhost:8080/）來(lái)觀察其運(yùn)行狀態(tài)Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式Y(jié)ARN和Standalone都是集群管理器，但在同一個(gè)集群的節(jié)點(diǎn)中，它們兩者不能同時(shí)運(yùn)行。因此，若要將Spark應(yīng)用程序提交到Y(jié)ARN集群，就必須先停止SparkStandalone集群服務(wù)，然后啟動(dòng)YARN服務(wù)Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式除了確保YARN和HDFS服務(wù)正常運(yùn)行，還有一項(xiàng)準(zhǔn)備工作要做。因?yàn)樘峤坏腟park計(jì)算任務(wù)是在YARN集群環(huán)境中運(yùn)行的，在運(yùn)行時(shí)要用到Spark安裝目錄中的jars依賴(lài)庫(kù)文件，所以應(yīng)該事先將這些jars依賴(lài)庫(kù)文件放到HDFS上，否則每次提交Spark應(yīng)用程序時(shí)都要上傳一遍，要等待較長(zhǎng)時(shí)間Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式繼續(xù)修改spark-env.sh配置文件Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式準(zhǔn)備工作就緒，現(xiàn)在以YARN集群模式啟動(dòng)PySparkShell交互式編程環(huán)境Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式PySparkShell交互式編程界面也不能在YARN中以cluster模式運(yùn)行Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式嘗試將示例應(yīng)用程序pi.py提交到Y(jié)ARN集群中運(yùn)行Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式再來(lái)嘗試是否能夠以cluster模式將示例應(yīng)用程序pi.py提交到Y(jié)ARN集群中，觀察一下執(zhí)行的過(guò)程Spark應(yīng)用程序部署實(shí)例（3）SparkonYARN模式當(dāng)示例應(yīng)用程序pi.py提交完畢后，終端窗體還會(huì)顯示提交的狀態(tài)信息，但程序運(yùn)行產(chǎn)生的計(jì)算結(jié)果不會(huì)在Linux終端窗體中輸出，這是因?yàn)閏luster模式的Driver進(jìn)程是在YARN集群的內(nèi)部運(yùn)行的，只能通過(guò)YARN集群的WebUI管理頁(yè)面才能看到（http://localhost:8088/）Spark應(yīng)用程序部署實(shí)例（3）Spark應(yīng)用的部署參數(shù)將Spark應(yīng)用程序部署到集群節(jié)點(diǎn)中運(yùn)行時(shí)，在命令行可以根據(jù)需要設(shè)置各種運(yùn)行參數(shù)，包括--master、--deploy-mode等，以便靈活控制使用的內(nèi)存、CPU等資源。下面列出一些常用的Spark部署參數(shù)（其中有圓圈標(biāo)記的使用頻率較高）?--master◎：集群的Master主節(jié)點(diǎn)地址，可以是spark://host:port、yarn、yarn-cluster、yarn-client、local、mesos://host:port等，默認(rèn)為local，即本地運(yùn)行。?--deploy-mode◎：指定Driver進(jìn)程所在的節(jié)點(diǎn)，可以是client或者cluster。前者是運(yùn)行在提交任務(wù)的客戶(hù)端，后者是在集群的一個(gè)節(jié)點(diǎn)上運(yùn)行。若不指定--deploy-mode參數(shù)，則默認(rèn)以client模式運(yùn)行。?--driver-memory◎：Driver進(jìn)程使用的內(nèi)存大小，如5