BigDataAnalyticswithSpark項(xiàng)目五Spark編程進(jìn)階項(xiàng)目概述前期學(xué)習(xí)中，無論是SparkRDD還是SparkSQL，主要是在SparkShell下完成數(shù)據(jù)分析處理工作。但是現(xiàn)實(shí)業(yè)務(wù)中，為了完成一個(gè)數(shù)據(jù)分析任務(wù)，可能需要編寫很多行代碼，需要用到很多類，這時(shí)SparkShell可能力不從心，因而需要用到IntelliJIDEA等專業(yè)開發(fā)工具。本項(xiàng)目將介紹主流的IDEA開發(fā)工具的安裝，并通過Sogou日志分析、疫苗流向數(shù)據(jù)分析兩個(gè)項(xiàng)目，介紹如何在IDEA下完成大數(shù)據(jù)開發(fā)工作。項(xiàng)目效果通過本項(xiàng)目實(shí)踐，可以在IntelliJIDEA環(huán)境下開發(fā)大數(shù)據(jù)應(yīng)用、分析疫苗流向數(shù)據(jù)，并在IDEA控制臺(tái)下輸出數(shù)據(jù)分析結(jié)果如下所示：目錄任務(wù)1搭建IntelliJIDEA開發(fā)環(huán)境IDEA下用SparkRDD分析Sogou日志IDEA下用SparkSQL分析疫苗流向數(shù)據(jù)使用RDD緩存機(jī)制提升效率任務(wù)3認(rèn)識(shí)RDD廣播變量和累加器任務(wù)5理解RDD的依賴關(guān)系任務(wù)2任務(wù)4任務(wù)6任務(wù)2任務(wù)4思維導(dǎo)圖搭建IntelliJIDEA

開發(fā)環(huán)境任務(wù)1SparkShell具有交互式特點(diǎn)。集成的開發(fā)環(huán)境IntelliJIDEA、Eclipse和NightlyBuilds。安裝IntelliJIDEA、安裝插件，以及創(chuàng)建工程、編寫代碼等工作。（1）下載安裝包在Ubuntu下，使用瀏覽器進(jìn)入jetbrains官網(wǎng)/idea/，下載社區(qū)版（Community版），該版本是免費(fèi)的、可滿足初學(xué)者基本學(xué)習(xí)需求。1.1下載安裝IntelliJIDEA（2）安裝IntelliJIDEA下載的完畢后，將IDEA包解壓到/usr/local目錄下，并啟動(dòng)IDEA；打開Ubuntu終端，輸入命令：解壓完畢后，進(jìn)入IDEA安裝目錄，啟動(dòng)IDEA，命令如下：1.1下載安裝IntelliJIDEAcd~/下載sudotar-zxvfideaIC-2019.3.3.tar.gz-C/usr/localcd/usr/local/idea-IC-193.6494.35/cdbin./idea.sh進(jìn)入啟動(dòng)界面后，IDEA讓用戶選擇UI風(fēng)格（SetUItheme），選擇個(gè)人喜好的風(fēng)格（如light）后，點(diǎn)擊“Next：DesktopEntry”。1.1下載安裝IntelliJIDEA在CreateDesktopEntry界面中，直接選擇“Next：LauncherScript”，這樣日后使用過程中可以直接在Ubuntu的應(yīng)用程序搜索中找到IntelliJIDEA，而不用在Ubuntu終端中輸入命令。1.1下載安裝IntelliJIDEA接下來，下載插件界面中，在Scala插件介紹下的“Install”，完成Scala插件的安裝后點(diǎn)擊“StartusingIntelliJIDEA”。1.1下載安裝IntelliJIDEA在圖5-6中，也可以暫不安裝插件，直接點(diǎn)擊“StartusingIntelliJIDEA”；進(jìn)入IDEA歡迎界面后，選擇“Plugins”后，進(jìn)入Plugins市場(chǎng)，搜索“Scala”插件，點(diǎn)擊“Install”也可以完成Scala插件的安裝。1.1下載安裝IntelliJIDEA插件安裝成功后，在IDEA歡迎界面，選擇“StructureforNewProjects”。1.1下載安裝IntelliJIDEA在彈出的“StructureforNewProjects”界面中，依次選擇“Project”、“New”后，設(shè)置JDK（選擇本機(jī)安裝的JDK目錄）。1.1下載安裝IntelliJIDEA在選擇“GlobalLibraries”，點(diǎn)擊+號(hào)，選擇“ScalaSDK”，在彈出的SDK選擇窗口中選擇與本機(jī)匹配的ScalaSDK版本；如沒有合適的版本，可以選擇“Download...”下載。1.1下載安裝IntelliJIDEA設(shè)置好ScalaSDK后，選擇剛剛添加的ScalaSDK，右鍵選擇“CopytoProjectLibraries…”，將ScalaSDK添加到工程中。1.1下載安裝IntelliJIDEA在IDEA歡迎界面，選擇“CreateNewProject”；進(jìn)入圖5-15后，選擇“Maven”后，點(diǎn)擊“Next”。1.2創(chuàng)建工程在“NewProject”窗口中，輸入工程名“mysparkproject”后，點(diǎn)擊“Next”。1.2創(chuàng)建工程進(jìn)入工程界面后，在工程區(qū)域，選擇“mysparkproject”工程，展開其目錄結(jié)構(gòu)；選中“main”目錄后，右鍵選擇“New”——“Directory”，在彈出的“NewDirectory”窗口中輸入新建的目錄名稱“scala”。1.2創(chuàng)建工程選擇新建的scala目錄，右鍵選擇“MarkDirectoryas”——“SourcesRoot”，將其設(shè)置為源目錄。1.2創(chuàng)建工程接下來，選擇原有的“java”目錄，右鍵選擇刪除該目錄。至此，我們完成了在IDEA下Scala工程創(chuàng)建工作。1.2創(chuàng)建工程在編寫代碼前，需要修改mysparkproject工程的pom.xml文件，以便加載相關(guān)依賴包；注意在上述pom.xml文件中，讀者要根據(jù)自己配置環(huán)境（軟件版本等）進(jìn)行適當(dāng)修改，例如<groupId>org.example</groupId>和<artifactId>mysparkproject</artifactId>標(biāo)簽為創(chuàng)建工程時(shí)設(shè)置的groupId和artifactId；而<spark.version>2.2.3</spark.version>及<scala.version>2.11</scala.version>，則需要根據(jù)自己的Spark、Scala版本填寫（作者安裝的Spark版本為2.23，Scala版本為2.11）；<dependency>*****</dependency>標(biāo)簽用于添加Maven依賴；后續(xù)開發(fā)中會(huì)根據(jù)使用的模塊，陸續(xù)添加其他依賴。1.3修改pom.xml的內(nèi)容修改完畢pom.xml后，在Mavenprojectstobeimported選項(xiàng)處，選擇“EnableAuto-Import”；此過程需聯(lián)網(wǎng)下載相關(guān)依賴包，請(qǐng)耐心等待（根據(jù)網(wǎng)絡(luò)情況，可能需要較長(zhǎng)時(shí)間）；執(zhí)行完畢后即可開展后續(xù)的工作（注意本項(xiàng)工作完成前，將無法新建scala程序）。1.3修改pom.xml的內(nèi)容接下來，我們編寫程序?qū)崿F(xiàn)單詞計(jì)數(shù)WordCount功能；在工程目錄中，選擇前面新建的文件夾“scala”，右鍵選擇“New”后，選擇“ScalaClass”；在彈出的“CreateNewScalaClass”中，填寫scala類的名稱為“WordCount”，在Kind類型中選擇“Object”后點(diǎn)擊“OK”鍵，完成scala類的創(chuàng)建。1.4新建scala文件、編寫程序在新建的WordCount.scala中，寫入單詞計(jì)數(shù)的代碼。該程序中，首先引入SparkConf、SparkContext類，然后定義object類WordCount（只有oject類才可以才可以訪問main方法）；在WordCount內(nèi)部定義main方法，實(shí)現(xiàn)單詞計(jì)數(shù)功能。注意，在IDEA環(huán)境下寫Spark程序與SparkShell下有所不同；SparkShell自帶一個(gè)SparkContext實(shí)例sc，但在IDEA下需要自己創(chuàng)建SparkContext對(duì)象（代碼7~9行）。1.4新建scala文件、編寫程序在WordCount.scala代碼窗口內(nèi)的任意位置，右鍵點(diǎn)擊，在彈出的菜單中選擇Run‘WordCount’。運(yùn)行的結(jié)果如下：注意WordCount.scala程序要求，必須存在/usr/local/spark/mycode/wordcount/words.txt這個(gè)文件，否則會(huì)報(bào)錯(cuò)；因此讀者需要?jiǎng)?chuàng)建words.txt文件并置于相應(yīng)目錄下。程序運(yùn)行后，IDEA控制臺(tái)輸出信息較多，可以拖動(dòng)尋找結(jié)果信息。1.5在IDEA中運(yùn)行程序在實(shí)際項(xiàng)目中，需要在IDEA中對(duì)程序進(jìn)行打包，以便提交到Spark集群中運(yùn)行。下面演示如何打包程序。首先，將鼠標(biāo)置于在IDEA的左下角灰色方塊中，在彈出的菜單中選擇“Maven”。1.6工程打包、提交集群運(yùn)行在IDEA中，出現(xiàn)Mave面板，選擇“Lifecycle”，點(diǎn)擊“package”選項(xiàng)，開始打包工作。1.6工程打包、提交集群運(yùn)行打包完成后在工程目錄的target文件夾中可以看到打包好的jar包。對(duì)于該jar包，右鍵選擇“CopyPath”獲得該jar包的路徑。復(fù)制得到j(luò)ar包的路徑后，在Linux終端中，使用命令“/usr/local/spark/bin/spark-submit--classWordCount復(fù)制得到的jar包路徑”來執(zhí)行程序。1.6工程打包、提交集群運(yùn)行

