《大数据分析技术应用》课件- 4.3 4.4Spark SQL处理结构化学生数据

项目四SparkSQL处理结构化学生数据项目概述SparkSQL是Spark生态中用于处理结构化数据的一个模块，开发人员可以轻松的借助API、SQL语句完成数据分析工作。SparkSQL支持多种数据源，实际开发中经常要读写MySQL数据库及Hive数据仓库，本项目从DataFrame的创建入手，介绍不同的DataFrame的创建方法及DataFrame的各种操作。针对某校的学生信息文件（含学院、姓名、性别、年龄等），使用SparkSQL进行分析，获取分析结果；项目效果使用SparkSQL可以完成学生信息的分析，提取有价值的信息。例如分学院、性别统计年龄最大、最小值，如下所示：除此之外，还可以实现与SparkSQL与MySQL、Hive的连接。利用SparkSQL技术完成数据分析后，结果可再次写入到MySQL、Hive中，如下所示：目录任务1初识结构化数据处理工具SparkSQL认识DataFrame、DataSet数据类型由学生信息创建DataFrameSparkSQL分析学生信息（1）任务2任务3任务4SparkSQL分析学生信息（2）任务5任务6SparkSQL分析学生信息（3）目录任务7SQL语法风格处理学生信息通过JDBC连接MySQL数据库任务9RDD、DataFrame与Dataset的相互转换其他类型数据创建DataFrame任务11SparkSQL读写Hive数据任务8任务10思维导图初识结构化数据处理工具SparkSQL任务1SparkSQL是Spark体系中处理结构化数据的有力工具。初步认识SparkSQL，了解其演化历程、特点，并初步体验其使用过程。早期Hadoop生态体系中，数据处理主要使用MapReduce组件。缺陷：MapReduce学习成本较高、需要较多的Java编程等知识。后续产生了Hive分布式数据仓库，它允许用户使用类似于SQL的语法（HQL）处理结构化数据，极大降低了使用门槛；Hive与Hadoop高度集成，将HQL语言自动换成MapReduce操作，可使用Yarn完成资源调度，最终完成结构化数据的处理任务；因其便捷性，Hive逐渐流行起来，成为搭建分布式数据仓库的主流方案之一。缺陷：其底层基于MapReduce（HQL最终转换为MapReduce操作），而MapReduce的shuffle需要大量的磁盘I/O，因此导致Hive的性能低下，复杂的操作可能运行数个小时，甚至数十个小时。为此，伯克利实验室开发了基于Hive的结构化数据处理组件Shark（SparkSQL的前身）。Shark是Spark上的数据仓库，最初设计成与Hive兼容；Shark在HiveQL方面重用了Hive中的HiveQL解析、逻辑执行计划翻译、执行计划优化等逻辑，但在执行层面将MapReduce作业替换成了Spark作业（把HiveQL翻译成Spark上的RDD操作）。因此与Hive相比，因其使用Spark的基于内存的计算模型，性能得到了极大提升。缺陷：一是执行计划优化完全依赖于Hive，对于其性能进一步提升造成了约束；二是Spark是线程级并行，而MapReduce是进程级并行，Spark在兼容Hive的实现上存在线程安全问题。此外，Shark继承了大量的Hive代码，因此后续优化、维护较为麻烦，特别是基于MapReduce设计的部分，成为整个项目的瓶颈。因此，2014年Shark项目中止，并转向SparkSQL的开发。1.1SparkSQL的产生早期，SparkSQL引入了SchemaRDD（即带有Schema模式信息的RDD），使用户可以在SparkSQL中执行SQL语句，数据既可以来自RDD，也可来自Hive、HDFS、Cassandra等外部数据源，还可以是JSON、Parquest、CSV等格式的数据。开发语言方面，SparkSQL支持Scala、Java、Python等语言，支持SQL-92规范。从Spark1.2升级到Spark1.3以后，SparkSQL中的SchemaRDD改为DataFrame，DataFrame相对于SchemaRDD有了较大改变,同时提供了更多便捷的API。SparkSQL可以使用JDBC、ODBC等标准数据库连接器，友好的支持各种SQL查询；这样其他第三方工具，如部分商业智能工具（PowerBI、Tableau等）可以接入Spark，借助Spark的强大计算能力完成大规模数据的处理。1.2

