大数据分析及应用项目教程(Spark SQL)（微课版） 课件全套 许慧 第1-6章 大数据分析概述、实践环境准备- Zepplin数据可视化

第1章大数据分析概述关于大数据分析第一部分什么是大数据分析大数据分析可视化大数据分析工具学习目标和要求1、了解大数据分析的概念、特点、类别、优缺点。2、知道大数据分析的相关工具。3、了解大数据分析可视化的概念及相关工具。什么是大数据分析12345Volume（容量）Velocity（速度）Variety（种类）Value（价值）Veracity（真实性）1、大数据的“5V”特征大数据是指无法在一定时间范围内用常规软件工具进行捕捉、管理和处理的数据集合，大数据分析就是指对规模巨大的数据进行数据分析。什么是大数据分析2、大数据分析概念数据分析量大1234数据处理速度快数据分析类型多数据价值密度低3、大数据分析的特点BigDataAnalysis5数据的可靠性低什么是大数据分析什么是大数据分析4、大数据分析类别预测分析关注的是对未来事件的预测。预测性分析规范性分析是指在发生问题之后，根据问题诊断性分析之后，结合预测性分析，做出相应的优化建议和行动。规范性分析针对过去已经发生的事情，分析该事件产生的原因。诊断性分析描述性分析是描述过去的数据，基于历史数据描述发生了什么，对过去的大量历史数据进行汇总分析描述，以简单可读的方式进行呈现。描述性分析为优质决策提供参考；提高产品开发创新力；改善客户服务体验；提升风险管理优势缺点信息透明化成本高数据质量低技术更新变化快什么是大数据分析5、大数据分析的优势与缺点大数据分析工具（1）ApacheSpark：具有SparkSQL、Streaming实时计算、机器学习和SparkGraphX图计算的内置功能。（2）Hbase：HBase是一个基于HDFS的面向列的分布式数据库。（3）Storm：Storm是流处理的代表性实现之一。Storm具有低延迟、高性能、分布式、可扩展、容错、可靠性、快速等特点。（4）Flink：

Flink是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。1、Hadoop生态圈中的大数据分析工具123编程语言Scala语言：Scala语言是基于JVM运行环境、面向对象和函数式编程的完美结合Python语言：Python在数据分析领域也是一个强大的语言工具。R语言：是大数据分析工具之一，可用于科学计算、统计分析、数据可视化等。大数据分析工具2、大数据分析编程语言RapidMiner其特点是拖拽操作，无需编程，运算速度快，具有丰富数据挖掘分析和算法功能，常用于解决各种商业关键问题。12MongoDB是一个基于分布式文件存储的数据库。旨在为WEB应用提供可扩展的高性能数据存储解决方案。大数据分析工具3、其他工具

数据可视化是利用计算机以图形图表的形式将原始的抽象信息和数据直观的表示出来。

大数据分析可视化工具有很多，比如Zeppelin、PowerBI、Tableau、Spass等等。大数据分析可视化认识SparkSQL第二部分SparkSQL背景简介SparkSQL运行原理学习目标和要求1、了解SparkSQL的背景、特点。2、知道SparkSQL的运行架构。3、掌握Catalyst查询编译器的工作流程。4、掌握SparkSQL运行流程。HiveSharkSparkSQLHive是最原始的SQL-on-Hadoop工具。是Facebook开发的构建于Hadoop集群之上的数据仓库应用，它提供了类似于SQL语句的HQL语句作为数据访问接口脱离了Hive的依赖，SparkSQL在数据兼容、组件扩展、性能优化方面都得到了极大的提升。Shark是使用Scala语言开发的开源SQL查询引擎。其设计目标是作为Hive的补充，性能比Hive提高了10-100倍。但是Shark对于Hive依旧存在很多的依赖。SparkSQL背景简介1、SparkSQL的背景多种性能优化技术组件扩展性用户可以对SQL的语法解析器、分析器以及优化器进行重新定义和开发，并动态扩展。采用内存列存储（In-MemoryColumnarStorage），字节码生成技术（byte-codegeneration），CostModel对查询操作进行动态评估、获取最佳物理计划等。支持多种数据源可以在Hive上运行SQL或者HQL;可以从RDD、parquet文件、JSON文件中获取数据。SparkSQL背景简介2、SparkSQL的特点SparkSQL背景简介多种性能优化技术内存列存储（In-MemoryColumnarStorage）JVM对象存储和内存列存储对比SparkSQL背景简介多种性能优化技术字节码生成技术（byte-codegeneration）例如执行selecta+bfromtable这条命令通用的SQL方法：首先将生成一个表达式，并多次调用虚函数。SparkSQL：在其catalyst模块的expressions中增加了codegen模块。使用动态字节码生成技术来优化其性能，对匹配的表达式采用特定的代码动态编译，然后运行。SparkSQL运行原理1、SparkSQL的运行架构SparkSQL的整体架构SparkSQL是由Catalyst，Core，Hive和Hive-Thriftserver四个子项目组成。SparkSQL运行原理（1）Catalyst：负责处理整个查询过程，包括解析、绑定、优化等，将SQL语句转换成物理执行计划。（2）Core：用于将Catalyst的逻辑查询计划转换为SparkRDD代码。（3）Hive：Hive组件包括HiveContext和SQLContext，允许用户使用HiveQL的子集编写查询。（4）Hive-Thriftserver：支持HiveServer和CLI。SparkSQL运行原理2、Catalyst查询编译器（1）Catalyst的组成：Parser、Analyzer、Optimizer、Planner（2）Catalyst的工作流程Catalyst运行流程SparkSQL运行原理3、SparkSQL的运行原理（1）传统SQL的运行流程词法和语法解析（Parse）绑定（Bind）优化（Optimize）执行（Execute）SparkSQL运行原理（2）SparkSQL运行流程SessionCatalog保存元数据ANTLR生成未绑定的逻辑计划Analyzer绑定逻辑计划Optimizer优化逻辑计划SparkPlanner生成可执行的物理计划CostModel选择最佳物理执行计划execute执行物理计划小结