如果要想將程序放置到集群中執(zhí)行，需要做如下修改：（1）WordCount.scala程序中setMaster("local[*]")，local[*]需要修改為“spark://集群主機(jī)IP:7077”；其中7077位Spark默認(rèn)的端口號(hào)（如已經(jīng)修改，則根據(jù)情況調(diào)整）。（2）進(jìn)入Spark的安裝目錄后，使用如下命令提交、運(yùn)行程序：1.6工程打包、提交集群運(yùn)行bin/spark-submit\--classmysparkproject.WordCount\--masterspark://主機(jī)IP:7077\--executor-memory2G\--total-executor-cores4\/****jar所在包目錄******/mysparkproject-1.0-SNAPSHOT.jarIDEA下用SparkRDD分析Sogou日志任務(wù)2Sogou實(shí)驗(yàn)室提供了某段時(shí)間搜索引擎記錄的網(wǎng)頁查詢需求及用戶點(diǎn)擊情況的日志數(shù)據(jù)；要求在IntelliJIDEA環(huán)境下，使用SparkRDD技術(shù)進(jìn)行分析?？紤]到學(xué)習(xí)者電腦配置，選擇迷你版數(shù)據(jù)集（某1天的搜索日志數(shù)據(jù)），命名為sogouoneday.txt，共計(jì)170余萬條記錄。按照數(shù)據(jù)集說明，其數(shù)據(jù)格式為：訪問時(shí)間\t用戶ID\t[查詢?cè)~]\t該URL在返回結(jié)果中的排名\t用戶點(diǎn)擊的順序號(hào)\t用戶點(diǎn)擊的URL；其中，用戶ID是根據(jù)用戶使用瀏覽器訪問搜索引擎時(shí)的Cookie信息自動(dòng)賦值，即同一次使用瀏覽器輸入的不同查詢對(duì)應(yīng)同一個(gè)用戶ID。在Linux終端中，使用head命令查看前10行數(shù)據(jù)。2.1數(shù)據(jù)說明進(jìn)一步分析發(fā)現(xiàn)該“URL在返回結(jié)果中的排名”與“用戶點(diǎn)擊的順序號(hào)”實(shí)際為空格分隔，而非\t；在這里暫時(shí)不處理，在程序代碼中處理。將數(shù)據(jù)集sogouoneday.txt置于用戶主目錄/usr/local/hadoop下，在Linux終端中使用如下命令啟動(dòng)Hadoophdfs服務(wù)，并將文件上傳到hdfs文件系統(tǒng)中。2.1數(shù)據(jù)說明cd/usr/local/hadoop/sbin./start-dfs.shcd/usr/local/hadoop/bin./hdfsdfs-mkdirsogou./hdfsdfs-put/home/hadoop/sogouoneday.txtsogou

