大數(shù)據(jù)分析與實(shí)戰(zhàn)項(xiàng)目4IDEA開發(fā)環(huán)境下分析碳排放數(shù)據(jù)現(xiàn)有一組我國碳排放數(shù)據(jù)，反映了1997-2019年各種燃料產(chǎn)生的二氧化碳排放量，要求借助IntelliJIDEA開發(fā)工具，編寫程序，完成對(duì)該組數(shù)據(jù)的分析。2022年政府工作報(bào)告中明確提出：完善減污降碳激勵(lì)約束政策，加快形成綠色生產(chǎn)生活方式。情境導(dǎo)入Spark項(xiàng)目分解Spark序號(hào)任務(wù)任務(wù)說明1配置IntelliJIDEA開發(fā)環(huán)境下載IntelliJIDEA開發(fā)工具，完成相關(guān)配置，并初步體驗(yàn)IDEA下的Spark應(yīng)用開發(fā)。2IDEA下分析碳排放數(shù)據(jù)在IntelliJIDEA工具下，使用SparkRDD技術(shù)，完成碳排放數(shù)據(jù)的分析。3緩存與檢查點(diǎn)機(jī)制引入RDD緩存機(jī)制、檢查點(diǎn)，提升程序效能。4廣播變量與累加器在原有程序基礎(chǔ)上，引入共享變量機(jī)制。理解RDD持久化策略，根據(jù)需求設(shè)置RDD檢查點(diǎn)、緩存。能夠安裝配置IDEA工具，創(chuàng)建工程，編寫、運(yùn)行、打包的應(yīng)用程序。了解累加器，能夠使用簡單的廣播變量實(shí)現(xiàn)數(shù)據(jù)共享；123學(xué)習(xí)目標(biāo)Spark項(xiàng)目4

編寫Scala程序處理新能源汽車銷售數(shù)據(jù)Spark任務(wù)1配置IDEA開發(fā)環(huán)境IDEA下編寫碳排放分析程序使用RDD持久化提升運(yùn)行效率任務(wù)2任務(wù)3認(rèn)識(shí)RDD共享變量任務(wù)4任務(wù)分析Spark安裝IntelliJIDEA、安裝插件，以及創(chuàng)建工程、編寫代碼等工作；在IDEA下創(chuàng)建工程，編寫Scala代碼，采用SparkRDD方式讀取carbon.csv碳排放數(shù)據(jù)文件，打印前3行。下載安裝IntelliJIDEASparkIntelliJIDEA是jetbrains推出的一款流行集成開發(fā)環(huán)境工具，借助IDEA可以便捷的開發(fā)Java、Scala等各種應(yīng)用。進(jìn)入官網(wǎng)（或本教材配套軟件包）下載社區(qū)版安裝包（Community版）。下載安裝IntelliJIDEASpark解壓安裝包，并啟動(dòng)sudotar-zxvfideaIC-2023.3.2.tar.gz-C/usr/local//解壓到指定目錄下。cd/usr/local/idea-IC-233.13135.103/bin//進(jìn)入IDEA的bin目錄./idea.sh//啟動(dòng)IDEA安裝Scala插件Spark創(chuàng)建工程——?jiǎng)?chuàng)建Scala文件夾Spark創(chuàng)建工程——添加ScalaSupportSpark創(chuàng)建工程——修改porm.xmlSpark編寫并運(yùn)行程序Spark編寫并運(yùn)行程序Sparkimportorg.apache.spark.SparkContextimportorg.apache.spark.SparkConfobjectSparkTest{defmain(args:Array[String]):Unit={//創(chuàng)建SparkConf對(duì)象confvalconf=newSparkConf().setMaster("local[*]").setAppName("SparkRDDTest")//根據(jù)conf，創(chuàng)建SparkContext對(duì)象sc

valsc=newSparkContext(conf)....................}}編寫并運(yùn)行程序Spark相關(guān)知識(shí)小結(jié)SparkSparkShell主要用于個(gè)人學(xué)習(xí)、測試功能，商業(yè)開發(fā)需要高效率的工具支持；IDEA是一款流行的開發(fā)工具。IDEA下，要想開發(fā)RDD應(yīng)用，需要手工創(chuàng)建創(chuàng)建SparkContext對(duì)象。安裝、配置完IntelliJIDEA后，讀取碳排放csv數(shù)據(jù)文件，打印前三行信息。任務(wù)實(shí)施項(xiàng)目4