本章首先对大数据分析进行了介绍，详细阐述了大数据分析的相关概念、特点、类别及优缺点；简单介绍了大数据分析的常用工具；并对大数据分析可视化的优势价值等进行了分析。然后对大数据分析工具中的SparkSQL展开了详尽的描述，包括SparkSQL的发展演变历程、特点、运行架构原理等内容。THANKS!第2章实践环境准备Hadoop集群环境搭建第一部分环境准备启动Hadoop集群运行经典案例wordcount安装Hadoop学习目标和要求1、掌握Hadoop集群环境搭建的环境准备工作，包括配置主机名、防火墙设置、免密登录设置、Java环境设置。2、掌握安装Hadoop的过程、配置文件设置及启动集群的方法。3、会在Hadoop集群运行经典案例wordcount。环境准备1、集群节点规划此集群由三个节点构成，分别是master、slaver01、slaver02。集群搭建部署均在虚拟机中完成，使用VMwareWorkstation16Pro虚拟计算机软件。环境准备2、配置主机名和IP（1）修改对应虚拟机的IP地址：

vi/etc/sysconfig/network-scripts/ifcfg-ens33（2）将三台虚拟机的主机名修改为master、slaver01、slaver02。

hostnamectlset-hostname主机名（3）重启网络，使网络配置生效。

systemctlrestartnetwork环境准备3、连接MobaXterm终端工具使用MobaXterm终端工具，为master、slaver01和slaver02创建SSH连接。环境准备4、关闭防火墙

防火墙是对服务器进行保护的一种服务，但有时候会带来很多麻烦，它会妨碍Hadoop集群间的相互通信，所以我们要关闭防火墙。关闭master、slaver01和slaver02主机的防火墙，并设置开机不自启。systemctlstatusfirewalldsystemctlstopfirewalldsystemctldisablefirewalldsystemctlstatusfirewalld环境准备5、关闭SElinuxSELINUX是对系统安全级别更细粒度的设置。关闭master、slaver01和slaver02主机的SElinux。vi/etc/sysconfig/selinux环境准备6、修改/etc/hosts文件修改master、slaver01和slaver02主机的/etc/hosts文件，建立主机和ip地址之间的映射关系。vi/etc/hosts7、配置免密登录首先在master节点创建生成密钥。将密钥拷贝到slaver01和slaver02两个节点，完成免密登录配置。环境准备8、配置Java环境在master节点上传JDK软件包并解压。在.bash_profile文件中配置环境变量。使用source.bash_profile命令，使.bash_profile文件配置生效。查看Java版本，验证安装成功。使用scp命令将jdk解压安装相关文件分发到slaver01和slaver02节点。安装Hadoop

由于三个节点都需要安装Hadoop，为了提高部署效率，先在master节点进行部署安装，然后将相关的文件和配置拷贝分发到另外两个节点中。上传Hadoop安装包并解压。修改环境变量。使环境变量生效。修改hadoop的配置文件hadoop-env.sh、core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml、workers。分发Hadoop和环境变量文件到slaver01和slaver02。格式化HDFS文件系统。启动Hadoop集群1、启动Hadoop集群只需要在master节点输入start-all.sh。接着查看各节点的服务进程。2、打开Google浏览器，输入地址30:8088，可以打开Yarn页面。3、输入地址30:9870，可以打开HDFS页面。案例wordcount通过经典案例wordcount，体验Hadoop运行MapReduce计算。1、在本地root目录下创建一个txt文件，输入一段自定义文字。2、上传文本到到hdfs。案例wordcount3、运行命令并查看结果hadoopjarhadoop-mapreduce-examples-3.2.1.jarwordcount/input/outputSpark集群部署与使用第二部分Spark安装启动SparkSpark集群测试学习目标和要求1、掌握Spark集群安装配置方法。2、会启动关闭Spark集群。3、能使用Spark-shell进行简单编程测试。Spark安装1、上传软件包使用MobaXterm工具，将软件包上传至master节点的root目录下，然后将其解压到/usr/local目录中。Spark安装2、在.bash_profile文件中修改环境变量执行source.bash_profile使环境变量生效Spark安装3、修改Spark配置文件（1）配置spark-env.sh文件（2）配置workers文件4、复制修改spark启动脚本，避免和hadoop的启动脚本冲突。5、分发Spark安装文件到slaver01和slaver02。启动Spark1、在master节点输入start-spark-all.sh，启动Spark。2、查看各节点服务进程状态（此处已经启动Hadoop集群）启动Spark3、Web查看Spark主页情况在浏览器中输入地址30:8080/，进行查看。Spark集群测试1、使用spark-submit工具提交Spark作业spark-submit提交任务及参数说明：--class：应用程序的主类，仅针对java或scala应用。--master：master的地址，提交任务到哪里执行，例如spark://host:port,yarn,local。--driver-memory：driver进程所使用的内存数量，以字节为单位。可以指定不同的后缀如“512m”或“15g”，默认是1G。--executor-memory：executor使用的内存数量，以字节为单位。可以指定不同的后缀如“512m”或“15g”，默认是1G。--total-executor-cores：所有executor总共的核数。仅仅在mesos或者standalone下使用。Spark集群测试Spark集群测试2、使用Spark-sql（1）输入spark-sql命令，启动spark-sql。（2）使用SQL命令，执行创建数据库、创建表等操作。Spark集群测试3、使用Sparkshell（1）输入spark-shell命令，启动spark-shell。（2）执行SparkShell常用命令Spark集群测试（2）执行SparkShell常用命令①:help命令，查看SparkShell常用的命令Spark集群测试（2）执行SparkShell常用命令②:paste命令，进入paster模式Scala安装第三部分下载安装包安装配置学习目标和要求1、会下载、安装配置Scala工具。2、能启动Scala进行编程测试。下载安装包通过连接地址/download/2.13.6.html直接下载scala-2.13.6.tgz压缩包。安装配置1上传安装包并解压2配置环境变量，添加Scala相关配置安装配置3使环境变量生效，查看Scala版本4启动Scala小结

本章通过详细的操作步骤与结果分析，介绍Hadoop集群环境搭建、Spark集群部署与使用、Scala的安装运行。详细阐述了各个集群环境搭建中的相关配置文件。THANKS!第3章学生信息处理分析班级基本情况分析第一部分学生所属班级和男女生数量以班级为单位整理学生信息情境导入

