Spark大数据技术与应用（第3版）（微课版）课件 项目3 查询和统计员工薪资数据-Spark Shell编程V1.0

查询和统计员工薪资数据

——SparkShell编程项目背景薪酬调整和绩效管理在现代企业的发展中发挥着重要的作用。一个企业的薪酬体系建立后，通常会不断完善。科学合理的企业薪酬体系能够提升员工的工作热情和积极性，激励他们为企业和社会做出更大贡献。同时，这也有助于加强团队凝聚力，保持企业高效、稳定的发展，从而增强企业竞争力及其对社会的贡献能力。项目背景公司有员工2023年上半年薪资文件（Employee_salary_first_half.csv）和下半年薪资文件（Employee_salary_second_half.csv），两份文件的数据格式和数据字段均相同，以员工2023年上半年的薪资文件为例，文件共有10个数据字段，字段说明如下。字段名称说明字段名称说明EmpID员工IDGROSS总薪资Name姓名Net_Pay实际薪资Gender性别Deduction薪资扣除部分Date_of_Birth出生日期Designation职位Age年龄Department部门创建SparkRDDRDD基础操作和进阶操作键值对RDD操作RDD连接操作RDD文件读写项目实施创建SparkRDDRDD是一个容错的、只读的、可进行并行操作的数据结构，是一个分布在集群各个节点中的存放元素的集合。创建RDD的3种方法如下：将程序中已存在的Seq集合（如列表、数组）转换成RDD。对已有RDD进行转换得到新的RDD。直接读取外部存储系统的数据创建RDD。从内存中已有数据创建RDDSparkContext类中有两个方法：parallelize()方法makeRDD()方法这两个方法均利用内存中已存在的集合，通过复制集合的元素来创建一个可用于并行计算的RDD。parallelize()方法parallelize()方法有两个输入参数：要转化的集合：必须是Seq集合，如List、Vector、ArrayBuffer等。分区数：若不设分区数，则RDD的分区数默认为该程序分配到的资源的CPU核心数。通过parallelize()方法用一个Seq集合的数据创建RDD，并设置分区数为4，创建后查看该RDD的分区数。makeRDD()方法makeRDD()方法有两种使用方式：第一种与parallelize()方法一致。第二种是通过接收一个Seq[(T,Seq[String])]参数类型创建RDD。第二种方式生成的RDD中保存的是T的值，Seq[String]部分的数据会按照Seq[(T,Seq[String])]的顺序存放到各个分区中，每个Seq[String]对应一个分区，并为数据提供位置信息。通过preferredLocations()方法可以根据位置信息查看每个分区的值。调用makeRDD()时，不能直接指定RDD的分区个数，分区个数与Seq[String]参数的个数保持一致。makeRDD()方法使用makeRDD()方法创建RDD，并根据位置信息查看每一个分区的值。从外部存储系统读取数据创建RDD从外部存储系统中读取数据创建RDD的方法可以有很多种数据来源，可通过SparkContext对象的textFile()方法读取数据集。textFile()方法支持多种类型的数据集，如目录、文本文件、压缩文件和通配符匹配的文件等，并且允许设定分区个数。通过textFile()方法读取Linux本地文件并创建RDD：通过textFile()方法读取HDFS文件创建RDD：vallocal_data=sc.textFile("file:///opt/data/test.txt")local_data.count()valhdfs_data=sc.textFile("/tipdm/data/test.txt")hdfs_data.count()创建SparkRDDRDD基础操作和进阶操作键值对RDD操作RDD连接操作RDD文件读写项目实施map()方法map()方法是一种基础的RDD转换操作，可以对RDD中的每一个数据元素通过某种函数进行转换并返回新的RDD。map()方法是懒操作，不会立即进行计算。转换操作是操作RDD的一种方式，通过转换已有RDD生成新的RDD。因为RDD是一个不可变的集合，所以如果对RDD中的数据进行了某种转换，那么会生成一个新的RDD。例如，用存放5个Int类型元素的列表创建RDD，通过map()方法对每个元素进行平方运算，生成新的RDD。valrdd=sc.parallelize(List(1,3,45,3,76))valsq_rdd=rdd.map(x=>x*x)sortBy()方法SortBy()方法用于对标准RDD进行排序，有3个可输入参数：函数f:(T)=>K：T表示要被排序对象中的每一个元素，K表示元素中要进行排序的值。ascending：决定排序后RDD中的元素是升序（true）还是降序(false)，默认为升序。numPartitions：决定排序后RDD的分区个数，默认与排序前相同，即this.partitions.size。sortBy()方法的第一个参数是必须输入的，而后面的两个参数可以不输入。例如，通过一个存放了3个二元组的列表创建一个RDD，对元组的第二个值进行降序排序，分区个数设置为1。valrdd=sc.parallelize(List((1,3),(45,3),(7,6)))valsort_rdd=rdd.sortBy(x=>x._2,false,1)collect()方法collect()方法是一种行动操作，可以将RDD中所有元素转换成数组并返回到Driver端，适用于返回处理后的少量数据。因为需要从集群各个节点收集数据到本地，经过网络传输，并且加载到Driver内存中，所以如果数据量比较大，会给网络传输造成很大的压力。因此，数据量较大时，尽量不使用collect()方法，否则可能导致Driver端出现内存溢出问题。collect()方法collect()方法主要有以下两种操作方式：collect():Array[T]：直接调用collect()返回该RDD中的所有元素，返回类型为Array[T]。collect[U](f:PartialFunction[T,U]):RDD[U]：这种方式需要提供一个标准的偏函数，将元素保存至一个RDD中，返回一个新的RDD。flatMap()方法flatMap()方法将指定的函数参数应用于RDD中的每一个元素，并将返回的迭代器（如数组、列表等）中的所有元素构成新的RDD。使用flatMap()方法时先进行map（映射）再进行flat（扁平化）操作，数据会先经过跟map()方法一样的操作，为每一条输入返回一个迭代器，然后将所得到的不同级别的迭代器中的元素全部当成同级别的元素，返回一个元素级别全部相同的RDD。这个转换操作通常用来切分单词。例如，分别用map()方法和flatMap()方法分割字符串。take()方法take(N)方法用于获取RDD的前N个元素，返回一个数组类型的数据。take()与collect()方法原理相似，collect()方法用于获取全部数据，take()方法获取指定个数的数据。例如，定义1个包含1到10的RDD，并获取RDD的前5个元素。filter()方法filter()方法是一种转换操作，用于根据特定条件过滤RDD中的元素，需要一个返回值为Boolean类型的过滤函数作为参数。filter()方法将返回值为true的元素保留，返回值为false的元素过滤掉，最终返回一个包含所有符合过滤条件元素的新RDD。例如，创建一个RDD，过滤掉每个元素（二元组）中第二个值小于等于1的元素。distinct()方法distinct()方法是一种转换操作，用于RDD的数据去重，去除完全相同的元素，无需参数。例如，创建一个带有重复数据的RDD，并使用distinct()方法去重。使用简单的集合操作RDD是一个分布式的数据集合，具有一些与数学中的集合操作类似的操作，如求并集、交集、补集和笛卡儿积等。Spark中的集合操作常用方法如下。方法描述union()参数是RDD，合并两个RDD的所有元素intersection()参数是RDD，求出两个RDD的共同元素subtract()参数是RDD，将原RDD里和参数RDD里相同的元素去掉cartesian()参数是RDD，求两个RDD的笛卡儿积union()方法union()方法用于将两个RDD合并成一个，不进行去重操作。合并时，两个RDD中每个元素的值的个数和数据类型需保持一致。例如，创建两个存放二元组的RDD，通过union()方法合并这两个RDD，不处理重复数据。intersection()方法intersection()方法用于求出两个RDD的共同元素，即找出两个RDD的交集，参数为另一个RDD，先后顺序与结果无关。例如，创建两个包含相同元素的RDD，通过intersection()方法求出这两个RDD的交集。subtract()方法subtract()方法用于删除前一个RDD中在后一个RDD出现的元素，即求补集，返回前一个RDD去除与后一个RDD相同元素后的新RDD。例如，创建两个RDD（分别为rdd1和rdd2，包含相同和不同元素），通过subtract()方法求彼此的补集，两个RDD的顺序会影响结果。cartesian()方法cartesian()方法将两个集合的元素两两组合，即求笛卡儿积。假设集合A有5个元素，集合B有10个元素，则结果返回50个元素组合。例如，创建两个RDD，分别有3个元素，通过cartesian()方法求两个RDD的笛卡儿积。创建SparkRDDRDD基础操作和进阶操作键值对RDD操作RDD连接操作RDD文件读写项目实施键值对RDD键值对RDD存储由键和值组成的二元组，RDD的基本转换操作同样适用于键值对RDD。由于键值对RDD包含二元组，传递的函数需从操作单个元素改为操作二元组。Spark的大部分RDD操作支持所有类型的单值RDD，但是也有少部分特殊的操作只能作用于键值对RDD。键值对RDD是由一组组的键值对组成的，键值对RDD也被称为PairRDD。键值对RDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口，如reduceByKey()、join()等方法。创建键值对RDD键值对RDD有多种创建方式。读取键值对类型数据时，可直接返回键值对RDD。若需将普通RDD转化为键值对RDD，可使用map()方法。以英语单词组成的文本行为例，提取每行首单词为键，整句为值，创建键值对RDD。keys与values方法键值对RDD包含键和值两部分。Spark提供keys和values两种方法，分别返回仅含键和仅含值的RDD。例如，通过keys和values方法分别查看kv_rdd的键与值。reduceByKey()方法当数据集以键值对形式呈现时，常需合并统计相同键的值。reduceByKey()方法用于合并具有相同键的值，作用对象是键值对，并且只对键的值进行处理。reduceByKey()方法需接收一个输入函数，键值对RDD中相同键的值将按此函数合并，生成新的RDD作为返回结果。首先，reduceByKey()方法先将相同键的前两个值（a、b）传给输入函数，生成新返回值（A）。随后，新返回值（A）与相同键的下一个值（c）再次传入函数，生成新的返回值（B）。此过程反复进行，直至每个键仅剩一个对应值。reduceByKey()方法不是一种行动操作，而是一种转换操作。reduceByKey()方法例如，定义一个含有多个相同键的键值对RDD，使用reduceByKey()方法对每个键的值进行求和。groupByKey()方法groupByKey()方法用于对具有相同键的值进行分组，可以对同一组的数据进行计数、求和等操作。对于一个由类型K的键和类型V的值组成的RDD，通过groupByKey()方法得到的RDD类型是[K,Iterable[V]]。例如，对rdd按键分组，查看各分组值，并统计每组值的数量。创建SparkRDDRDD基础操作和进阶操作键值对RDD操作RDD连接操作RDD文件读写项目实施join()方法将有键数据与另一组有键数据按键连接，是键值对数据的常用操作。与合并不同，连接会对键相同的值进行合并，连接方式多种多样，包含内连接、右外连接、左外连接、全外连接，不同的连接方式需要使用不同的连接方法。连接方法描述join()对两个RDD进行内连接rightOuterJoin()对两个RDD进行连接操作，确保第二个RDD的键必须存在（右外连接）leftOuterJoin()对两个RDD进行连接操作，确保第一个RDD的键必须存在（左外连接）fullOuterJoin()对两个RDD进行全外连接join()方法join()方法rightOuterJoin()方法leftOuterJoin()方法fullOuterJoin()方法zip()方法zip()方法用于将两个RDD组合成键值对RDD，要求两个RDD的分区数量和元素数量相同，否则会抛出异常。例如，将两个元素个数和分区个数都相同的非键值对RDD组合成一个键值对RDD。combineByKey()方法combineByKey()方法是Spark中一个比较核心的高级方法，键值对的一些其他高级方法的底层均是使用combineByKey()方法实现的，如groupByKey()方法、reduceByKey()方法等。combineByKey()方法用于将键相同的数据合并，且允许返回与输入数据的类型不同的返回值。combineByKey(createCombiner,mergeValue,mergeCombiners,numPartitions=None)combineByKey()方法接收3个重要的参数：createCombiner:V=>C，V是键值对RDD中的值部分，将该值转换为另一种类型的值C，C会作为每一个键的累加器的初始值。mergeValue:(C,V)=>C，该函数将元素V合并到之前的元素C(createCombiner)上（这个操作在每个分区内进行）。mergeCombiners:(C,C)=>C，该函数将两个元素C进行合并（这个操作在不同分区间进行)。combineByKey()方法例如，使用combineByKey()方法对一个含有多个相同键值对的数据求平均值。lookup()方法lookup(key:K)方法用于返回键值对RDD指定键的所有对应值。例如，通过lookup()方法查询rdd中键为panda的所有对应值。创建SparkRDDRDD基础操作和进阶操作键值对RDD操作RDD连接操作RDD文件读写项目实施读取与存储JSON文件Spark支持的一些常见文件格式：格式名称结构化描述JSON半结构化常见的基于文本的格式，半结构化；大多数库要求每行一条记录CSV结构化非常常见的基于文本的格式，通常在电子表格应用中使用SequenceFile结构化一种用于键值对数据的常见Hadoop文件格式ObjectFile非结构化用来存储Spark作业中的数据，改变类时会失效，因为对象文件依赖于Java序列化纯文本非结构化普通的文本文件，每一行一条记录读取与存储JSON文件JSON全称为JavaScriptObjectNotation，是JavaScript对象表示法。JSON是一种使用较广泛的半结构化数据格式，被设计用于可读的数据交换，是轻量级的文本数据交换格式。JSON解析器和JSON库支持许多不同的编程语言。JSON数据的书写格式是名称/值对形式。名称/值对包括字段名称（在双引号中）、冒号和值。数据由逗号分隔，花括号用于保存对象，方括号用于保存数组。读取与存储JSON文件读取JSON文件是指将JSON文件作为文本文件读取，再通过JSON解析器将RDD中的值映射为相应的数据结构。类似地，也可以通过JSON序列化库将数据转为字符串，再写出字符串作为JSON文件。Java和Scala可以使用Hadoop自定义格式操作JSON数据。要求文件每行是一条JSON记录，如果记录跨行，则需要读取整个文件，对文件进行解析。在Scala中有很多包可以实现JSON文件的读取。解析JSON文件通常需要将记录读入一个含有数据结构格式的类中，再根据这个格式解析JSON文件。读取与存储JSON文件读取JSON文件，将其作为文本文件，再对JSON数据进行解析。要求文件每行是一条JSON记录，如果记录跨行，则需要读取整个文件，对文件进行解析。在Scala中有很多包可以实现JSON文件的读取。解析JSON文件通常需要将记录读入一个含有数据结构格式的类中，再根据这个格式解析JSON文件。例如，解析JSON文件testjson.json，其数据如下。{"name":"jack","age":12}{"name":"lili","age":22}{"name":"cc","age":11}{"name":"vv","age":13}{"name":"lee","age":14}读取JSON文件自定义一个Person类，“implicitvalformats”定义了隐式参数formats，该参数是parse()方法和extract()方法转换数据所依赖的参数。其中DefaultFormats是org.json4s提供的一种默认转换类型。如果未使用parse(x).extract[Person]，那么代码会出错。存储JSON文件JSON文件的存储比读取更加简单，不需要考虑格式错误问题，只需将结构化数据解析成的RDD转化为字符串RDD，再使用Spark的文本文件API写入即可。例如，将解析后的JSON文件重新转化成JSON文件并保存，结果采用RDD的repartition()方法将多个分区数据写到一个分区中。importorg.json4s.JsonDSL._valdata_json=data.map{x=>("name"->)~("age"->x.age)}valdata_string=data_json.map{x=>compact(render(x))}sc.parallelize(data_string.toSeq).repartition(1).saveAsTextFile("/tipdm/data/json_out")读取与存储CSV文件逗号分隔值（CommaSeparatedValues，CSV）文件每行都有固定数目的字段，字段间用逗号隔开。在制表符分隔值文件（TabSeparatedValue，TSV）中用制表符隔开。若CSV文件所有数据字段不含换行符，可使用textFile()方法读取并解析。同读取JSON文件一样，读取CSV文件时，先读取文本，再通过解析器解析数据。读取CSV文件读取CSV文件时，需先将文件作为文本文件读取，再对数据进行处理。以CSV格式的数据文件testcsv.csv为例，在spark-shell中读取testcsv.csv文件的数据。0firstfirstline1secondsecondline读取CSV文件若字段中嵌有换行符，需完整读入文件后再解析各字段。如果文件很大，那么读取和解析的过程可能会成为性能瓶颈。读取嵌有换行符的CSV文件时，需根据数据结构定义一个Data类，将文件数据加载到该类中，数据按类结构读取。存储CSV文件CSV文件存储较为简单，可通过重用输出编码器加速。CSV格式数据输出时不记录每条数据的字段名，需创建映射关系以保持输出顺序一致。一种方法是通过函数将各字段转化为指定顺序的数组，但需注意数据字段必须是已知的。例如，对转化后的数据进行存储。importjava.io.{StringReader,StringWriter}import.bytecode.opencsv.{CSVReader,CSVWriter}importscala.jdk.CollectionConverters._data.map(data=>List(data.index,data.title,data.content).toArray).mapPartitions{data=>valstringWriter=newStringWriter()valcsvWriter=newCSVWriter(stringWriter)csvWriter.writeAll(data.toList.asJava)Iterator(stringWriter.toString)}.saveAsTextFile("/tipdm/data/csv_out")存储SequenceFile文件SequenceFile是由无固定结构的键值对组成的常用Hadoop文件格式。SequenceFile文件的存储非常简单，首先需要确保有一个键值对RDD，直接调用saveAsSequenceFile()方法保存数据，可以自动将键和值的类型转化为Writable类型。例如，将一个存储二元组的列表转化为RDD，其中二元组的第一个值表示动物名，第二个值表示该动物的数量，将RDD存储为序列化文本。importorg.apache.hadoop.io.{IntWritable,Text}valrdd=sc.parallelize(List(("Panda",3),("Monkey",6),("Snail",2)))rdd.repartition(1).saveAsSequenceFile("/tipdm/data/Sequence_out")存储SequenceFile文件在HDFS的Web端中查看保存结果（/tipdm/data/Sequence_out/part-00000），如下所示。显示的数据虽然是乱码，但存储并没有问题。数据存储时，键的类型转化为Text类型，值的类型转化为IntWritable类型。读取SequenceFile文件Spark有专门读取SequenceFile文件的接口，例如，SparkContext类中的sequenceFile(path,keyClass,valueClass,minPartitions)方法。SequenceFile文件的数据类型是Hadoop的Writable类型，所以keyClass和valueClass参数必须定义为正确的Writable类。例如，读取SequenceFile文件。数据有两个字段，分别是动物名和动物数量，因此，keyClass是Text类型的，valueClass是IntWritable类型的。存储ObjectFile文件ObjectFile是Spark特有的格式，用于将RDD中的元素基于Java序列化后存储为二进制文件。ObjectFile文件与SequenceFile文件类似，二者均采用二进制序列化的形式对数据进行存储。SequenceFile要求键值对必须实现Hadoop的Writable接口，结构相对复杂；而ObjectFile文件中直接存储序列化Java对象。ObjectFile文件的存储使用的是saveAsObjectFile()方法，由于是将整个RDD中的元素直接序列化，因此要求RDD中的元素必须是可序列化的，即必须实现Serializable接口。存储ObjectFile文件例如，定义一个样例类Person，类中包括String类型的姓名和Int类型的年龄两个属性，随后基于包括Person类的实例集合创建RDD。由于Scala的样例类在底层自动实现了Serializable接口，因此Person类对象默认便是可序列化的，可以将RDD保存为ObjectFile文件。caseclassPerson(name:String,age:Int)valrdd=sc.parallelize(Seq(Person("Alice",18),Person("Bob",19),Person("Charlie",20)))rdd.repartition(1).saveAsObjectFile("/tipdm/data/Object_out")存储ObjectFile文件在HDFS的Web端中查看保存结果（/tipdm/data/Object_out/part-00000），如下所示。由于是序列化后的数据，因此数据看起来同样像是乱码。读取ObjectFile文件读取ObjectFile文件同样与读取SequenceFile文件类似，读取SequenceFile文件使用的是sequenceFile()方法，读取ObjectFile文件用的则是objectFile()方法。读取ObjectFile文件的过程本质上是反序列化的逆过程，程序需要确知如何将字节流还原为对象，因此，必须在方法中显式指定元素的类型。例如，读取ObjectFile文件。读取时需传入[Person]泛型标记，这样程序才能正确地将二进制数据反序列化成Person对象实例。读取与存储纯文本文件文本文件是一种典型的顺序文件，从文件的逻辑结构来看，又属于流式文件。文本文件中只能存储文件有效字符信息，包括能用ASCII字符表示的回车符、换行符等信息，不能存储其他信息。因此，文本文件无法存储声音、动画、图像、视频等多媒体信息。在Windows系统中，文本文件的扩展名是“.txt”。读取与存储纯文本文件文本文件可通过textFile()方法即可直接读取，一条记录（一行）作为一个元素。例如，读取HDFS的/tipdm/data目录下的文本文件bigdata.txt。文本文件的存储也是很常用的，对数据进行处理之后，通常需要将结果保存以用于分析或存储。RDD数据可以直接调用saveAsTextFile()方法将数据存储为文本文件。例如，将创建的RDD数据存储为一个文本文件，并通过repartition()方法将分区数量设置为1。rdd.repartition(1).saveAsTextFile("/tipdm/data/bigdata_out")创建SparkRDDRDD基础操作和进阶操作键值对RDD操作RDD连接操作RDD文件读写项目实施读取员工薪资数据创建RDD本任务将读取员工2023年上、下半年薪资数据，并创建RDD。由于数据比较多，因此适合通过读取存储在HDFS上的数据来创建RDD。将数据上传至HDFS的/tipdm/data目录下，并在spark-shell中读取HDFS上的员工上、下半年薪资数据创建RDD，之后查看数据前五行示例。查询上半年实际薪资Top3的员工姓名本任务将查询上半年实际薪资排名前3的员工信息。为此，需要对上半年的实际薪资进行排序。由于创建RDD时，textFile()方法是将每一行数据作为一条记录存储的，所以在排序前需要先对数据进行转换，具体实现步骤如下：读取CSV文件，将第一行字段名称删除。将数据按分隔符“,”分隔，取出第2列员工姓名和第7列实际薪资数据，并将实际薪资数据转换成Int类型数据。通过sortBy()方法根据实际薪资进行降序排列。通过take()方法获取上半年实际薪资排名前3的员工信息。查询上半年或下半年实际薪资大于20万元的员工姓名本任务将输出上半年或下半年实际薪资大于20万元的员工姓名，首先需要过滤出两个RDD中实际薪资大于20万元的员工姓名，再将两个RDD得到的员工姓名合并到一个RDD中，对员工姓名进行去重。valfilter_first=split_first.filter(x=>x._2>200000).map(x=>x._1)valfilter_second=split_second.filter(x=>x._2>200000).map(x=>x._1)valname=filter_first.union(filter_second).distinct()name.collect().foreach(println)输出每位员工2023年的总实际薪资本任务将统计每位员工2023年的总实际薪资，首先需要将数据合并到一个RDD中，通过相同的键对同一个员工的上半年实际薪资和下半年实际薪资进行累加，具体实现步骤如下：获取上、下半年员工薪资数据并将其转换为RDD，分别为split_first