下面逐條分析需求，找出解決問題的思路：（1）找出用戶搜索量最大的3個(gè)時(shí)段top3；當(dāng)前數(shù)據(jù)中的訪問時(shí)間格式為HH:mm:ss，可以截取出前2個(gè)字符為小時(shí)（24小時(shí)制），以小時(shí)為統(tǒng)計(jì)單位，找出訪問量最大的三個(gè)時(shí)段；可以考慮將RDD轉(zhuǎn)換為（小時(shí)，1）類型的鍵值對(duì)，然后使用reduceByKey方法，計(jì)算各小時(shí)的訪問量；最后使用sortBy方法排序，得到top3。（2）找出搜索次數(shù)最多的20個(gè)用戶top20；處理方法與第1問類似，用戶ID為用戶的唯一標(biāo)識(shí)，可以考慮將RDD轉(zhuǎn)換為（用戶ID，1）類型的鍵值對(duì)，然后使用reduceByKey方法，計(jì)算各用戶的訪問量；最后使用sortBy方法排序，得到top20。（3）求出平均每個(gè)用戶的搜索量；首先使用count方法，得出總的記錄數(shù)totalNum；對(duì)于用戶ID，使用distinct方法得出用戶數(shù)userNum；最后totalNum除以u(píng)serNum。（4）用戶點(diǎn)擊的URL在返回結(jié)果中的平均排名。對(duì)于所有的“URL在返回結(jié)果中排名”加和，得到rankTotal；rankTotal除以totalNum即可。2.2需求分析在IDEA工程中，新建object類文件Sogou.scala，代碼如下：importorg.apache.log4j.{Level,Logger}importorg.apache.spark.SparkContextimportorg.apache.spark.SparkConfimportscala.util.Success2.3IDEA下編寫程序objectSogou{defmain(args:Array[String]):Unit={Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)//創(chuàng)建sparkConf實(shí)例valconf=newSparkConf().setMaster("local[2]").setAppName("SogouDataPro")//創(chuàng)建SparkContext實(shí)例valsc=newSparkContext(conf)//sogouoneday.txt文件的路徑，可根據(jù)實(shí)際情況修改valpath="hdfs://localhost:9000/user/hadoop/sogou/sogouoneday.txt"http://讀取文件生成RDDvalinputRDD=sc.textFile(path)//對(duì)于inputRDD中的元素使用split方法切割成單詞valsplitRDD=inputRDD.map(x=>x.split("\t"))//取出“時(shí)間”字符串中的前兩個(gè)字符，即為“小時(shí)”；組成鍵值對(duì)（小時(shí)，1）valhourPairs=splitRDD.map(x=>(x(0).trim.substring(0,2),1))//鍵值對(duì)（小時(shí)，1）進(jìn)行reduceByKey運(yùn)算，得到各時(shí)段的訪問量valhourAdd=hourPairs.reduceByKey((a,b)=>a+b)//按照訪問量進(jìn)行排序，降序valsortedHour=hourAdd.sortBy(x=>x._2,false)//打印結(jié)果println("訪問量最多的時(shí)段Top3:")sortedHour.take(3).foreach(println)2.3IDEA下編寫程序//將用戶ID從splitRDD取出，組成鍵值對(duì)（用戶ID，1）valuserPairs=splitRDD.map(x=>(x(1).trim,1))//reduceByKey方法，得到每個(gè)用戶的訪問量；valuserAdd=userPairs.reduceByKey((a,b)=>a+b)//按照用戶的訪問量降序排列valsortedUser=userAdd.sortBy(x=>x._2,false)//打印結(jié)果println("訪問次數(shù)最多的用戶Top10:")sortedUser.take(10).foreach(println)2.3IDEA下編寫程序//計(jì)算數(shù)據(jù)集的總行數(shù)valtotalNum=splitRDD.count()//取出用戶ID組成RDDvalusers=splitRDD.map(x=>x(1))//用戶ID去重valdistinctUsers=users.distinct()//得到用戶的數(shù)量valuserNum=distinctUsers.count()//打印結(jié)果println("平均每個(gè)用戶搜索次數(shù)為："+totalNum/(userNum+0.0))

/*“URL在返回結(jié)果中的排名”與“用戶點(diǎn)擊的順序號(hào)”實(shí)際為空格分隔，而非\t；所以初次分割時(shí)，并未將二者分開，這里再次分割，取出“URL在返回結(jié)果中的排名”*/valsplitRDD2=splitRDD.map(x=>x(3)).map(x=>x.split("")).map(x=>x(0).trim)/*將“URL在返回結(jié)果中的排名”轉(zhuǎn)換成整數(shù)，轉(zhuǎn)換過程中可能拋出異常，使用case匹配；這里也可以使用模式匹配，但執(zhí)行效率可能會(huì)下降。*/valrank=splitRDD2.map(x=>{scala.util.Try(x.toInt)match{caseSuccess(_)=>x.toIntcase_=>0}})//使用reduce方法，求出所有“URL在返回結(jié)果中的排名”之和rankTotal。valrankTotal=rank.reduce((a,b)=>a+b)//打印結(jié)果println("URL在返回結(jié)果中平均排名:"+(rankTotal+0.0)/totalNum)}}2.3IDEA下編寫程序在IDEA中，運(yùn)行程序Sogou.scala，得到所示結(jié)果；訪問數(shù)量最多的3個(gè)時(shí)段分別為16時(shí)、21時(shí)、20時(shí)；訪問次數(shù)最多的用戶ID為“6383499980790535”，該用戶當(dāng)日訪問385次；平均每個(gè)用戶訪問的次數(shù)為3.32次；URL在返回結(jié)果中的平均排名為15.67。2.4運(yùn)行程序IDEA下用SparkSQL分析疫苗流向數(shù)據(jù)任務(wù)3在IDEA環(huán)境下使用SparkSQL技術(shù)分析疫苗采購數(shù)據(jù)。幫助人們了解疫苗流向等相關(guān)信息。數(shù)據(jù)文件為vaccine.csv（置于/home/hadoop/目錄下），包括藥品名name、來源（國(guó)產(chǎn)、進(jìn)口）scr、生產(chǎn)企業(yè)company、銷售省份（含直轄市）prov、記錄的年份year、數(shù)量（單位千）quantity。數(shù)據(jù)樣式如下所示：3.1數(shù)據(jù)說明對(duì)于疫苗流向數(shù)據(jù)用戶需求，現(xiàn)逐條分析如下：（1）找出中標(biāo)次數(shù)最多的5家企業(yè)；vaccine.csv文件中，每一行為一個(gè)中標(biāo)記錄；將數(shù)據(jù)轉(zhuǎn)為DataFrame后，通過groupBy操作即可得出各醫(yī)藥公司中標(biāo)次數(shù)，然后排名找出top5。（2）找出中標(biāo)數(shù)量最多的5家企業(yè)按照醫(yī)藥公司分組，計(jì)算各公司的疫苗的銷量，排序后即可找出top5。（3）對(duì)于狂犬疫苗，分析各公司的市場(chǎng)份額首先篩選出狂犬疫苗數(shù)據(jù)（狂犬疫苗數(shù)據(jù)包含多種數(shù)據(jù)樣式，規(guī)格、制法等有所不同，均為狂犬疫苗類別，計(jì)算總銷量、各公司的銷量，進(jìn)而求出各公司市場(chǎng)份額、排名情況。3.2需求分析（4）分析長(zhǎng)生生物醫(yī)藥公司2018年各類疫苗的流向首先篩選出長(zhǎng)生生物醫(yī)藥公司2018年的相關(guān)數(shù)據(jù)，然后根據(jù)疫苗類型分組統(tǒng)計(jì)，得出該公司各類疫苗的流向。3.2需求分析運(yùn)行Vaccine.scala程序，相應(yīng)輸出如下結(jié)果：（1）中標(biāo)單數(shù)前5名：（2）中標(biāo)疫苗數(shù)量（訂單量）前5名：3.4運(yùn)行程序（3）狂犬疫苗占有率情況：3.4運(yùn)行程序（4）長(zhǎng)生醫(yī)藥公司所生產(chǎn)的疫苗流向情況：3.4運(yùn)行程序使用RDD緩存機(jī)制

提升效率任務(wù)4對(duì)反復(fù)使用的RDD數(shù)據(jù)集進(jìn)行緩存處理。用RDD的緩存機(jī)制提升計(jì)算效率。緩存是指將多次使用的數(shù)據(jù)長(zhǎng)時(shí)間存儲(chǔ)在集群各節(jié)點(diǎn)的內(nèi)存（或磁盤等其他介質(zhì)）中，以達(dá)到“隨用隨取、減少數(shù)據(jù)的重復(fù)計(jì)算”的目的，從而節(jié)約計(jì)算資源和時(shí)間，提升后續(xù)動(dòng)作的執(zhí)行速度。緩存RDD是RDD持久化方案中的一種，對(duì)于迭代算法和快速交互式分析是一個(gè)很關(guān)鍵的技術(shù)。4.1Spark緩存機(jī)制概述（1）為什么要緩存RDD？Spark默認(rèn)情況下，為了充分利用相對(duì)有限的內(nèi)存資源，RDD并不會(huì)長(zhǎng)期駐留在內(nèi)存中；如果內(nèi)存中的RDD過多，當(dāng)有新的RDD生成時(shí)，會(huì)按照以LRU（最近經(jīng)常使用）算法移除最不常用的RDD，以便騰出空間加入新的RDD。緩存RDD目的是讓后續(xù)的RDD計(jì)算速度加快（通常運(yùn)行速度會(huì)加快10倍），是迭代計(jì)算和快速交互的重要工具。4.1Spark緩存機(jī)制概述（2）緩存RDD的方法開發(fā)人員可以使用RDD的persist或者cache方法記錄持久化需求（cache方法可以看做是persist方法的簡(jiǎn)化版）；由于RDD具有惰性計(jì)算的特點(diǎn)，調(diào)用persist或cache方法后，RDD并不會(huì)立即緩存起來，而是等到該RDD首次被施加action操作的時(shí)候，才會(huì)真正的緩存數(shù)據(jù)。同時(shí)，Spark的緩存也具備一定的容錯(cuò)性：如果RDD的任何一個(gè)分區(qū)丟失了，Spark將自動(dòng)根據(jù)其原來的血統(tǒng)信息重新計(jì)算這個(gè)分區(qū)。Spark能夠自動(dòng)監(jiān)控各個(gè)節(jié)點(diǎn)上緩存使用率，并且以LRU的方式將老數(shù)據(jù)逐出內(nèi)存。開發(fā)人員也可以手動(dòng)控制，調(diào)用RDD.unpersist()方法可以刪除無用的緩存。4.1Spark緩存機(jī)制概述（3）緩存的級(jí)別每個(gè)持久化的RDD可以使用不同的存儲(chǔ)級(jí)別，比如根據(jù)業(yè)務(wù)需要可以把RDD保存在磁盤上，或者以java序列化對(duì)象保存到內(nèi)存里，或者跨節(jié)點(diǎn)多副本，或者使用Tachyon存到虛擬機(jī)以外的內(nèi)存里。這些存儲(chǔ)級(jí)別都可以由persist()的參數(shù)StorageLevel對(duì)象來控制。cache()方法本身就是一個(gè)使用默認(rèn)存儲(chǔ)級(jí)別做持久化的快捷方式，默認(rèn)存儲(chǔ)級(jí)別是StorageLevel.MEMORY_ONLY。4.1Spark緩存機(jī)制概述4.1Spark緩存機(jī)制概述緩存級(jí)別如右表所示：對(duì)于需要重復(fù)使用的RDD，建議開發(fā)人員調(diào)用persist方法緩存數(shù)據(jù)。比如，本單元任務(wù)2分析Sogou搜索日志數(shù)據(jù)中，splitRDD被多次調(diào)用，可以考慮緩存機(jī)制以提升效率。另即使用戶沒有調(diào)用persist，Spark也會(huì)自動(dòng)持久化一些Shuffle操作（如reduceByKey）的中間數(shù)據(jù)；因?yàn)镾huffle操作需要消耗較多計(jì)算資源，Spark的自動(dòng)持久化機(jī)制可以避免因某節(jié)點(diǎn)失敗而重新計(jì)算。4.1Spark緩存機(jī)制概述如何選擇存儲(chǔ)級(jí)別？Spark的存儲(chǔ)級(jí)別主要可于在內(nèi)存使用和CPU占用之間做一些權(quán)衡。建議根據(jù)以下步驟來選擇一個(gè)合適的存儲(chǔ)級(jí)別：①如果RDD能使用默認(rèn)存儲(chǔ)級(jí)別（MEMORY_ONLY），則盡量使用默認(rèn)級(jí)別。這是CPU效率最高的方式，所有RDD算子都能以最快的速度運(yùn)行。②如果RDD不適用默認(rèn)存儲(chǔ)級(jí)別（MEMORY_ONLY），可以嘗試MEMORY_ONLY_SER級(jí)別，并選擇一個(gè)高效的序列化協(xié)議（selectingafastserializationlibrary），這將大大節(jié)省數(shù)據(jù)的存儲(chǔ)空間，同時(shí)速度也還不錯(cuò)。③盡量不要把數(shù)據(jù)寫到磁盤上，除非數(shù)據(jù)集重新計(jì)算的代價(jià)很大或者數(shù)據(jù)集是從一個(gè)很大的數(shù)據(jù)源中過濾得到的結(jié)果。④如果需要支持容錯(cuò)，可以考慮使用帶副本的存儲(chǔ)級(jí)別；雖然所有的存儲(chǔ)級(jí)別都能夠以重算丟失數(shù)據(jù)的方式來提供容錯(cuò)性，但是帶副本的存儲(chǔ)級(jí)別可以讓你的應(yīng)用持續(xù)的運(yùn)行，而不必等待重算丟失的分區(qū)。4.1Spark緩存機(jī)制概述接下來，通過實(shí)例體驗(yàn)RDD緩存與否帶來的計(jì)算性能差異；現(xiàn)有數(shù)據(jù)集user_view.txt記載了用戶瀏覽店鋪的日志信息，包括用戶ID、店鋪ID、時(shí)間戳，數(shù)據(jù)字段間用“\t”分割。（1）要實(shí)現(xiàn)的功能在Sparkshell中，統(tǒng)計(jì)所有店鋪的數(shù)量（不重復(fù)）、所有用戶的數(shù)量（不重復(fù)）以及所有記錄數(shù)。（2）數(shù)據(jù)準(zhǔn)備假設(shè)user_view.txt文件現(xiàn)位于/home/hadoop目錄下，打開一個(gè)Linux終端，使用如下命令將該文件上傳到HDFS文件系統(tǒng)中。4.2