现有一份某校信息工程学院所有班级学生的基本情况数据，如下图所示。数据包含8个字段，分别为学号、姓名、性别、年级、班级、语文成绩、数学成绩、英语成绩。当下辅导员想通过一位学生的学号获知其所属的班级，同时也想对各班级基本情况进行了解，以便后续的工作开展。学习目标和要求掌握Scala判断与循环、函数式编程。掌握Scala集合操作。掌握Scala数据类型、常量与变量、运算符、数组。学生所属班级和男女生数量——理论基础1、数据类型数据类型类型描述Byte表示8位有符号补码整数。其数值区间从-128到127。Short表示16位有符号补码整数。其数值区间为-32768到32767。Int表示32位有符号补码整数。其数值区间为-2147483648到2147483647。表示只能保存整数。Long表示64位有符号补码整数。其数值区间为-9223372036854775808到9223372036854775807。Float表示32位IEEE754标准的单精度浮点数。如果浮点数后面有f或者F后缀时，表示这是一个Float类型，否则就是一个Double类型的。Double表示64位IEEE754标准的双精度浮点数。Char表示16位无符号Unicode字符，区间值为U+0000到U+FFFF。String表示字符序列。即字符串，使用时需要使用双引号包含一系列字符。Boolean其值为true或false。Unit表示无值，和其他语言中void等同。用于不返回任何结果的方法的结果类型。Unit只有一个实例值，写成()。Nullnull或空引用NothingNothing类型在Scala的类层级的最底端，是任何其他类型的子类型。AnyAny是所有其他类的超类。AnyRefAnyRef类是Scala里所有引用类(referenceclass)的基类。Scala与Java的数据类型相同，但是Scala中的数据类型都是对象即Scala没有java中的原生类型。因此Scala可以对数字等基础类型调用方法。学生所属班级和男女生数量——理论基础2、常量与变量1）定义方法在Scala中，定义常量的关键字是“val”，语法格式如下：

val常量名称:数据类型=初始值val一经初始化就无法再进行修改，否则会报错。在Scala中，定义变量的关键字是“var”，语法格式如下：

var变量名称:数据类型=初始值学生所属班级和男女生数量——理论基础2）数据类型推断

Scala具备数据类型推断的功能，因此在定义常量或者变量时，可以不用特地说明数据的类型，即可以不在常量或变量名称后面添加“:type”数据类型项。3）多变量声明学生所属班级和男女生数量——理论基础A.若声明的多个变量初始值相同，则只需用逗号分隔：B.若声明的多个变量初始值不相同，则需将它们声明为元组：3、运算符运算符类别运算符描述举例算数运算符举例中假设a=1，b=2算数运算符

+加号a+b运算结果为3-减号a-b运算结果为-1*乘号a*b运算结果为2/除号b/a运算结果为2%取余b%a运算结果为0Scala中运算符即方法、方法即运算符。学生所属班级和男女生数量——理论基础学生所属班级和男女生数量——理论基础关系运算符举例中假设a=1，b=2关系运算符==等于a==b运算结果为false>大于a>b运算结果为false<小于a<b运算结果为true>=大于等于a>=b运算结果为false<=小于等于a<=b运算结果为true!=不等于a!=b运算结果为true逻辑运算符举例中假设变量a为1，b为0逻辑运算符&&与，当两个条件均成立则为真，否则为假a&&b运算结果为false||或，当两个条件有一个成立则为真，否则为假a||b运算结果为true！非，对当前条件取反!(a&&b)运算结果为true学生所属班级和男女生数量——理论基础位运算符举例中假设a=00111100，b=00001101位运算符&按位与，数据按照二进制位进行运算，两位均为1则结果为1，否则为0a&b的运算结果为00001100|按位或，数据按照二进制位进行运算，两位有一个为1则结果为1a|b的运算结果为00111101^按位异或，数据按照二进制位进行运算，两位不同时结果为1，相同时为0a^b的运算结果为00110001~按位取反，数据按照二进制位进行运算，是1则变为0，是0则变成1~a的运算结果为11000011<<数据按照二进制位进行运算，左移a<<2的运算结果为11110000>>数据按照二进制位进行运算，右移a>>2的运算结果为00001111>>>数据按照二进制位进行运算，无符号右移a>>>2的运算结果为00001111学生所属班级和男女生数量——理论基础赋值运算符=指定右边操作结果赋值给左边c=a+b将a+b的运算结果赋值给c+=左右两侧相加后赋值给左边c+=a即c=c+a-=左右两侧相减后赋值给左边c-=a即c=c-a*=左右两侧相乘后赋值给左边c*=a即c=c*a/=左右两侧相除后赋值给左边c/=a即c=c/a%=左右两侧求余后赋值给左边c%=a即c=c%a<<=按位左移后再赋值给左边c<<=a即c=c<<a>>=按位右移后再赋值给左边c>>=a即c=c>>a&=按位与运算后赋值给左边c&=a即c=c&a|=按位或运算后再赋值给左边c|=a即c=c|a^=按位异或运算后再赋值给左边c^=a即c=c^a学生所属班级和男女生数量——理论基础4、数组

在Scala语言中，提供了一种数据结构叫作数组，数组是用于存储相同类型变量的集合。数组分为定长数组(Array)和变长数组(ArrayBuffer)。1）声明与定义方式使用new的完整形式，其语法如下：vararrname:Array[type]=newArray[type](size)使用new的简写形式，其语法如下：vararrname=newArray[type](size)使用定义并初始化的方式，其语法如下：vararrname=Array(element1,element2,element3)变长数组：定义与声明方式与定长数组的相同，但是需要先导入importscala.collection.mutable.ArrayBuffer依赖包。定长数组：学生所属班级和男女生数量——理论基础2）基本操作方法