編寫Scala程序處理新能源汽車銷售數(shù)據(jù)Spark任務(wù)1配置IDEA開發(fā)環(huán)境IDEA下編寫碳排放分析程序使用RDD持久化提升運(yùn)行效率任務(wù)2任務(wù)3認(rèn)識(shí)RDD共享變量任務(wù)4任務(wù)分析Spark針對(duì)給定的碳排放數(shù)據(jù)，在IDEA開發(fā)環(huán)境下，使用SparkRDD技術(shù)進(jìn)行以下指標(biāo)的分析：（1）找出煤炭產(chǎn)生的排放量最高的年份Top3；（2）計(jì)算2010-2019年的10年內(nèi)，石油類能源的碳排放總量。文件首行的處理Spark在很多原始數(shù)據(jù)文件中，文件的第一行并不是真正的需要分析的數(shù)據(jù)，而是“表頭信息”，即表明文件的字段（列名稱）等；可以借助Linux命令去掉；cd/home/hadoop/data#進(jìn)入carbon.csv的存儲(chǔ)目錄sed'1d'carbon.csv>carbon-without-head.csv#去掉文件首行，另存為新文件文件首行的處理Spark除了使用sed命令，通過編程的方式也可以去掉數(shù)據(jù)文件的首行；（1）RDD中filter過濾的方式，去掉首行；（2）純Scala代碼，讀取文件所有內(nèi)容后，去掉首行后，寫入新文件中。文件首行的處理（純Scala代碼）Sparkvalreader=newBufferedReader(newFileReader(inputFilePath))//創(chuàng)建BufferReader對(duì)象valwriter=newPrintWriter(newFileWriter(outputFilePath))//創(chuàng)建PrintWriter對(duì)象varlineNumber=0while(reader.ready()){if(lineNumber>0){valcurrentLine=reader.readLine()//讀取一行數(shù)據(jù)writer.println(currentLine)//寫入一行數(shù)據(jù)}else{reader.readLine()//跳過第一行

}lineNumber+=1}缺失值的處理Spark數(shù)據(jù)分析之前，需要觀察數(shù)據(jù)，找出其中的“臟數(shù)據(jù)”；根據(jù)業(yè)務(wù)需求，針對(duì)這些“臟數(shù)據(jù)”實(shí)施修正、丟棄、填充等策略。示例：碳排放數(shù)據(jù)文件中2016-2019年的“洗精煤“的排放量缺失，采用2015年的數(shù)據(jù)（120.98）填充。valrdd1=sc.textFile(filePath)valrdd2=rdd1.map(x=>x.split(","))//x(2)代表“洗精煤”；如為空字符串，則修改為"120.98"valrdd3=rdd2.map(x=>x(2)match{case""=>{x(2)="120.98";x.mkString(",")}//如x(2)為空串，則修改為"120.98"；case_=>x.mkString(",")//如果x(2)為非空字符串，則直接將數(shù)組x的所有元素連接成字符串})首行的去除：RDD方式、純Scala代碼、Linux命令等缺失值的處理：填充、刪除等。Spark綜合利用本任務(wù)中的知識(shí)儲(chǔ)備，找出煤炭產(chǎn)生的排放量最高的年份Top3，并石油類能源的碳排放總量。任務(wù)實(shí)施相關(guān)知識(shí)小結(jié)Spark項(xiàng)目4