SparkRDD緩存體驗(yàn)cd/usr/local/hadoop/sbin./start-all.sh//啟動(dòng)hadoop服務(wù)，如果服務(wù)已經(jīng)開啟，則本步驟可省略。cd/usr/local/hadoop/bin./hdfsdfs-put/home/hadoop/user_view.txt/user/hadoop//文件上傳到HDFS（3）代碼實(shí)現(xiàn)打開一個(gè)Linux終端，輸入如下命令啟動(dòng)Spark并進(jìn)入SparkShell環(huán)境。4.2

SparkRDD緩存體驗(yàn)cd/usr/local/spark/sbin./start-all.shcd/usr/local/spark/bin./spark-shell--masterlocal[*]在SparkShell環(huán)境下，輸入以下命令，完成相關(guān)統(tǒng)計(jì)工作。valpath="hdfs://localhost:9000/user/hadoop/user_view.txt"http://讀取文件生成RDD，對(duì)其元素進(jìn)行字符串切割后形成鍵值對(duì)RDDvalinput=sc.textFile(path).map(x=>x.split("\t")).map(x=>(x(0),1))//input緩存數(shù)據(jù)input.cache()//reduceByKey操作，得到（用戶ID，訪問數(shù)量）為元素的RDDvaluser=input.reduceByKey((a,b)=>a+b)//輸出用戶數(shù)量user.count//根據(jù)用戶訪問量進(jìn)行排序，取前10名user.sortBy(x=>x._2).take(10)//數(shù)input中元素的數(shù)量（關(guān)鍵點(diǎn)）input.count4.2