数组的基本操作方法：基本操作描述arr.sum对于Int数据类型的数组，返回数组各元素的和arr.max对于Int数据类型的数组，返回数组各元素的最大值arr.min对于Int数据类型的数组，返回数组各元素的最小值arr.length返回数组的长度arr.sorted.toBuffer对于Int数据类型的数组，返回数组各元素由小到大排序结果arr.reverse.toBuffer对于Int数据类型的数组，返回数组各元素倒序arr.contains(x)判断数组中是否包含x元素arr.isEmpty判断数组是否为空学生所属班级和男女生数量——理论基础5、判断与循环Scala的判断与循环语法与其他计算机语言一样。if判断可以对数据进行过滤判断处理，while和for循环可以在某一条件下将某段代码进行重复执行。1）if判断if判断有三种形式，分别为if语句、if…else语句、if…elseif…else语句和if…else嵌套语句。2）循环在Scala语言中循环也有三种方式，分别为while循环、do...while循环和for循环。学生所属班级和男女生数量——理论基础6、函数式编程1）函数的定义和调用def函数名[参数列表]:[返回值类型]={

函数体return[表达式]}函数调用的方法如下：

函数名[参数列表]函数的定义方法：学生所属班级和男女生数量——理论基础2）匿名函数匿名函数是指不含函数名称的函数。使用“=>”定义，“=>”的左边为参数列表，“=>”右边为函数体表达式。匿名函数虽然没有函数名，但是可以将其赋值给一个常量或者变量，然后通过常量或者变量名进行函数调用。学生所属班级和男女生数量——理论基础3）高阶函数

高阶函数是指使用其他函数作为参数，或者使用函数作为输出结果的函数。举例：使用其他函数作为参数的高阶函数学生所属班级和男女生数量——理论基础举例，使用函数作为输出结果的高阶函数学生所属班级和男女生数量——理论基础4）闭包闭包是一个函数，是可以访问一个函数里面局部变量的另外一个函数。定义一个函数：valmore=2defaddmore=(x:Int)=>x+moreaddmore(3)定义addmore函数的过程就是“捕获”其自由变量从而“闭合”该函数的动作。学生所属班级和男女生数量——理论基础5）函数柯里化

函数编程中，接受多个参数的函数都可以转化为接受单个参数的函数，这个转化过程就是柯里化。柯里化本身也用到了闭包。//未柯里化方式defmul(x:Int,y:Int)=x*ymul(2,2)//闭包方式defmulclosure(x:Int)=(y:Int)=>x*yprintln(mulclosure(3)(4))//函数柯里化方式defmulcurry(x:Int)(y:Int)=x*yprintln(mulcurry(5)(6))学生所属班级和男女生数量——理论基础6）嵌套函数嵌套函数即在函数内部定义函数，其中内部函数称为局部函数。学生所属班级和男女生数量——编程分析实现1、判断学生所属班级

完成此任务首先需要用数组存储各个班级学生的学号，然后定义函数识别某个学生的学号获取其所属班级。1）定义班级数组2）定义distinguish函数，利用if…elseif…else语句进行判断识别学号。2、统计男女生人数信息

定义函数count(sex:String)统计男女生人数。在count(sex:String)函数中利用数组存储数据信息，通过for循环遍历数组并利用if判断数组中元素是否包含指定性别，并进行sum计数。以班级为单位整理学生信息——理论基础1、列表Scala列表与数组类似，所有元素的类型都相同。它们的区别首先是列表是不可变的，其次列表具有递归的结构而数组不是。1）定义列表定义不可变列表语法：//创建一个包含值1,值2,值3,值...的不可变列表var/val变量名:List[元素类型]=List[元素类型](值1,值2,值3,值...)//使用“::”拼接方式来创建列表，必须在最后添加一个Nilval/var变量名=值1::值2::Nil以班级为单位整理学生信息——理论基础//导入相关依赖包importscala.collection.mutable.ListBuffer//创建一个空的可变列表val/var变量名:ListBuffer[类型]=newListBuffer[类型]()//创建一个包含值1，值2，值3...的可变列表val/var变量名=ListBuffer(值1，值2，值3...)定义可变列表语法：以班级为单位整理学生信息——理论基础2）列表的常用操作以下是可变和不可变列表的常用操作。方法（对列表a和b进行操作）描述a.isEmpty判断列表是否为空a++b拼接两个列表a.head获取列表的首个元素a.tail获取列表除首个元素以外的剩余部分a.reverse反转列表a.take(num)获取指定个数的前缀a.drop(num)删除从左边开始的num个元素a.toString转换字符串a.toArray将列表转换为数组List.concat(a,b)合并两个列表ersect(b)对两个列表取交集a.diff(b)对两个列表取差集a(索引值)获取列表中指定索引值的元素以班级为单位整理学生信息——理论基础可变列表的常用操作方法（对列表a和b进行操作）描述a+=元素为列表添加元素a++=b追加一个列表a-=元素删除列表中的元素a.toList将可变列表转换为不可变列表

虽然可变和不可变列表都有添加、删除、合并等操作，但是两者有一个非常大的差别。对不可变列表进行操作，会产生一个新的列表，原来的列表并没有改变。对可变列表进行操作，改变的是该列表本身。2、Set集合以班级为单位整理学生信息——理论基础Set是无序，不可重复的集合。Set集合分为可变的和不可变的。Scala默认创建的是不可变的Set集合，若要创建可变的Set集合，需要先导入依赖包。1）定义Set集合

val

变量名:Set[类型]=Set[类型](值1,值2,值3...)举例：vala=Set(9,2,5,4,8)以班级为单位整理学生信息——理论基础可变和不可变Set集合的常用操作方法方法（对Set集合a或b进行操作）描述a.drop(num)删除从左边开始的num个元素a.contains(元素)判断集合是否包含此元素a.size集合的大小a++b合并Set集合2）Set的基本操作以班级为单位整理学生信息——理论基础不可变Set集合的添加/删除操作方法方法（对不可变Set集合a和b进行操作）描述a+元素添加元素a-元素删除元素可变Set集合的添加/删除操作方法方法（对可变Set集合a和b进行操作）描述b.add(元素)添加元素b.remove(元素)删除元素b+=元素添加元素b-=元素删除元素a++=b元素合并Set集合3、元组以班级为单位整理学生信息——理论基础

元组是不同类型值的聚集,它可以将不同类型的值放在一个变量中进行存储。定义元组的语法一：valtuple=(元素1,元素2,元素3…)定义元组的语法二：valt=newTuplen(元素1,元素2,元素3...元素n)

