第5章 Spark SQL-结构化数据文件处理

SparkSQL

——结构化数据文件处理基于大数据技术对房价进行分析和预测，是科学精准调控，促进房地产市场平稳健康发展，实现“房住不炒”的重要手段。现有一份房屋销售数据文件house.csv，记录了某地区的房屋销售情况，包含销售价格、销售日期、房屋评分等共14个数据字段，字段说明如下。（1英尺=0.3048米）任务背景字段名称说明

字段名称说明selling_price销售价格（单位：美元）built_area建筑面积（单位：平方英尺）bedrooms_num卧室数basement_area地下室面积（单位：平方英尺）bathroom_num浴室数year_bulit修建年份housing_area房屋面积（单位：平方英尺）year_repair修复年份parking_area停车区面积（单位：平方英尺）latitude纬度floor_num楼层数longitude经度sales_data销售日期housing_rating房屋评分在进行房价数据分析之前，由于无法直接判断出各个数据字段之间的关系，因此需要先对数据进行基础的探索，探索各个数据字段间的关系并加以分析。本章将使用SparkSQL即席查询框架解决房价数据探索分析的问题。首先介绍SparkSQL框架及其编程数据模型DataFrame的基本概念，并对SparkSQL相关环境进行配置。其次介绍DataFrame的查询、输出操作API的用法。最后结合房价数据探索分析实例，帮助读者掌握DataFrame的基础操作。任务背景1掌握DataFrame基础操作目录认识SparkSQL2探索分析房屋售价数据3使用SparkSQL探索分析房价数据前，需要先了解SparkSQL是什么，有什么作用。本节的任务如下。了解SparkSQL框架、SparkSQL的编程模型DataFrame和SparkSQL的运行过程。对SparkSQL相关环境进行配置。任务描述SparkSQL是一个用来处理结构化数据的Spark框架可被视为一个分布式的SQL查询引擎，并且提供了一个抽象的可编程数据模型DataFrame。SparkSQL可以直接处理RDD，也可以处理Parquet文件或者JSON文件，甚至可以处理外部数据库中的数据以及Hive中存在的表数据。了解SparkSQL基本概念SparkSQL架构了解SparkSQL基本概念将hive-site.xml复制到$SPARK_HOME/conf目录下。在$SPARK_HOME/conf/spark-env.sh文件中配置MySQL驱动。启动MySQL服务。启动Hive的metastore服务。修改日志级别。启动Spark集群。启动spark-sql。配置SparkSQL启动spark-shell即可使用SparkSQL的Shell交互接口。如果在spark-shell中执行SQL语句，需要使用SparkSession对象来调用sql()方法。HiveContext现在不仅支持HiveQL语法解析器，同时也支持SQL语法解析器。在spark-shell启动的过程中会初始化SparkSession对象为spark，此时初始化的spark对象既支持SQL语法解析器，也支持HiveQL语法解析器。也就是使用这个spark可以执行SQL语句和HQL语句。如果需要支持Hive，还需要启用enableHiveSupport()方法，并且Hive的配置文件hive-site.xml已经存在于工程中。了解SparkSQL与Shell交互1掌握DataFrame基础操作目录认识SparkSQL2探索分析房屋售价数据3SparkSQL提供了一个抽象的编程数据模型DataFrame，DataFrame是由SchemaRDD发展而来的，从Spark1.3.0开始，SchemaRDD更名为DataFrame。SchemaRDD直接继承自RDD，而DataFrame则自身实现RDD的绝大多数功能。可以将SparkSQL的DataFrame理解为一个分布式的Row对象的数据集合，该数据集合提供了由列组成的详细模式信息。本节的任务是学习DataFrame对象的创建方法及基础的操作。任务描述通过结构化数据文件创建DataFrame通过外部数据库创建DataFrame通过RDD创建DataFrame通过Hive中的表创建DataFrame创建DataFrame对象一般情况下，结构化数据文件存储在HDFS中，较为常见的结构化数据文件是Parquet文件或JSON文件。SparkSQL可以通过load()方法将HDFS上的结构化文件数据转换为DataFrame，load()方法默认导入的文件格式是Parquet。若加载JSON格式的文件数据，将其转换为DataFrame，则还需要使用format()方法。也可以直接使用json()方法将JSON文件数据转换为DataFrame。创建DataFrame对象1.通过结构化数据文件创建DataFrameSparkSQL可以从外部数据库（比如MySQL、Oracle等数据库）中创建DataFrame。使用这种方式创建DataFrame需要通过JDBC连接或ODBC连接的方式访问数据库。读取MySQL数据库的表数据创建DataFrame对象2.通过外部数据库创建DataFrame利用反射机制推断RDD模式，使用这种方式首先需要定义一个caseclass，因为只有caseclass才能被Spark隐式地转换为DataFrame。创建DataFrame对象3.通过RDD创建DataFrame采用编程指定Schema的方式将RDD转换成DataFrame。加载数据创建RDD。使用StructType创建一个和步骤（1）的RDD中的数据结构相匹配的Schema。通过createDataFrame()方法将Schema应用到RDD上，将RDD数据转换成DataFrame。Spark