SparkSQL的特点在早期的Spark1.X版本中，Spark结构化数据处理的入口为SQLContext和HiveContext；其中，SQLContext仅支持SQL语法解析器，而HiveContext继承了SQLContext，HiveContext即支持SQL语法解析器又支持HiveQL语法解析器。Spark2.0版本之前，使用Spark必须先创建SparkContext，SparkContext是程序入口和程序执行的“调度者”；下面代码演示Spark1.X版本下，依次创建SparkConf、SparkContext、SQLContext实例：//创建SparkConf实例valsparkConf=newSparkConf().setAppName("SparkSessionZipsExample").setMaster("local")//创建SparkContext实例valsc=newSparkContext(sparkConf)//创建SQLContext实例valsqlContext=neworg.apache.spark.sql.SQLContext(sc)1.3体验SparkSQL不同版本的操作差异在Spark2.0以后版本中，只要创建一个SparkSession实例就够了，SparkConf、SparkContext和SQLContext都已经被封装在SparkSession当中（SparkSession是程序入口、SparkSQL的上下文环境）；HiveContext类已经被移除，其功能迁移到SparkSession类中。下面代码演示Spark2.X版下，通过SparkSession实例访问SparkContext实例和SQLContext实例：//创建SparkSession实例valspark=SparkSession.builder().appName("SparkSQLExample").getOrCreate()//由SparkSession创建SparkContext、SQLContext实例valsc=spark.sparkContextvalsqlContext=spark.sqlcontext注意，在SparkShell环境下，已经建好了一个SparkSession对象spark，可以直接使用；但在独立应用程序中，则需要手工建立。1.3体验SparkSQL不同版本的操作差异认识DataFrame、DataSet数据类型任务2Spark可将RDD封装为DataFrame、DataSet，支持SQL类的操作。介绍DataFrame、Dataset数据类型，认识RDD、DataFrame、Dataset三者的区别。SparkSQL中DataFrame是其核心数据抽象，其前身是SchemaRDD（Spark1.3中首次引入DataFrame的概念）。DataFrame借助Schema（模式信息）将数据组织到一个二维表格中（类似于关系型数据库的表），每一列数据都存在列名。基于RDD进行数据分析时，因为RDD的不可修改性，为了得到最终结果，需要进行若干次转换、生成若干RDD；用DataFrame进行分析时，一条SQL语句也可能包含多次转换，但转换操作在其内部发生，并不会频繁产生新的RDD，从而获得更高的计算性能。2.1认识DataFrameDataFrame与RDD的主要区别在于，前者带有schema信息，即DataFrame所表示的二维表数据集的每一列都带有名称和数据类型。这使得SparkSQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行效率的目的。反观RDD，由于无法得知其元素的具体内部结构，SparkCore只能在stage层面进行简单、通用的流水线优化。如图4-6所示，假设有若干人员Person数据，包含name、age两项信息。若用RDD进行处理，则需要定义个Person类，将用户数据封装到Person类型对象中，RDD的每一个元素都是Person类型，而Spark并不清楚其内部结构。如果要把数据存放到DataFrame中，则每一个元素都会被封装为Row类型，DataFrame提供了详细的结构信息，SparkSQL可以清楚地知道该数据集中包含多少列、每列的名称和数据类型，如图4-6所示。2.1认识DataFrameDataset是DataframeAPI的一个扩展，是Spark1.6版本加入的新的数据抽象，也是Spark2.0之后管理结构化数据的主要数据抽象。Dataframe是Dataset的特列，DataFrame=Dataset[Row]，Row是一个类型，跟Car、Person这些的用户定义类型一样，所有的表结构信息都用Row来表示。DataSet是强类型的，可以有Dataset[Car]、Dataset[Person]等，而DataFrame的每一行数据则只能为Row类型。除此之外，DataFrame只是知道字段，但不知道字段的类型，所以在执行这些操作的时候是没办法在编译的时候检查是否类型失败的；比如对一个String进行减法操作，编译时不会报错，在执行的时候才报错；而DataSet不仅仅知道字段，而且知道字段类型，所以有更严格的错误检查。2.2认识Dataset如下所示，对于人员Person数据，可以自行创建一个Person样例类（caseclassPerson(name:String,age:Int)），而后将每个人的数据封装为一个Person对象，最后放入Dataset中；即Dataset的每一行都是一个Person类实例。这种形式更加符合面向对象编程的思路，更加贴合业务场景，便于处理业务中的数据关系。2.2认识DatasetRDD中，明确知道每个元素的具体类型，但不知道元素的具体属性，需要加以判别。DataFrame中，每一个元素（每一行）均为Row类型，可以知道每个元素有多少列，每列的名称是什么，但不知道每列的数据类型。DataSet集成了RDD和DataFrame的优点，可以明确知道每一个元素（每一行）的具体类型（预先定义的类），进而知道每列数据的名称，也知道其数据类型；在Spark2.X版，DataSet、DataFrame的API已做了统一。2.3RDD、DataFrame、Dataset三者的区别由学生信息创建DataFrame任务3要使用SparkSQL进行数据分析，创建DataFrame；由JSON数据文件构建DataFrame以供后续处理DataFrame的相关数据输出、打印等方法。结构化文件（JSON，CSV，Parquest等）、数据库（mysql、Oracle等）、分布式数据库等均可以生成DataFrame，从而进行数据处理。下面演示由JSON文件创建DataFrame，其他生成方式，后续陆续讲解。首先在Linux终端中，使用以下命令将people.json文件上传到hadoop文件系统：然后使用spark.read.json(“文件路径”)或spark.read.format("json").load(“文件路径”)形式创建DataFrame，如下所示（代码中spark为SparkShell自动生成的SparkSession对象）：3.1由学生信息JSON文件创建DataFrame./hdfsdfs-put/home/hadoop/people.json/user/hadoopSparkDataFrame类派生于RDD类，因此与RDD类似，DataFrame的操作也分为转换操作和行动操作，同时DataFrame也具有惰性操作特点（只有提交行动操作时才真正执行计算）。SparkRDD经常用take、collect等行动操作查看数据，DataFrame同样提供了若干类似方法，常用的方法如下所示。3.2printSchema打印DataFrame的数据模式使用printSchema打印DataFrame的数据模式，可以看到df的列名称、数据类型以及是否可以为空。3.2printSchema打印DataFrame的数据模式show相关方法有多个，常用的如下所示3.3show方法显示DataFrame、中的数据（1）show方法默认显示20行3.3show方法显示DataFrame中的数据（2）show（numRows:Int）显示前numRows行3.3show方法显示DataFrame中的数据（3）show（5，false），显示5行、显示全部字符3.3show方法显示DataFrame中的数据（1）first获取DataFrame第一行数据，返回值类型为org.apache.spark.sql.Row（2）head获取DataFrame首元素，返回值类型为org.apache.spark.sql.Row3.4获取DataFrame若干行记录（3）take（numRows:Int），获取numRows行数据，返回值类型为Array[org.apache.spark.sql.Row]。（4）takeAsList（numRows:Int），获取numRows行数据，返回值类型为java.util.List[org.apache.spark.sql.Row]。3.4获取DataFrame若干行记录（1）collect获取DataFrame的所有记录，返回值类型为Array[org.apache.spark.sql.Row]3.5获取DataFrame所有记录（2）collectAsList获取DataFrame的所有记录，返回值类型为java.util.List[org.apache.spark.sql.Row]。3.5获取DataFrame所有记录SparkSQL

分析学生信息（1）

任务4where、filter、select等操作而进一步分析学生信息。DataFrame基础操作方法描述where条件查询select/selectExpr/col/apply查询指定字段的数据信息limit查询前n行记录orderby排序查询groupby分组查询join连接查询直接在DataFrame对象上进行查询，DataFrame提供了很多查询的方法where方法主要用于筛选出符合条件的行，它有两种参数形式。（1）参数为条件字符串（conditionExpr:String），方法定义如下所示，用于筛选出符合conditionExpr条件的数据，其返回值类型为org.apache.spark.sql.Dataset。例如，要筛选出年龄大于21岁的学生信息，实现代码如下所示。4.1where方法在条件语句conditionExpr中，可以使用and、or等连接词；