当需要访问元组中的某个元素的值时，可以通过“元组名_元素索引”进行访问。4、Map映射以班级为单位整理学生信息——理论基础1）映射定义：

映射是一系列键值对的集合。映射中的键都是唯一的，并且可以通过键获取值。可变映射的定义需要先导入相关依赖包。2）映射的常用操作：方法（对映射a和b进行操作）描述a.keys获取所有的keya.values获取所有的valuea(key)使用key获取value，如果key不存在会报错a.contains(key)查看Map中是否存在指定的keya.isEmpty检查map是否为空a++b合并两个map，如果key重复，++后者会替换掉前边的key对应的value5、函数组合器以班级为单位整理学生信息——理论基础

函数组合器在实际操作中非常实用。组合器包括map、filter、foreach、groupBy、flatten、drop、zip等。组合器可以对集合中的每个元素进行指定的操作，并将结果形成集合输出。组合器描述map将某个函数应用到中的每个元素，并将结果形成集合输出。filter指定条件，对集合中的元素进行过滤foreach对集合中的每个元素进行作用，但是没有返回值。groupBy对集合中的元素进行分组操作，得到一个Map。flatten可以把嵌套的结构展开。drop去掉集合前面的n个元素zip将两个集合结合在一起以班级为单位整理学生信息——编程分析实现

以班级为单位整理学生信息，可以先将数据保存到列表中，然后使用groupBy组合器以班级为条件进行分组。1）在scala命令输入行输入以下命令，创建列表保存数据。valdata=List("21103,严林石,男,21级,大数据1班,89,97,91","21108,庄伟康,男,21级,大数据1班,67,94,87","21208,陈心雨,女,21级,大数据2班,84,92,91","21107,邹和俊,男,21级,大数据1班,77,83,88")以班级为单位整理学生信息——编程分析实现2）接着应用groupBy组合器进行分组，分组条件是班级。因此需要选取班级列进行操作，使用“,”对列表中每条数据进行split分隔，班级数据是在第4列。由此实现了以班级为单位整理学生信息。学生基本情况获取第二部分学生特长情况学生成绩情况分析情境导入

在掌握了前一节中Scala的基本知识后，李雷同学提出了新的疑问。Scala作为一门纯粹的面向对象的语言它的类和对象又有什么特点？定义和使用方法又是如何的？通过接下来对Scala类和对象、模式匹配等知识的学习，能对学生特长情况与成绩情况进行分析。学习目标和要求1、了解Scala类和对象、模式匹配的基本概念。2、掌握Scala类和对象、模式匹配的定义及操作方法。3、能综合应用Scala基础知识编写应用程序进行数据分析。1、类和对象学生特长情况——理论基础