DataFrame基础操作使用SparkSession对象并调用sql()方法查询Hive中的表数据并将其转换成DataFrame。创建DataFrame对象4.通过Hive中的表创建DataFrame查看及获取数据的常用函数或方法将movies.dat电影数据上传至HDFS中，加载数据为RDD并将其转换为DataFrame.查看DataFrame数据方法描述printSchema打印数据模式show查看数据first/head/take/takeAsList获取若干行数据collect/collectAsList获取所有数据printSchema函数查看数据模式，打印出列的名称和类型查看DataFrame数据1.printSchema：输出数据模式方法解释show()显示前20条记录show(numRows:Int)显示numRows条show(truncate:Boolean)是否最多只显示20个字符，默认为trueshow(numRows:Int,truncate:Boolean)显示numRows条记录并设置过长字符串的显示格式查看DataFrame数据2.show()：查看数据show()方法与show(true)方法一样，只显示前20条记录并且最多只显示20个字符若是要显示所有字符，需要使用show(false)方法查看DataFrame数据show(numRows:Int)查看前n行记录查看DataFrame数据方法解释first获取第一行记录head(n:Int)获取前n行记录take(n:Int)获取前n行记录takeAsList(n:Int)获取前n行数据，并以列表的形式展现查看DataFrame数据3．first()/head()/take()/takeAsList()：获取若干条记录collect方法可以将DataFrame中的所有数据都获取到，并返回一个数组。collectAsList方法可以获取所有数据，返回一个列表。查看DataFrame数据4．collect()/collectAsList()：获取所有数据将DataFrame注册成为临时表，然后通过SQL语句进行查询掌握DataFrame查询操作直接在DataFrame对象上进行查询，DataFrame提供了很多查询的方法掌握DataFrame查询操作方法描述where条件查询select/selectExpr/col/apply查询指定字段的数据信息limit查询前n行记录orderby排序查询groupby分组查询join连接查询（1）where()方法DataFrame可以使用where(conditionExpr:String)根据指定条件进行查询参数中可以使用and或or该方法的返回结果仍然为DataFrame类型掌握DataFrame查询操作1．where()/filter()方法（2）filter()方法DataFrame还可使用filter筛选符合条件的数据掌握DataFrame查询操作（1）select()方法：获取指定字段值select方法根据传入的string类型字段名，获取指定字段的值，以DataFrame类型返回掌握DataFrame查询操作2．select()/selectExpr()/col()/apply()方法（2）selectExpr()方法：对指定字段进行特殊处理selectExpr：对指定字段进行特殊处理，可以对指定字段调用UDF函数或者指定别名。selectExpr传入String类型的参数，返回DataFrame对象。掌握DataFrame查询操作（3）col()/apply()方法col或者apply也可以获取DataFrame指定字段col或者apply只能获取一个字段，并且返回对象为Column类型掌握DataFrame查询操作limit方法获取指定DataFrame的前n行记录，得到一个新的DataFrame对象不同于take与head，limit方法不是Action操作。掌握DataFrame查询操作3．limit()方法orderBy方法是根据指定字段排序，默认为升序排序若是要求降序排序，可以使用desc(“字段名称”)或者$”字段名”.desc或者在指定字段前面加“-”来表示降序排序掌握DataFrame查询操作4．orderBy()/sort()方法sort方法与orderBy方法一样，也是根据指定字段排序，用法也与orderBy一样掌握DataFrame查询操作groupBy方法是根据字段进行分组操作groupBy方法有两种调用方式，可以传入String类型的字段名，也可传入Column类型的对象。掌握DataFrame查询操作5．groupBy()方法groupBy方法返回的是GroupedData对象，GroupedData的操作方法如下表。掌握DataFrame查询操作方法描述max(colNames:String)获取分组中指定字段或者所有的数值类型字段的最大值min(colNames:String)获取分组中指定字段或者所有的数字类型字段的最小值mean(colNames:String)获取分组中指定字段或者所有的数字类型字段的平均值sum(colNames:String)获取分组中指定字段或者所有的数字类型字段和值count()获取分组中的元素个数DataFrame提供了三种join方法用于连接两个表掌握DataFrame查询操作6．join()方法方法描述join(right:DataFrame)两个表做笛卡尔积join(right:DataFrame,joinExprs:Column)根据两表中相同的某个字段进行连接join(right:DataFrame,joinExprs:Column,joinType:String)根据两表中相同的某个字段进行连接并指定连接类型join(right:DataFrame)掌握DataFrame查询操作join(right:DataFrame,joinExprs:Column)掌握DataFrame查询操作join(right:DataFrame,joinExprs:Column,joinType:String)连接类型joinType只能是inner、outer、left_outer、right_outer、semijoin中的一种掌握DataFrame查询操作save()方法可以将DataFrame数据保存成文件。saveAsTable()方法可以将DataFrame数据保存成持久化的表，并在Hive的元数据库中创建一个指针指向该表的位置，持久化的表会一直保留，即使Spark程序重启也没有影响，只要连接至同一个元数据服务即可读取表数据。读取持久化表时，只需要用表名作为参数，调用spark.table()方法方法即可得到对应DataFrame。默认情况下，saveAsTable()方法会创建一个内部表，表数据的位置是由元数据服务控制的。如果删除表，那么表数据也会同步删除。掌握DataFrame输出操作将DataFrame数据保存为文件，实现步骤如下。首先创建一个映射对象，用于存储save()方法需要用到的数据，这里将指定文件的头信息及文件的保存路径。从user数据中选择出userId、gender和age这3列字段的数据。调用save()方法将步骤（2）中的DataFrame数据保存至copyOfUser.json文件夹中。在HDFS的/user/root/sparkSql目录下查看保存结果。掌握DataFrame输出操作使用saveAsTable()方法将DataFrame对象copyOfUser保存为copyUser表。掌握DataFrame输出操作1掌握DataFrame基础操作目录认识SparkSQL2探索分析房屋售价数据3从购房者角度出发，影响用户购房的主要因素有房屋价格、房屋属性。从地产公司角度出发，除了房屋本身的居住属性外，地产公司也会考虑什么样的房屋属性在市场上更具有价值。本节的任务如下。使用SparkSQL实现房价数据的探索分析，从数据中的房屋售价、房屋评分和房屋销售日期等维度进行探索分析，获得房价波动的内在原因。任务描述在Hive中创建数据库house。在house数据库下创建king_county_house表。将数据导入表king_county_house中。通过spark.table()方法读取Hive中king_county_house表的数据。获取数据定义一个函数null_count，函数中使用na()方法可以统计数据每个字段的缺失值。字段值分布探索的代码将封装在max_min_mean_std函数中，使用selectExpr()方法查询指定字段的信息，结合max()、min()、mean()及stddev()方法，统计指定字段中的最大值、最小值、平均值及标准差。数据中的sale_data、year_built和year_rapair字段的含义均表示时间，进行字段值分布的探索并无实际意义，因此将使用for循环遍历数据所有字段，并使用ifelse语句进行筛选。探索字段值分布字段值分布和缺失值数量情况探索字段值分布字段名称最大值最小值平均值标准差值缺失值数量sale_data\