SparkRDD緩存體驗(yàn)（4）SparkWebUI中查看結(jié)果上述代碼執(zhí)行完畢后，在瀏覽器中輸入localhost:4040進(jìn)入Spark監(jiān)控頁面，選取Stages可以看到上述代碼各階段執(zhí)行的時(shí)長(zhǎng)（受硬件、環(huán)境配置等因素影響，顯示結(jié)果可能會(huì)有不同）。4.2

SparkRDD緩存體驗(yàn)在Storage選項(xiàng)卡中，還可以查看緩存情況；StorageLevel為MemoryDeserialized1xReplicated，表明數(shù)據(jù)緩存在JVM內(nèi)存中，緩存有1個(gè)副本。4.2

SparkRDD緩存體驗(yàn)（5）不使用緩存結(jié)果分析為了演示不使用緩存效果，可以先退出Sparkshell后，再次進(jìn)入；輸入如下代碼（取消對(duì)input的緩存）。valpath="hdfs://localhost:9000/user/hadoop/user_view.txt"valinput=sc.textFile(path).map(x=>x.split("\t")).map(x=>(x(0),1))valuser=input.reduceByKey((a,b)=>a+b)user.countuser.sortBy(x=>x._2).take(10)input.count代碼執(zhí)行完畢后，在瀏覽器中輸入localhost:4040進(jìn)入Spark監(jiān)控頁面；得到下圖所示結(jié)果，發(fā)現(xiàn)最后的count階段執(zhí)行時(shí)間邊長(zhǎng)（增加1S），可見不緩存的情況下執(zhí)行效率下降了。4.2

SparkRDD緩存體驗(yàn)（6）主動(dòng)釋放緩存對(duì)于開發(fā)者主動(dòng)緩存的RDD數(shù)據(jù)，執(zhí)行完畢后要予以釋放，以騰出內(nèi)存空間；釋放空間可以使用方法RDD.unpersist(true)；注意，釋放緩存的RDD要找到正確的時(shí)機(jī)，釋放前一般要確保該RDD不會(huì)再次頻繁使用。4.2

