Spark大数据分析与实战（第二版） 课件 项目4 IDEA下分析碳排放数据

大数据分析与实战项目4IDEA开发环境下分析碳排放数据现有一组我国碳排放数据，反映了1997-2019年各种燃料产生的二氧化碳排放量，要求借助IntelliJIDEA开发工具，编写程序，完成对该组数据的分析。2022年政府工作报告中明确提出：完善减污降碳激励约束政策，加快形成绿色生产生活方式。情境导入Spark项目分解Spark序号任务任务说明1配置IntelliJIDEA开发环境下载IntelliJIDEA开发工具，完成相关配置，并初步体验IDEA下的Spark应用开发。2IDEA下分析碳排放数据在IntelliJIDEA工具下，使用SparkRDD技术，完成碳排放数据的分析。3缓存与检查点机制引入RDD缓存机制、检查点，提升程序效能。4广播变量与累加器在原有程序基础上，引入共享变量机制。理解RDD持久化策略，根据需求设置RDD检查点、缓存。能够安装配置IDEA工具，创建工程，编写、运行、打包的应用程序。了解累加器，能够使用简单的广播变量实现数据共享；123学习目标Spark项目4

编写Scala程序处理新能源汽车销售数据Spark任务1配置IDEA开发环境IDEA下编写碳排放分析程序使用RDD持久化提升运行效率任务2任务3认识RDD共享变量任务4任务分析Spark安装IntelliJIDEA、安装插件，以及创建工程、编写代码等工作；在IDEA下创建工程，编写Scala代码，采用SparkRDD方式读取carbon.csv碳排放数据文件，打印前3行。下载安装IntelliJIDEASparkIntelliJIDEA是jetbrains推出的一款流行集成开发环境工具，借助IDEA可以便捷的开发Java、Scala等各种应用。进入官网（或本教材配套软件包）下载社区版安装包（Community版）。下载安装IntelliJIDEASpark解压安装包，并启动sudotar-zxvfideaIC-2023.3.2.tar.gz-C/usr/local//解压到指定目录下。cd/usr/local/idea-IC-233.13135.103/bin//进入IDEA的bin目录./idea.sh//启动IDEA安装Scala插件Spark创建工程——创建Scala文件夹Spark创建工程——添加ScalaSupportSpark创建工程——修改porm.xmlSpark编写并运行程序Spark编写并运行程序Sparkimportorg.apache.spark.SparkContextimportorg.apache.spark.SparkConfobjectSparkTest{defmain(args:Array[String]):Unit={//创建SparkConf对象confvalconf=newSparkConf().setMaster("local[*]").setAppName("SparkRDDTest")//根据conf，创建SparkContext对象sc

valsc=newSparkContext(conf)....................}}编写并运行程序Spark相关知识小结SparkSparkShell主要用于个人学习、测试功能，商业开发需要高效率的工具支持；IDEA是一款流行的开发工具。IDEA下，要想开发RDD应用，需要手工创建创建SparkContext对象。安装、配置完IntelliJIDEA后，读取碳排放csv数据文件，打印前三行信息。任务实施项目4

编写Scala程序处理新能源汽车销售数据Spark任务1配置IDEA开发环境IDEA下编写碳排放分析程序使用RDD持久化提升运行效率任务2任务3认识RDD共享变量任务4任务分析Spark针对给定的碳排放数据，在IDEA开发环境下，使用SparkRDD技术进行以下指标的分析：（1）找出煤炭产生的排放量最高的年份Top3；（2）计算2010-2019年的10年内，石油类能源的碳排放总量。文件首行的处理Spark在很多原始数据文件中，文件的第一行并不是真正的需要分析的数据，而是“表头信息”，即表明文件的字段（列名称）等；可以借助Linux命令去掉；cd/home/hadoop/data#进入carbon.csv的存储目录sed'1d'carbon.csv>carbon-without-head.csv#去掉文件首行，另存为新文件文件首行的处理Spark除了使用sed命令，通过编程的方式也可以去掉数据文件的首行；（1）RDD中filter过滤的方式，去掉首行；（2）纯Scala代码，读取文件所有内容后，去掉首行后，写入新文件中。文件首行的处理（纯Scala代码）Sparkvalreader=newBufferedReader(newFileReader(inputFilePath))//创建BufferReader对象valwriter=newPrintWriter(newFileWriter(outputFilePath))//创建PrintWriter对象varlineNumber=0while(reader.ready()){if(lineNumber>0){valcurrentLine=reader.readLine()//读取一行数据writer.println(currentLine)//写入一行数据}else{reader.readLine()//跳过第一行

}lineNumber+=1}缺失值的处理Spark数据分析之前，需要观察数据，找出其中的“脏数据”；根据业务需求，针对这些“脏数据”实施修正、丢弃、填充等策略。示例：碳排放数据文件中2016-2019年的“洗精煤“的排放量缺失，采用2015年的数据（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)为非空字符串，则直接将数组x的所有元素连接成字符串})首行的去除：RDD方式、纯Scala代码、Linux命令等缺失值的处理：填充、删除等。Spark综合利用本任务中的知识储备，找出煤炭产生的排放量最高的年份Top3，并石油类能源的碳排放总量。任务实施相关知识小结Spark项目4