\\\1selling_price688500075000542874.928372925.7651bedrooms_num1003.3680.8931bathroom_num7.7502.1170.7741housing_area98903902082.488922.8791parking_area165135957215352.73445776.2291floor_num3.511.5020.5441housing_rating1337.6651.1741built_area88603901791.475829.4491basement_area48200291.014446.6411year_built\\\\1year_repair\\\\1latitude47.777647.159347.5600.1381longitude-121.315-122.519-122.2150.1391由于原始数据表中销售时间字段“sale_data”为string类型数据，所以需要先将该字段转换为日期格式，并通过withColumn()方法在原数据上新增一列转换后的字段date。使用quarter()方法将时间划分为4个季度，并使用withColumn()方法将划分结果存放在新的字段quarter中。通过select()方法查询出销售日期（date）和季度（quarter）字段。统计各季度房屋销量和销售额1．各季度房屋销量统计通过groupBy()方法根据划分的季度进行分组，并统计每组的销售数量。统计各季度房屋销量和销售额使用groupBy()方法根据季度进行分组，统计每个季度的房屋销售总额。统计各季度房屋销量和销售额2．各季度房屋销售额统计使用groupBy()方法根据housing_rating字段进行分组，并对每个分组下的房屋数量进行统计，即可得出不同评分的房屋分布情况。探索分析房屋评分通过groupBy()方法根据housing