类和对象是Scala面向对象语言的两个重要概念。类是对象的抽象，而对象是类的具体实例。类的定义方式如下：Class类名称(参数列表){//定义类的字段和方法}

继承是面向对象开发语言中的一个概念，继承可以使得子类具有父类的属性和方法或者重新定义、追加属性和方法等。可以使得子类别具有父类别的各种属性和方法，而不需要再次编写相同的代码，可以有效复用代码。但是Scala是单继承，只能继承一个父类。继承类需要用到关键字extends。学生特长情况——理论基础继承的语法如下：class子类名extends父类名{类体}注意：子类继承父类中非抽象的方法（即已经实现的方法）必须使用override修饰符。2、单例对象和伴生对象学生特长情况——理论基础1）单例对象单例对象是一种特殊的类，有且只有一个实例。定义一个单例对象的语法如下：object单例对象名{}2）半生对象

若在同一个代码文件内部，同时出现了classA和objectA，即类名和单例对象名完全相同，那么它们两者就互为伴生关系。3、模式匹配学生特长情况——理论基础

模式匹配（patternmatching）是Scala中十分强大的一个语言特性。模式匹配基本语法如下所示：xmatch{ casepattern1=>doSomething casepattern2=>doothers...}学生特长情况——编程分析实现

学生特长情况数据speciality.txt包括三个字段，分别是学生学号、姓名、特长类别。现要获知某一类特长的学生情况，可以先定义一个object单例对象。在此对象中读取数据文件，并转换成数组。接着遍历数组，搜索包括某类特长的学生数据并打印。思路分析：综合案例——学生成绩情况分析data.txt原始数据字段信息字段含义stuid学号name姓名sex性别grade年级class班级Chinese语文成绩Math数学成绩English英语成绩

现要求使用Scala函数式编程，综合本章前面所学知识，进行学生成绩分析统计。获得各门课程的平均成绩和及格人数（大于60），每个同学所有课程的总成绩。小结

本章详细介绍Scala的一些基本知识、操作方法，包括函数、表达式、判断与循环、数据结构、类和对象等。并完成了“班级基本情况分析”和“学生基本情况获取”。THANKS!第4章房产大数据分析与探索某房产公司销售人员业绩分析第一部分数据集处理数据操作分析情境导入

员工业绩考核分析是企业工作总结的重要组成部分。从业绩分析中员工可以清楚的知道自己的工作应该达到何种标准，发现自己的工作长处和不足，激发工作的积极性。管理者可以清楚的了解员工的工作情况，通过分析不足，帮助员工改进、提高业绩，促进企业的发展进步。假设现有某房产公司一个部门员工的某年销售业绩数据，请帮助部门经理对其进行统计分析，获取相关信息。学习目标和要求1、掌握将数据构建为RDD的方法。2、能使用RDD的各种转换和行动操作对数据集进行处理分析。3、会对处理完成的数据进行存储操作。数据集处理

在Spark中最核心最基础的概念是弹性分布式数据集（ResilientDistributedDatasets，RDD）。它是一种分布式的内存抽象，可以基于任何数据结构进行创建。

创建RDD的方法有两种，一种是基于内存（集合）创建RDD，另一种是从外部数据集创建。1、基于内存（集合）创建RDD。

从集合中创建RDD，主要提供了两个方法：使用SparkContext类中的parallelize和makeRDD两个方法。数据集处理（1）parallelize方法创建RDD

此方法有两个参数可以输入，第一个参数用来接收一个集合。第二个参数是可选的，用于指定创建的RDD的分区数。（2）makeRDD方法创建RDDmakeRDD方法和parallelize方法类似，但它可以指定每个分区的首选位置。数据集处理2、从外部存储数据构造RDD数据集处理

读取一个放在文件系统中的数据进行创建RDD。从本地文件创建，主要用于测试;使用HDFS文件获取外部数据集创建,实践操作中最常用。通过调用SparkContext的textFile方法读取数据集。调用命令如下：textFile("/my/directory")textFile("/my/directory/*.txt")textFile("/my/directory/*.gz")（1）加载HDFS上存储的CSV文件构造一个RDD数据集处理

从HDFS中的数据创建RDD，首先需要定义文件在HDFS上存储的路径，然后根据路径使用sc.textFile()方法进行创建。//定义HDFS上test.csv文件的存储路径valHDFSfile="/Chapter4/test.csv"

//使用textFile方法，加载文件并构造RDDvalHDFSrdd=sc.textFile(HDFSfile)

//查看返回值HDFSrdd.collect()（2）加载本地Linux中的文件构造一个RDD数据集处理

读取本地Linux中的文件构造RDD也是通过sc.textFile("path")的方法，在path路径前面加上“file:///”表示从本地文件系统读取数据。3、操作练习——构建员工业绩RDD数据集处理（1）上传数据集到HDFS文件系统中。（2）从HDFS文件系统中读取文件并创建RDD。数据操作分析1、RDD支持的两种操作。（1）转换操作：就是对RDD中的数据进行各种转换。方法名称作用map将RDD中的每一个数据元素通过func函数转换，返回新的RDD。flatmap首先将map方法应用于RDD的所有元素，然后将结果扁平化拆分，返回一个新的RDD。sortby通过指定条件对RDD中的元素进行排序。filter通过指定条件对RDD中的元素进行过滤。distinct对RDD中的所有元素去重，返回一个去重后的RDD。union将两个RDD进行合并，返回合并后的数据集。keys返回PairRDD中的“键”形成的新的RDD。values返回PairRDD中的“值”形成的新的RDD。reducebykey对“键”相同的值使用指定的函数进行聚合操作。groupbykey对“键”相同的值根据指定条件进行分组。sortbykey根据“键”对RDD内部的元素进行排序。join根据“键”对两个RDD进行连接。数据操作分析（2）行动操作：RDD的行动操作则是向驱动器程序返回结果或者把结果写入外部系统的操作，会触发实际的计算方法名称作用count返回RDD中元素的个数。take返回RDD中前n个元素值。first返回RDD中第一个元素值。collect返回RDD中所有元素的列表。top返回RDD中排名前n的元素值。数据操作分析2、统计部门人员人数。count()行动操作，返回的是RDD内元素的个数。举例：利用序列Seq(1,2,3,4,5,6)创建的rdd中有6个数字元素操作练习：统计部门人员人数数据操作分析3、分别统计上/下半年业绩排名，取前三位及最后三位利用map、flatmap、sortby、take，first、collect等方法对数据进行操作实现。（1）map转换操作map转换操作是最常用的转换算子，对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。举例：数据操作分析（2）flatmap转换操作

此方法首先将map函数应用于RDD的所有元素，然后将返回的结果平坦化。举例：对map转换操作中的str1进行flatmap转换操作。

对比map和flatMap操作的输出结果图，可以很容易的看出两者的区别，因此flatMap操作通常用来切分单词。数据操作分析（3）sortby转换操作此方法是对RDD进行排序，有3个参数可以输入。 sortby(参数一,参数二,参数三)参数一是要进行排序的对象值；参数二是排序方式，默认是正序排序，使用false参数就是倒序排序；参数三是分区个数。数据操作分析//加载一个RDD，命名为testvaltest=sc.parallelize(Array(("dog",3),("cat",1),("monkey",2),("pig",3),("bird",2)))//根据test中的第二位元素进行降序排序valtest_desc=test.sortBy(_._2,false)//查看结果test_desc.collect//根据test中的第二位元素进行升序排序valtest_asc=test.sortBy(_._2)//查看结果test_asc.collectsortby举例：数据操作分析（4）take，first、collect行动操作take行动操作返回RDD的前n个元素值。first行动操作返回的是RDD中的第一个元素值。collect行动操作是以数组的形式将RDD中所有的元素返回。数据操作分析（5）进行分析，实现任务。1）按照前面的问题分析思路，首先对上半年和下半年的业绩数据集firstrdd、secondrdd进行map转换，将每行数据分割为4列。2）利用sortBy方法对map_firstrdd和map_secondrdd数据集中的业绩列分别进行降序和升序排序，取出排名前三和后三的信息。对上半年业绩列进行降序和升序排序： valsort_firstrdd=map_firstrdd.sortBy(x=>x._4,false) sort_firstrdd.take(3) valsort_firstrdd=map_firstrdd.sortBy(x=>x._4) sort_firstrdd.take(3)同理对下半年业绩列进行降序和升序排序。数据操作分析4、统计上/下半年业绩超过5000万的人员。

要获取这个问题的数据，首先需要分别过滤出上下半年业绩超过5000万的人员。但是有些人员只有上半年或者下半年符合要求，有些人员是上下半年都符合要求，因此还需要对过滤出来的人员信息进行去重。可以利用filter、distinct、union等方法对数据进行操作。（1）filter转换操作此方法返回满足指定过滤条件的元素，不满足条件的元素被忽略。//读取一个数字列表创建RDDvalrdd=sc.parallelize(List(1,3,6,2,8,8,5,6,7))//过滤出大于5的数字rdd.filter(x=>x>5).collect数据操作分析（2）distinct转换操作

此方法是对RDD中的数据去重，把完全相同的元素去除。对前述filter转换操作中的rdd数据进行操作。由输出结果可见，原先列表中重复的数字6和8被去重了。数据操作分析（3）union转换操作

此方法可以将两个RDD进行合并，返回两个RDD的并集，并且不去重。但是要求两个RDD中每个元素中的值的个数及数据类型保持一致。//定义加载两个RDD，命名为test1和test2valtest1=sc.parallelize(Array(("dog",3),("cat",1),("monkey",2)))valtest2=sc.parallelize(Array(("monkey",2),("pig",3),("bird",2)))//执行union操作合并两个RDDtest1.union(test2).collect数据操作分析（4）进行分析，实现任务。1）利用filter方法过滤出业绩大于5000万的人员，最终返回人员编号信息。2）利用union方法，将上述过滤后的数据集filter_firstrdd和filter_secondrdd进行合并，并通过distinct方法去除重复的人员编号信息。数据操作分析5、统计此部门当年的房屋销售套数。

要获取这个问题的数据，只需取出数据表的第三列“房屋销售套数”数据，然后利用求和函数进行求和。可以利用RDD的相关描述性统计函数实现。（1）min()、max()函数min()方法返回RDD中的最小值，max()方法返回RDD中的最大值。数据操作分析（2）mean()、sum()函数mean()函数返回RDD中的平均值。sum()函数返回RDD中的总和。举例：求RDD数据集的平均值和总和。数据操作分析（3）variance()、stdev()函数variance()计算RDD中所有元素的总体方差。stdev()计算RDD的标准差。举例：求RDD数据集的总体方差和样本方差。数据操作分析（4）进行分析，实现任务。

将经过map操作处理好的上/下半年业绩数据集map_firstrdd和map_secondrdd进行合并，取出第三列数据，利用sum函数计算总数。数据操作分析6、查看全年总业绩最高人员。

此任务需要将每个人上、下半年的业绩进行相加，然后取最高者。这个过程需要对人员编号相同的业绩数据进行识别并相加。在RDD中可以通过键值对（Key-Value）的操作完成。Spark为包含键值对类型的RDD提供了一些专有的操作，这些RDD被称为Pair

RDD。数据操作分析（1）创建PairRDD

创建PairRDD的方法有很多，当需要将一个普通RDD转换为PairRDD时，可以使用map方法来实现。//定义一个RDDvalrdd=sc.parallelize(List("dog","cat","monkey","pig"))//创建键值对valpairrdd=rdd.map(word=>(word,1))（2）keys()、values()转换操作keys()、values()操作返回的类型是RDD。//取pairrdd中的键pairrdd.keys.collect//取pairrdd中的值pairrdd.values.collect数据操作分析（3）reducebykey()转换操作

此方法应用于键值对数据集操作，对Key相同的Value使用指定的函数进行聚合操作，返回一个键值对的数据集。reduceByKey((x,y)=>x+y)，对PairRDD中键相同的值进行相加操作。数据操作分析（4）groupbykey()转换操作

此方法会对相同键的值进行分组，形成二元元组，第一个字段为相同的键，第二个字段为具备相同键的值的集合。对前一个任务中的map_test数据集进行操作。 valgroup_test=map_test.groupByKey() group_test.collect数据操作分析（5）sortByKey()转换操作此方法返回一个根据“键”进行排序的RDD。举例：数据操作分析（6）进行分析，实现任务1）将上、下半年的的业绩合并到同一个RDD中。valtotal=map_firstrdd.union(map_secondrdd)2）将数据转换成（员工编号，业绩）键值对，利用reduceByKey方法对相同员工编号对应的业绩值进行相加。数据操作分析3）对业绩总和进行排序，获得最高业绩人员信息。//利用sortBy方法对指定列进行排序 yeji.sortBy(x=>x._2,false) yeji.sortBy(x=>x._2,false).take(1)数据操作分析7、存储以上统计分析信息

数据存储的方式有多种，比如将数据存储为JSON文件、CSV文件、文本文件等。文本文件的存储可以直接调用saveAsTextFile(path)进行存储。

对上、下半年的业绩以及全年总业绩数据进行join连接，再将结果数据以文本文件的形式存储到HDFS中。操作练习：某城市近年房产销售状况分析第二部分数据准备数据探索与分析情境导入

房子是一个价格昂贵的“商品”，但也是每个人生活必不可少的一件“商品”。房子的价格、所处的位置、装修情况等等因素都是购房者关心的主要问题。了解房产市场的变化规律以及各个因素对价格的影响，对于消费者来说是购房之前首先要做的功课。

现有一份来自某网站的某城市近年房产销售数据集house-price.csv，包含成交时间、价格、装修情况、区域等信息。学习目标和要求

能灵活综合的应用各种RDD操作和各个算子对数据进行分析，解决以下问题： 1、每年房产销售量趋势如何？ 2、此城市各区域房产销量如何？均价如何？ 3、在2018年1月1日的销量有多少？ 4、查询2018年1月1日到1月31日之间，满五年房产的销售比例。 5、哪种装修类型的房子销量最高？ 6、所售卖的房子中，电梯有无的比例，地铁有无情况？数据准备

此房产销售数据集是一个在系统外部的csv文件，因此需要将其上传至文件系统中，加载为RDD后再做处理分析。1、利用MobaXterm工具，将house-price.csv文件上传到本地Linux的/root/data/Chapter4文件夹内。2、启动Hadoop和Spark集群，将/root/data/Chapter4目录下的house-price.csv文件上传到HDFS文件系统的/Chapter4目录中。3、启动spark-shell，加载数据到RDD。数据探索与分析

在进行数据探索分析之前，还需要将前面加载到RDD中的数据进行转换，把每一条数据根据字段含义进行分割，以便后续数据处理分析。1、使用map方法转换RDD，以“,”将每一行数据分割。数据探索与分析2、对户人数的基本情况进行统计，获得记录条目的总数，以及在这些记录中包括多少个用户ID。//查看数据总条数m_houseprice.count//查看这些记录中的用户数m_houseprice.map(_(1)).distinct.count数据探索与分析3、了解房产销量趋势，统计每年房产销售量情况。

要获得每年的房产销售量情况，需要先从日期中提取年份数据；然后利用reduceByKey方法按年进行统计汇总；最后利用sortBy方法对统计汇总的数据进行排序，获得每年销量从高到低的排序情况。数据探索与分析4、分析此数据集中购买房屋所属的区域是哪些？各个区域的房产销售情况如何？各个区域的房屋均价是多少？（1）购买房屋所属的区域情况及数量

使用distinct方法去重就能获得购买房屋所属的各“区域”的情况。对于各个区域的销售量，可以使用reduceByKey方法，对相同键（区域）的值进行相加获得。（2）各个区域的房屋均价情况1）使用map取出“区域”、“区域均价”两列数据，将“区域均价”转换位Double类型数据，并对每个价格计数1，计数1的目的是为了之后统计Key的个数。2）对相同区域的“均价”值进行相加操作，再求平均。数据探索与分析5、查询2018年1月1日的日销量有多少。查询某一天或某一条件下的数据，可以使用filter转换操作。6、查询2018年1月1日到1月31日之间，满五年房产的销售数量。使用filter操作，可以圈定某一条件范围。数据探索与分析7、数据集中，房子的装修类型有精装、简装、毛坯、其他，四种类型，分析哪种装修类型的房子销量高。

此问题的分析思路与第4点统计区域房产销售情况一致，可以使用reduceByKey(_+_)方法对装修类型相同的房屋数量进行求和统计。8、了解在所售卖的房子中，电梯有无的比例，地铁有无情况，这些条件对房屋售卖量的影响程度。

对电梯有无数据列创建键值对，其中“键”指有无电梯，每个“值”赋予“1”。然后使用groupByKey()方法对相同“键”的键值对进行分类，分类后对“值”进行sum求和。有无地铁情况的数据处理思路相同。小结

本章介绍了Spark中弹性分布式数据集RDD的创建方法、利用RDD的各个算子对数据进行操作分析的方法。通过“房产大数据分析与探索”和“某城市近年房产销售状况分析”两个案例，灵活综合的应用各种RDD操作和各个算子对数据进行分析处理。THANKS!第5章电商大数据分析与探索女装电子商务评论情况分析第一部分数据准备数据转换数据分析数据输出数据清洗情境导入

网购已经成为人们生活中不可或缺的一件事情，如何经营好一家线上商店是很多商家关注的问题。对于线上商店，客户对商品的评论情况对商品的销量、商店的经营发展起到了重要的作用。现有某女装线上商店的一份围绕客户评论的女装电子商务数据集Clothing-Reviews.csv。现在商家希望能帮助他们对此份数据进行分析，获得一些建设性意见，帮助商店更好的发展。字段含义order_id订单编号clothing_id服装编号age年龄review_text 评论rating评级recommended_IND是否推荐positive_feedback_count积极反馈计数class_name服装分类学习目标和要求1、掌握从不同数据源创建DataFrames的方法。2、掌握操作DataFrame进行数据清洗的基本方法。3、掌握操作DataFrame进行数据转换的基本方法。4、掌握使用SparkSQL语句进行数据分析的方法。5、掌握持久存储数据的方法。数据准备

在实际业务中，数据的来源多种多样。SparkSQL在数据兼容方面，不仅可以直接处理RDD，也可以处理Parquet文件或者JSON文件，更可以处理外部数据库中的数据。

在Spark中，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格，是SparkSQL最核心的编程抽象。DataFrame与RDD的主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得SparkSQL得以洞察更多的结构信息，从而可以对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行针对性的优化，最终达到大幅提升运行效率的目标。数据准备

使用Spark库可以将不同数据来源的数据转换为DataFrame，并对数据结果进行展示。创建DataFrame的三种方法：1、使用toDF函数创建DataFrame（1）本地seq+toDF创建DataFrame。

（2）通过caseclass+toDF创建DataFrame数据准备2、使用createDataFrame函数创建DataFrame数据准备3、从外部数据创建DataFrame

（1）使用parquet文件创建valdf=sqlContext.read.parquet("/Chapter5/goods.parquet")

（2）使用json文件创建valdf=spark.read.json("/Chapter5/goods.json")

（3）使用CSV（Comma-SeparatedValues）文件创建1）读取CSV文件，使用类型推断创建DataFrame2）读取CSV文件，自定义schema创建DataFrame。数据准备4、读取“女装电子商务评论”数据集（1）启动Hadoop集群和Spark集群（2）将“Clothing-Reviews.csv”女装电子商务评论数据集上传到HDFS文件系统的/Chapter5/目录下。（3）读取CSV文件，使用自定义schema方式转换为DataFrame。数据清洗

数据清洗是对数据进行重新审查和校验的过程，目的在于删除重复信息、纠正存在的错误，并提供数据一致性。在Clothing-Reviews.csv数据集中，存在一些重复数据，也有部分缺失，需要我们对其进行数据清洗。1、drop(cols)

按照列名cols删除DataFrame中的列，返回新的DataFrame。此方法可以删除数据表中无用的或者不想要的数据列。删除“publisher”列。数据清洗2、dropDuplicates(subset=None)

此方法用于删除DataFrame中的重复行，subset用于指定删除重复行的时候考虑哪几列。删除重复的行：删除author列重复的行：数据清洗3、na.drop

此方法用于删除DataFrame中的null空数据，加入“any”和“all”参数可以指定删除条件，加入数字参数指定有多少个空值进行删除，加入字段名删除指定字段中的空值。对有任意一个为缺失值的行进行删除：对全部为缺失值的行进行删除：对有3个缺失值的数据行进行删除：对有两个缺失值以上的数据行进行删除：数据清洗4、na.fill()

此方法用于将DataFrame中所有缺失值填充为一个给定的字符串或数值。可以为所有缺失值指定字符串或数值，也可以为指定列指定不同字符串或数值。将表中所有的null空数据指定为“不明”：数据清洗对不同的列中缺失值填充不同的字符串：5、对“Clothing-Reviews.csv”女装电子商务评论数据集进行数据清洗，删除重复的数据，对有空缺数据的行进行删除。数据转换

数据转换是将数据从一种格式或结构转换为另一种格式或结构的过程。数据转换对于数据集成和数据管理等活动至关重要。数据转换可以包括一系列活动：可以转换数据类型，丰富数据或执行聚合，具体取决于项目的需要。1、withColumn(colName,col)

此方法用于为数据表添加新列，返回一个新的DataFrame增加colNAme列，如果原来本身就有colName的列，就进行替换。数据转换2、withColumnRename(existing,new)

此方法用于对已经存在的列名进行重命名。existing为原来的名字，