編寫Scala程序處理新能源汽車銷售數(shù)據(jù)Spark任務(wù)1配置IDEA開發(fā)環(huán)境IDEA下編寫碳排放分析程序使用RDD持久化提升運(yùn)行效率任務(wù)2任務(wù)3認(rèn)識(shí)RDD共享變量任務(wù)4任務(wù)分析Spark有時(shí)為了節(jié)約計(jì)算資源、提升運(yùn)算效率，需要對(duì)反復(fù)使用的RDD進(jìn)行緩存；而對(duì)于某些寬依賴中的計(jì)算冗余問題，可引入檢查點(diǎn)機(jī)制。本任務(wù)要求在4.2代碼的基礎(chǔ)上，加入緩存及檢測點(diǎn)，從而提升程序運(yùn)行的效率。RDD緩存機(jī)制Spark緩存是指將多次使用的數(shù)據(jù)長時(shí)間存儲(chǔ)在集群各節(jié)點(diǎn)的內(nèi)存（或磁盤等其他介質(zhì)）中，以達(dá)到“隨用隨取、減少數(shù)據(jù)的重復(fù)計(jì)算”的目的，從而節(jié)約計(jì)算資源和時(shí)間，提升后續(xù)動(dòng)作的執(zhí)行速度。RDD4.cache()或RDD4.persist()RDD緩存機(jī)制Spark檢查點(diǎn)機(jī)制Spark所謂檢查點(diǎn)機(jī)制，本質(zhì)是通過將RDD寫入磁盤，從而實(shí)現(xiàn)持久化存儲(chǔ)。如果RDD的血統(tǒng)過長會(huì)造成容錯(cuò)成本過高，這樣在中間階段做檢查點(diǎn)容錯(cuò)性能更優(yōu)；檢查點(diǎn)機(jī)制下，如果檢查點(diǎn)后的某節(jié)點(diǎn)出現(xiàn)問題而丟失分區(qū)，可以直接從檢查點(diǎn)的RDD（從磁盤中讀取）開始重做計(jì)算，這樣可以減少開銷。scala>sc.setCheckpointDir("hdfs://localhost:9000/user/hadoop/chekpoint")scala>rdd1.checkpoint()緩存與檢查點(diǎn)的區(qū)別Sparkcache緩存是將數(shù)據(jù)臨時(shí)存放起來（內(nèi)存），不切斷RDD間的血緣依賴。persist緩存可以將數(shù)據(jù)保存到磁盤中，不切斷RDD間的血緣依賴。checkpoint檢查點(diǎn)是將RDD數(shù)據(jù)長期保存到磁盤（通常為HDFS系統(tǒng)），數(shù)據(jù)安全性極高，因此會(huì)切斷血緣依賴。緩存：RDD臨時(shí)存儲(chǔ)起來，包括cache、persist兩種方法。檢查點(diǎn)：RDD寫入HDFS中，持久化保存。Spark綜合利用本任務(wù)中的知識(shí)儲(chǔ)備，引入緩存、檢查點(diǎn)，提升程序效率。任務(wù)實(shí)施相關(guān)知識(shí)小結(jié)Spark項(xiàng)目4

編寫Scala程序處理新能源汽車銷售數(shù)據(jù)Spark任務(wù)1配置IDEA開發(fā)環(huán)境IDEA下編寫碳排放分析程序使用RDD持久化提升運(yùn)行效率任務(wù)2任務(wù)3認(rèn)識(shí)RDD共享變量任務(wù)4任務(wù)分析Spark為了實(shí)現(xiàn)多個(gè)任務(wù)之間變量共享，或者在任務(wù)和任務(wù)控制節(jié)點(diǎn)之間數(shù)據(jù)共享，Spark提供廣播變量和累計(jì)器。現(xiàn)已知2010年-2019年我國的人口數(shù)據(jù)，本任務(wù)要求計(jì)算該時(shí)間段內(nèi)我國人均汽油碳排量（使用廣播變量機(jī)制，將人口數(shù)據(jù)發(fā)送到各個(gè)計(jì)算節(jié)點(diǎn)）。廣播變量Spark廣播變量是一種只讀的共享變量，它是在集群的每個(gè)計(jì)算節(jié)點(diǎn)上保存一個(gè)緩存，而不是每個(gè)任務(wù)保存一份副本；這樣不需要在不同任務(wù)之間頻繁地通過網(wǎng)絡(luò)傳遞數(shù)據(jù)，從而減少了網(wǎng)絡(luò)開銷，同時(shí)也減少了CPU序列化與反序列化的次數(shù)。scala>valbroadcastVar=sc.broadcast(Array(1,2,3))scala>valdata=broadcastVar.value累加器Spark在Spark中，計(jì)算任務(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）。scala>valaccum=sc.accumulator(0,"MyAccumulator")scala>valrdd=sc.parallelize(Array(1,2,3