SparkRDD緩存體驗(yàn)認(rèn)識(shí)RDD廣播變量和

累加器任務(wù)5廣播變量和累計(jì)器兩種機(jī)制。廣播變量和累計(jì)器的基本用法。Spark在默認(rèn)情況下，當(dāng)集群的不同節(jié)點(diǎn)的多個(gè)任務(wù)上并行運(yùn)行一個(gè)函數(shù)時(shí)，函數(shù)中使用的變量都會(huì)以副本的形式復(fù)制到各個(gè)機(jī)器節(jié)點(diǎn)上，如果更新這些變量副本的話，這些更新并不會(huì)傳回到驅(qū)動(dòng)器（driver）程序；有時(shí)候也需要在多個(gè)任務(wù)之間共享變量，或者在任務(wù)（Task）和任務(wù)控制節(jié)點(diǎn)（DriverProgram）之間共享變量。為了滿足這些需求，Spark提供了兩種類型的變量：廣播變量（broadcastvariables）和累加器（accumulators）。廣播變量可以實(shí)現(xiàn)變量在所有節(jié)點(diǎn)的內(nèi)存之間進(jìn)行共享；累加器則支持在不同節(jié)點(diǎn)之間進(jìn)行累加計(jì)算（如計(jì)數(shù)或者求和）。5.1共享變量廣播變量提供了一種只讀的共享變量，它是在每個(gè)機(jī)器節(jié)點(diǎn)上保存一個(gè)緩存，而不是每個(gè)任務(wù)保存一份副本。這樣不需要在不同任務(wù)之間頻繁地通過網(wǎng)絡(luò)傳遞數(shù)據(jù)，從而減少了網(wǎng)絡(luò)開銷，同時(shí)也減少了CPU序列化與反序列化的次數(shù)。采用廣播變量時(shí)，通?？梢栽诿總€(gè)節(jié)點(diǎn)上保存一個(gè)較大的輸入數(shù)據(jù)集，這要比常規(guī)的變量副本更高效（普通變量是每個(gè)任務(wù)一個(gè)副本，而一個(gè)節(jié)點(diǎn)上可能有多個(gè)任務(wù)）。5.2廣播變量SparkContext提供了broadcast()方法用于創(chuàng)建廣播變量，例如對(duì)于對(duì)于變量v，只需調(diào)用SparkContext.broadcast(v)即可得到一個(gè)廣播變量；這個(gè)廣播變量是對(duì)變量v的一個(gè)包裝，要訪問其值，可以調(diào)用廣播變量的value方法。代碼示例如下：scala>valbroadcastVar=sc.broadcast(Array(1,2,3))broadcastVar:org.apache.spark.broadcast.Broadcast[Array[Int]]=Broadcast(0)scala>broadcastVar.valueres0:Array[Int]=Array(1,2,3)5.2廣播變量在某些關(guān)聯(lián)查詢場(chǎng)景中，可對(duì)一些公共數(shù)據(jù)進(jìn)行廣播；假設(shè)現(xiàn)有（號(hào)碼段，歸屬地，運(yùn)營(yíng)商）數(shù)據(jù)，例如（1371001，廣州，中國(guó)移動(dòng)）；要求對(duì)數(shù)據(jù)（戶主姓名，電話號(hào)碼）進(jìn)行補(bǔ)全，輸出戶主姓名、電話號(hào)碼、歸屬地、運(yùn)營(yíng)商信息；使用廣播變量的實(shí)現(xiàn)過程如下：//構(gòu)造一個(gè)Map：號(hào)碼段—>（歸屬地，運(yùn)營(yíng)商）scala>valtelephoneDetail=Map("1371001"->("廣州","中國(guó)移動(dòng)"),"1371350"->("深圳","中國(guó)移動(dòng)"),"1331847電信"),"1324240"->("深圳","中國(guó)聯(lián)通"))telephoneDetail:scala.collection.immutable.Map[String,(String,String)]=Map(1371001->(廣州,中國(guó)移動(dòng)),1371350->(深圳,中國(guó)移動(dòng)),1331847->(珠海,中國(guó)電信),1324240->(深圳,中國(guó)聯(lián)通))//調(diào)用SparkContext的broadcast方法，將telephoneDetail廣播發(fā)送scala>valtdBroadCast=sc.broadcast(telephoneDetail)tdBroadCast:org.apache.spark.broadcast.Broadcast[scala.collection.immutable.Map[String,(String,String)]]=Broadcast(2)5.2廣播變量//構(gòu)建一個(gè)包含（電話號(hào)碼，用戶名）的Listscala>valcustomer=List((,"tom"),(,"jerry"))customer:List[(String,String)]=List(tom),jerry))//將customer轉(zhuǎn)換為RDDscala>valcusRDD=sc.parallelize(customer)cusRDD:org.apache.spark.rdd.RDD[(String,String)]=ParallelCollectionRDD[2]atparallelizeat<console>:26//使用廣播變量tdBroadCast補(bǔ)全用戶信息scala>valcustomerDetail=cusRDD.map(x=>{|valshorttel=x._1.substring(0,7)|valdetail=tdBroadCast.value(shorttel)|(x._1,x._2,detail._1,detail._2)|}|)customerDetail:org.apache.spark.rdd.RDD[(String,String,String,String)]=MapPartitionsRDD[3]atmapat<console>:325.2廣播變量scala>customerDetail.collectres4:Array[(String,String,String,String)]=Array(tom,珠海,中國(guó)電信),jerry,深圳,中國(guó)移動(dòng)))//釋放tdBroadCast廣播變量scala>tdBroadCast.unpersist實(shí)際業(yè)務(wù)中，需要廣播的數(shù)據(jù)往往是通過讀取數(shù)據(jù)庫表或者讀取文件生成，而非示例中手工生成；當(dāng)廣播變量不再使用后，要及時(shí)釋放。在主流的分布式計(jì)算框架中，都存在Spark廣播變量類似的應(yīng)用，其主要目的就是減少數(shù)據(jù)傳遞開銷及減少對(duì)CPU資源的消耗。5.2廣播變量廣播變量提供了一種只讀的共享變量，它是在每個(gè)機(jī)器節(jié)點(diǎn)上保存一個(gè)緩存，而不是每個(gè)任務(wù)保存一份副本。這樣不需要在不同任務(wù)之間頻繁地通過網(wǎng)絡(luò)傳遞數(shù)據(jù)，從而減少了網(wǎng)絡(luò)開銷，同時(shí)也減少了CPU序列化與反序列化的次數(shù)。采用廣播變量時(shí)，通?？梢栽诿總€(gè)節(jié)點(diǎn)上保存一個(gè)較大的輸入數(shù)據(jù)集，這要比常規(guī)的變量副本更高效（普通變量是每個(gè)任務(wù)一個(gè)副本，而一個(gè)節(jié)點(diǎn)上可能有多個(gè)任務(wù)）。5.2廣播變量累加器是Spark提供的另一種共享變量機(jī)制；在Spark中，每一個(gè)任務(wù)可能會(huì)分配到不同節(jié)點(diǎn)中執(zhí)行；在執(zhí)行過程中，如果需要將多個(gè)節(jié)點(diǎn)中的數(shù)據(jù)累加到一個(gè)變量中，則可以通過累計(jì)器實(shí)現(xiàn)，即利用累加器可以實(shí)現(xiàn)計(jì)數(shù)（類似MapReduce中的計(jì)數(shù)器）或者求和（SUM）。Spark原生支持了數(shù)字類型的累加器，開發(fā)者也可以自定義新的累加器。如果創(chuàng)建累加器的時(shí)設(shè)置了名稱，則該名稱會(huì)展示在SparkUI上，有助于了解程序運(yùn)行處于哪個(gè)階段。5.3累加器開發(fā)人員可以調(diào)用SparkContext.accumulator(v)可以創(chuàng)建一個(gè)累加器，v為累加器的初始值。累加器創(chuàng)建后，可以使用add方法或者+=操作符來進(jìn)行累加操作。注意：任務(wù)本身并不能讀取累加器的值，只有驅(qū)動(dòng)器程序可以使用value方法訪問累加器的值。以下代碼展示了如何使用累加器對(duì)一個(gè)元素?cái)?shù)組求和：scala>valaccum=sc.accumulator(0,"MyAccumulator")scala>valrdd=sc.parallelize(Array(1,2,3,4))rdd:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[4]atparallelizeat<console>:24scala>rdd.foreach(x=>accum+=x)scala>accumres7:org.apache.spark.Accumulator[Int]=105.3累加器Spark內(nèi)置了整數(shù)累加器、長(zhǎng)精度浮點(diǎn)數(shù)累加器和幾個(gè)累加器，上述代碼使用的累加器即為整型累加器；開發(fā)人員也可以通過繼承AccumulatorParam來自定義累加器。AccumulatorParam主要有兩個(gè)方法：（1）zero，這個(gè)方法為累加器提供一個(gè)“零值”；（2）addInPlace，將收到的兩個(gè)參數(shù)值進(jìn)行累加。例如，假設(shè)我們需要為Vector提供一個(gè)累加機(jī)制，那么可能的實(shí)現(xiàn)方式如下：objectVectorAccumulatorParamextendsAccumulatorParam[Vector]{defzero(initialValue:Vector):Vector={Vector.zeros(initialValue.size)}defaddInPlace(v1:Vector,v2:Vector):Vector={v1+=v2}}5.3累加器//使用如下方式創(chuàng)建累加器//valvecAccum=sc.accumulator(newVector(...))(VectorAccumulatorParam)對(duì)于在action算子中更新的累加器，Spark保證每個(gè)任務(wù)對(duì)累加器的更新只會(huì)被應(yīng)用一次，例如，某些任務(wù)如果重啟過，則不會(huì)再次更新累加器。而如果在transformation算子中更新累加器，那么用戶需要注意，一旦某個(gè)任務(wù)因?yàn)槭”恢匦聢?zhí)行，那么其對(duì)累加器的更新可能會(huì)實(shí)施多次。累加器并不會(huì)改變Spark懶惰求值的運(yùn)算特性。如果在RDD算子中更新累加器，那么其值只會(huì)在RDD做action計(jì)算的時(shí)候被更新一次。因此，在transformation算子（如map）中更新累加器，其值并不能保證一定被更新；如以下代碼所示：valaccum=sc.accumulator(0)data.map{x=>accum+=x;f(x)}//這里，accum的值任然是0，因?yàn)闆]有action算子，所以map也不會(huì)進(jìn)行實(shí)際的計(jì)算5.3累加器理解RDD的依賴關(guān)系任務(wù)6理解SparkRDD的執(zhí)行流程及RDD間的依賴關(guān)系。新的數(shù)據(jù)保存方式——檢查點(diǎn)機(jī)制。SparkRDD的操作分為轉(zhuǎn)換操作和行動(dòng)操作兩大類，由于RDD的不可修改性，需要由舊RDD不斷產(chǎn)生新RDD，以供給下次操作使用，直到最后一個(gè)RDD經(jīng)過行動(dòng)操作后，產(chǎn)生需要的結(jié)果并輸出；RDD采用了惰性計(jì)算機(jī)制，即在RDD的執(zhí)行過程中，真正的計(jì)算發(fā)生在RDD的“行動(dòng)”操作時(shí)刻，對(duì)于“行動(dòng)”之前的所有“轉(zhuǎn)換”操作，Spark只是記錄下“轉(zhuǎn)換”操作使用的部分基礎(chǔ)數(shù)據(jù)集以及RDD生成的軌跡，即RDD之間的依賴關(guān)系，而不會(huì)觸發(fā)真正的計(jì)算。對(duì)于輸入數(shù)據(jù)Input，Spark從邏輯上生成RDD1和RDD2兩個(gè)RDD，經(jīng)過一系列“轉(zhuǎn)換”操作，邏輯上生成了RDDn；但上述RDD并未真正生成，他們是邏輯上的數(shù)據(jù)集，Spark只是記錄了RDD之間的生成和依賴關(guān)系。當(dāng)RDDn要進(jìn)行輸出時(shí)（執(zhí)行“行動(dòng)操作”時(shí)），Spark才會(huì)根據(jù)RDD的依賴關(guān)系生成DAG（有向無環(huán)圖），并從起點(diǎn)開始真正的計(jì)算。6.1SparkRDD的執(zhí)行流程上述處理過程中，RDD之間前后相連，形成了“血緣關(guān)系（Lingeage）”，通過血緣關(guān)系連接起來的一系列RDD操作就可以實(shí)現(xiàn)管道化（pipeline），避免了多次轉(zhuǎn)換操作之間數(shù)據(jù)同步的等待，而且不用擔(dān)心有過多的中間數(shù)據(jù)，因?yàn)檫@些具有血緣關(guān)系的操作都管道化了，一個(gè)操作得到的結(jié)果不需要保存為中間數(shù)據(jù)，而是直接管道式地流入到下一個(gè)操作進(jìn)行處理。同時(shí)，這種通過血緣關(guān)系把一系列操作進(jìn)行管道化連接的設(shè)計(jì)方式，也使得管道中每次操作的計(jì)算變得相對(duì)簡(jiǎn)單，保證了每個(gè)操作在處理邏輯上的單一性；在HadoopMapReduce的設(shè)計(jì)中，為了盡可能地減少M(fèi)apReduce過程，在單個(gè)MapReduce中會(huì)寫入過多復(fù)雜的邏輯。6.1SparkRDD的執(zhí)行流程RDD的每次轉(zhuǎn)換操作都會(huì)產(chǎn)生一個(gè)新的RDD，那么前后RDD之間便形成了一定的依賴關(guān)系；RDD中的依賴關(guān)系分為窄依賴（NarrowDependency）與寬依賴（WideDependency），兩種依賴之間的區(qū)別如下圖所示。6.2