编写Scala程序处理新能源汽车销售数据Spark任务1配置IDEA开发环境IDEA下编写碳排放分析程序使用RDD持久化提升运行效率任务2任务3认识RDD共享变量任务4任务分析Spark有时为了节约计算资源、提升运算效率，需要对反复使用的RDD进行缓存；而对于某些宽依赖中的计算冗余问题，可引入检查点机制。本任务要求在4.2代码的基础上，加入缓存及检测点，从而提升程序运行的效率。RDD缓存机制Spark缓存是指将多次使用的数据长时间存储在集群各节点的内存（或磁盘等其他介质）中，以达到“随用随取、减少数据的重复计算”的目的，从而节约计算资源和时间，提升后续动作的执行速度。RDD4.cache()或RDD4.persist()RDD缓存机制Spark检查点机制Spark所谓检查点机制，本质是通过将RDD写入磁盘，从而实现持久化存储。如果RDD的血统过长会造成容错成本过高，这样在中间阶段做检查点容错性能更优；检查点机制下，如果检查点后的某节点出现问题而丢失分区，可以直接从检查点的RDD（从磁盘中读取）开始重做计算，这样可以减少开销。scala>sc.setCheckpointDir("hdfs://localhost:9000/user/hadoop/chekpoint")scala>rdd1.checkpoint()缓存与检查点的区别Sparkcache缓存是将数据临时存放起来（内存），不切断RDD间的血缘依赖。persist缓存可以将数据保存到磁盘中，不切断RDD间的血缘依赖。checkpoint检查点是将RDD数据长期保存到磁盘（通常为HDFS系统），数据安全性极高，因此会切断血缘依赖。缓存：RDD临时存储起来，包括cache、persist两种方法。检查点：RDD写入HDFS中，持久化保存。Spark综合利用本任务中的知识储备，引入缓存、检查点，提升程序效率。任务实施相关知识小结Spark项目4

编写Scala程序处理新能源汽车销售数据Spark任务1配置IDEA开发环境IDEA下编写碳排放分析程序使用RDD持久化提升运行效率任务2任务3认识RDD共享变量任务4任务分析Spark为了实现多个任务之间变量共享，或者在任务和任务控制节点之间数据共享，Spark提供广播变量和累计器。现已知2010年-2019年我国的人口数据，本任务要求计算该时间段内我国人均汽油碳排量（使用广播变量机制，将人口数据发送到各个计算节点）。广播变量Spark广播变量是一种只读的共享变量，它是在集群的每个计算节点上保存一个缓存，而不是每个任务保存一份副本；这样不需要在不同任务之间频繁地通过网络传递数据，从而减少了网络开销，同时也减少了CPU序列化与反序列化的次数。scala>valbroadcastVar=sc.broadcast(Array(1,2,3))scala>valdata=broadcastVar.value累加器Spark在Spark中，计算任务可能会分配到不同节点中执行；在执行过程中，如果需要将多个节点中的数据累加到一个变量中，则可以通过累计器实现，即利用累加器可以实现计数（类似MapReduce中的计数器）或者求和（SUM）。scala>valaccum=sc.accumulator(0,"MyAccumulator")scala>valrdd=sc.parallelize(Array(1,2,3