RDD間的依賴關(guān)系窄依賴：一個(gè)RDD對(duì)它的父RDD，只有簡(jiǎn)單的一對(duì)一的依賴關(guān)系，也就是說，RDD中的每個(gè)partition，僅僅依賴于父RDD中的一個(gè)partition，父RDD和子RDD的partition之間是一對(duì)一的關(guān)系。這種情況下，是簡(jiǎn)單的RDD之間的依賴關(guān)系，也被稱之為窄依賴。6.2

RDD間的依賴關(guān)系寬依賴：本質(zhì)就是shuffle，也就是說每一個(gè)父RDD中的partition中的數(shù)據(jù)，都可能會(huì)傳輸一部分到下一個(gè)RDD的每一個(gè)partition，也就是說，每一個(gè)父RDD和子RDD的partition之間，具有交互錯(cuò)雜的關(guān)系，那么這種情況就叫做兩個(gè)RDD之間是寬依賴，同時(shí)他們之間發(fā)生的操作是shuffle。6.2

RDD間的依賴關(guān)系總體而言，如果父RDD的一個(gè)分區(qū)只被一個(gè)子RDD的一個(gè)分區(qū)所使用就是窄依賴，否則就是寬依賴。窄依賴典型的操作包括map、filter、union等，寬依賴典型的操作包括groupByKey、sortByKey等。6.2

RDD間的依賴關(guān)系Spark的這種依賴關(guān)系設(shè)計(jì)，使其具有了天生的容錯(cuò)性，大大加快了Spark的執(zhí)行速度。相對(duì)而言，在兩種依賴關(guān)系中，窄依賴的失敗恢復(fù)更為高效，它只需要根據(jù)父RDD分區(qū)重新計(jì)算丟失的分區(qū)即可，而且可以并行地在不同節(jié)點(diǎn)進(jìn)行重新計(jì)算。對(duì)于寬依賴，通常子RDD分區(qū)通常來自多個(gè)父RDD分區(qū)，重新計(jì)算的開銷較大。上圖所示，RDDa、RDDb直接為窄依賴，當(dāng)RDDb的分區(qū)b1丟失時(shí)，只需要重新計(jì)算父RDDa的a1分區(qū)即可。而RDDc、RDDe之間為寬依賴，當(dāng)RDDe的分區(qū)e1丟失時(shí)，則需要重新計(jì)算RDDe的所有分區(qū)，這就產(chǎn)生了冗余計(jì)算（c1、c2、c3中對(duì)于e2的數(shù)據(jù)）。6.2

RDD間的依賴關(guān)系當(dāng)Spark集群中某一個(gè)節(jié)點(diǎn)由于宕機(jī)導(dǎo)致數(shù)據(jù)丟失，可以通過Spark中RDD的容錯(cuò)機(jī)制恢復(fù)丟失的數(shù)據(jù)。RDD提供了兩種故障恢復(fù)方式：血統(tǒng)方式和設(shè)置檢查點(diǎn)方式。如前所述，血統(tǒng)方式主要是根據(jù)RDD之間的依賴關(guān)系對(duì)丟失數(shù)據(jù)的RDD進(jìn)行數(shù)據(jù)恢復(fù)。如果父子RDD間是窄依賴，則只需把父RDD對(duì)應(yīng)分區(qū)重新計(jì)算即可，不需要依賴于其他節(jié)點(diǎn)，計(jì)算過程也不會(huì)產(chǎn)生冗余計(jì)算；若父子RDD間是寬依賴，則需要父RDD的所有分區(qū)都要從頭到尾計(jì)算，計(jì)算過程存在冗余。為了解決寬依賴中的計(jì)算冗余問題，Spark又提供了另一種數(shù)據(jù)容錯(cuò)方式——設(shè)置檢查點(diǎn)方式6.3檢查點(diǎn)機(jī)制設(shè)置檢查點(diǎn)方式，本質(zhì)是通過將RDD寫入詞磁盤，是為了協(xié)助血統(tǒng)做容錯(cuò)輔助。如果血統(tǒng)過長(zhǎng)會(huì)造成容錯(cuò)成本過高，這樣在中間階段做檢查點(diǎn)容錯(cuò)性能更優(yōu)；檢查點(diǎn)機(jī)制下，如果檢查點(diǎn)后的某節(jié)點(diǎn)出現(xiàn)問題而丟失分區(qū)，可以直接從檢查點(diǎn)的RDD（從磁盤中讀?。╅_始重做Lineage，這樣可以減少開銷；通常情況下，Spark通過將數(shù)據(jù)寫入到HDFS文件系統(tǒng)實(shí)現(xiàn)RDD檢查點(diǎn)功能，而HDFS是多副本的高可靠存儲(chǔ)（通過多副本