大数据分析及应用项目教程(Spark SQL)（微课版） 课件 许慧 第5、6章 电商大数据分析与探索、Zepplin数据可视化

第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为原来的名字，new为新的列名，若名称不存在则这个操作不做任何事情。3、定义UDF，将数据表中的两列合并成一列。UDF是指用户定义函数，用于扩展系统的内置功能。我们可以在SparkSQL中自定义实际需要的UDF来处理数据。//创建一个udf来连接两个列值valconnect=udf((name:String,author:String)=>{name+"-"+author})

//应用该udf来创建合并的列，并添加为新的列valconnectdf=books.withColumn("bookname-author",connect($"name",$"author"))数据转换4、定义UDF，对表格中的列添加字符或者数字。

定义一个udf，为price列的值增加10，此处需要利用toDouble将原来字符串类型的price值转换为double类型数值：valaddconst=udf((price:String)=>{if(price==null)0elseprice.toDouble+10})

当对表中已经存在的列的值追加新的列时，如果新列与旧列同名，则会覆盖旧列：数据转换5、na.replace

此方法用于对现有的字符或者数字删除或替换字符或者数字。如果在replace中列名参数是“*”，那么将会对所有的列进行相应的替换。6、对“Clothing-Reviews.csv”女装电子商务评论数据集进行数据转换，将recommended_IND（是否推荐）列中的1替换为推荐，0替换为不推荐。数据分析

使用SQL语句对数据进行探索分析是大多数软件开发和维护人员比较习惯的方式。SparkSQL支持直接应用标准SQL语句进行查询。在进行SQL操作之前，需要新建注册临时表，然后才能进行SparkSQL查询。1、注册临时表的方法：（1）createGlobalTempView(name)

此方法用于为DataFrame创建一个全局的临时表。其生命周期和启动的应用的周期一致，既启动的spark应用存在则这个临时的表就一直能访问，直到退出应用为止。name为新视图的名称，名称不能与已存在的一样，否则会报错。数据分析（2）createOrReplaceGlobalTempView(name)

此方法用于创建或替换视图。当视图名称name已经存在，则进行替换，若其不存在，则进行创建。（3）createTempView(name)

此方法用于创建临时视图。name为新视图的名称，名称不能与已存在的一样，否则会报错。（4）createOrReplaceTempView(name)此方法用于为DataFrame创建本地的临时视图，其生命周期只限于当前的SparkSession。当视图名称name已经存在，则进行替换，若其不存在，则进行创建。临时表创建完成后，就可以执行SQL语句，对表进行查询。数据分析2、案例分析：以本章数据集中的books.csv为例，进行操作理解。（1）自定义schema，读取books.csv文件，删除有空缺数据的行及重复数据。（2）注册临时表：book2.createOrReplaceTempView("btemp")数据分析（3）进行简单的SparkSQL查询在临时表中执行SQL，查询book_price价格小于30的书。数据分析3、对Clothing-Reviews.csv数据进行SQL探索分析。（1）查看40岁以下，不同年龄段的客户人数。

将统计的年龄数量count(age)重新命名为total_nu进行输出。并对统计的结果根据年龄进行升序排序。数据分析（2）统计recommended_IND为“不推荐”的订单的rating评分分布。选择rating列进行count(1)计数，并限定条件为recommended_IND='不推荐'，最后根据rating进行group分类。（3）为进一步了解用户“不推荐”的原因，可以查看用户具体评论。查看评分小于等于3且为“不推荐”的订单的具体评论。选择clothing_id，rating，review_text三列进行显示，并限定条件为rating<=3并且recommended_IND='不推荐'。数据分析（4）查询了解，不推荐的这些订单，分别属于哪些服装分类，各种服装类的不推荐数量情况是如何的。

选择服装分类class_name进行count()统计并命名为classnum，限定条件设置为recommended_IND='不推荐'，最后根据class_name分类。（5）统计分析不同评分等级的服装编号clothing_id情况。获得不同服装编号clothing_id的五星好评数量，并且降序排序。

选择clothing_id进行count()统计，并重命名为five_stars_num输出。限定条件为五星好评rating=5，以clothing_id进行分类，最后根据五星好评数量five_stars_num进行降序排序。数据分析（6）使用SparkSQL的内置函数进行相关性等分析。

在数据分析中，相关性分析可以用于分析两个变量之间的相关方向和相关强度，从而实现对其中一个数据进行改变来影响另一个数据的变化。可以利用协方差函数cov()判断两个变量之间的相关方向。可以利用相关性函数corr()计算两个变量之间的相关性。//协方差给出了这两个变量之间的相关方向valcov=reviews3.stat.cov("rating","positive_feedback_count")//corr()给出了这两个变量之间关系强度的大小valcorr=reviews3.stat.corr("rating","positive_feedback_count")1）分析“rating”评分字段和“positive_feedback_count”积极反馈数量字段之间的相关性。数据分析2）freqItems()提取数据列中最常出现的项。

在数据分析中，根据业务需求，可能需要知晓数据列中最常出现的元素，从而对决策提供依据。可以利用freqItems()函数提取数据列中最常出现的项，freqItems()函数中第一个参数是指定的分析列，第二个参数是指定的支持级别。

选择服装分类class_name作为分析列，在DataFramereviews3中，找到服装分类出现概率大于40%的类别，即支持级别为0.4。数据分析3）approxQuantile()计算DataFrame中数字列的近似分位数。

计算分位数的意义是表示在这个样本集中，从小至大排列之后，小于某值的样本子集占总样本集的比例。

对age列计算近似分位数，指定的分位数概率为0.25、0.5和0.75(注意：0是最小值，1是最大值，0.5是中值median)。4）crosstab()探索年龄age和评分rating等级间是否有相关性。crosstab()可以对两个变量中的每个数据进行相关性计算，从而观察这两个变量之间的相互关系程度与趋势。数据输出

数据分析后的结果最终需要保存输出，以供决策者进行查看引用。DataFrame提供了很多种输出操作方法，可以存储到数据库，也可以保存成文件。保存到文件里有两种方法，以将数据写入到HDFS中为例进行详细解释。方法一：利用save()函数保存数据。//调用save()函数，保存DataFrame数据到/Chapter5/reviews目录中。selectData.select("*").write.format("csv").save("/Chapter5/reviews")write.format()支持输出的格式有JSON、parquet、JDBC、orc、csv、text等文件格式，save()定义保存的位置。数据输出方法二：通过df.rdd.saveAsTextFile("file:///")转化成rdd再保存。此方法可以将数据存储为txt格式。将数据保存输出到HDFS文件系统的/Chapter5目录下，并命名为reviews_two.txt在线销售订单数据分析第二部分数据查询操作数据分析探索情境导入

随着电商直播与购物的盛行，大型零售商正在积极寻找增加利润的方法。销售分析是大型零售商通过了解客户的购买行为和模式来增加销售额的关键技术之一。查找客户购买行为数据集，探索数据集上下文中组合在一起的项目之间的关系。现有两份在线销售数据集orders.csv和order-details.csv，是关于不同地区、多种商品和电子产品等的销售情况。现在希望通过数据分析了解以下情况：1、此数据集中每个客户的下单数量是多少?2、每个城市的订单有多少?3、每个客户的消费总额是多少?4、客户的平均订单消费额是多少？学习目标和要求1、灵活使用DataFrame数据操作的各种方法进行数据查询。2、会使用聚合、连接等操作对DataFrame数据进行转换。3、能使用聚合函数对DataFrame数据进行分析。数据查询操作1、查看数据（1）show

该方法主要用于测试，查看输入的数据、获得的结果是否正确。该方法能够以表格的形式展示DataFrame中的数据。1）show()没有参数的show方法默认最多能够显示20条数据。2）show(numRows:Int)此方法可以指定要显示的行数。数据查询操作（2）collect()collect方法类似与SparkRDD的collect方法。与show方法不同的是，collect方法可以获取DataFrame中的所有数据，并返回一个Array对象。数据查询操作2、过滤查询1）where

在SQL语法中，可以通过where条件语句对数据进行过滤。在SparkSQL中，同样可以像SQL一样使用where(conditionExpr:String)条件语句来对数据进行过滤。可以指定条件参数conditionExpr来进行查询，参数中可以使用and或or。数据查询操作2）filterfilter方法和where方法类似，可以筛选出符合条件的数据。数据查询操作3、指定列查询1）select

在SparkSQL中可以通过select方法来查询指定字段，类似于SQL中的select方法。查询emp中，字段名称为name和salary的值。数据查询操作2）selectExpr

这个方法是select方法的一个变体，可以接受一个SQL表达式，或对某些字段做一些特殊处理。数据查询操作4、排序操作1）orderBy

此方法能够使数据按照指定的字段排序，默认升序排序。如果需要降序排序，可以使用desc(“字段名称”)或者$“字段名”.desc进行处理。对emp中的salary薪水列进行降序排序：数据查询操作2）sortsort方法与orderBy方法一样，也是对指定字段进行排序。对emp中的salary薪水列进行升序排序。数据查询操作5、分组聚合1）groupBy分组

此方法可以根据字段进行分组操作。调用方式可以传入String类型的字段名，也可以是Column类型的对象。

分别以传入String类型的字段名和Column类型的对象，对emp中的position职位进行分组。groupBy方法返回的是RelationalGroupedDataset对象，该对象提供了max()、min()、mean()、sum()、count()等相关的聚合方法，可以对分组后的数据进行指定的聚合操作。数据查询操作2）agg聚合

该方法一般与groupBy方法配合使用。agg(expers:column*)可以传入一个或多个Coumn对象，返回DataFrame。

以position职位进行分组，计算各职位薪水的平均值，并命名为average_wages。数据查询操作6、聚合函数

聚合是大数据分析领域中最常用的特性之一。SparkSQL提供了很多常用的聚合函数，比如sum、count、avg等等。（1）count函数count可以执行以下2种操作：对指定列进行计数；使用count(*)或count(1)对所有列进行计数。（2）countDistinct函数：此函数可以计算每个组的唯一项。数据查询操作（3）approx_count_distinct函数

此函数的功能与countDistinct一样，但是统计的结果允许有误差。这里approx_count_distinct的另一个参数就指定了可容忍的误差的最大误差。（4）sum函数：计算一个数字列中的所有值的总和。（5）sumDistinct函数：计算一个数字列中不同值的总和，重复的数值将不被计算。数据查询操作（6）avg函数：计算一个数字列的平均值。（7）min函数、max函数：统计出指定列的最小值和最大值。数据查询操作7、连接操作

实际工作中，有时候数据并不是在同一个表中，此时根据业务需求，就需要连接两个表才可以查询相关的需求结果。在DataFrame中提供了join功能实现连接的效果。1）内连接

内连接查询操作是列出与连接条件匹配的数据行，它使用比较运算符比较被连接列的列值。数据查询操作2）左外连接

此连接方式是以左表为基准，将数据进行连接，然后将左表中所有数据都显示，右表对应的列设置为NULL。数据查询操作3）右外连接

是以右表为基准，将数据进行连接，然后将右表所有的数据进行显示，左表对应的列设置为NULL。数据查询操作4）全外连接全外连接是在等值连接的基础上将左表和右表的未匹配数据都加上。数据查询操作5）LeftSemi-Join这种join类型的行为类似于内连接类型，连接后的数据集只包含匹配的行。6）交叉连接(又称为Cartesian-笛卡尔连接)当执行join连接时不指定参数，得到的也是笛卡尔积结果。数据探索分析

基于前述的知识与方法，本节将对在线销售数据集orders.csv和order-details.csv进行分析探索，解决情景导入中提出的问题。1、数据准备（1）上传orders.csv和order-details.csv数据集到HDFS文件系统中。（2）加载数据集到RDD及DataFrame中。数据探索分析2、问题分析（1）此数据集中每个客户的下单数量是多少?

每个客户的下单情况数据在DataFrameorders中。orders中的每一条数据就是一个订单。因此，只需要对其中的full_name用户姓名列进行groupBy()分类并利用count()计算总数即可。最后还可以利用sort()方法，对客户下单量进行降序排序，获得下单最多的用户的信息。数据探索分析（2）每个城市的订单有多少?此问题的分析思路与前一题相同，只需将分析的列替换为city城市。数据探索分析（3）每个客户的消费总额是多少?

针对此问题，需要对数据做一些转换。观察order-details.csv中的数据可以发现，每一条记录中记录着商品的单价和销售数量，并且有些商品是有折扣的。每条记录中，有的商品销售记录是来自同一个订单，也就是同一个客户购买的。所以要计算每个客户的年消费总额，首先要计算出每条记录的消费数量，接着根据订单编号进行求和，获得每个客户的消费总额情况。1）计算orders_details中每行商品的实际销售金额。实际销售金额=商品价格*已订购数量-商品价格*已订购数量*折扣百分比2）按照订单编号order_id统计每张订单的总金额。

根据订单编号"order_id"利用groupBy()对前面计算获得的数据orders_details1进行分类，再利用sum()函数对消费金额expend_amount列进行求和并重命名为OrderTotal，从而获得每张订单的总金额。数据探索分析3）进行等值内连接，并增加订单总金额。

客户编号、客户名字等信息均在DateFrameorders中，每张订单的总金额数据在ordertotal中，要想获得“每个客户的消费总额”，就需要连接相关的数据表。数据探索分析（4）数据集中，客户的平均订单消费额是多少？

对客户编号cust_id和客户名字full_name进行groupBy分类，再对Total_consumption做avg平均即可得到客户的平均订单消费额。小结

本章介绍了使用SparkSQL进行数据分析与探索。通过“女装电子商务评论情况分析”和“在线销售订单数据分析”两个案例掌握数据准备、清洗、转换、分析、输出、查询过程中用到的DataFrame操作方法。THANKS!6Zepplin数据可视化Zepplin安装部署第一部分下载与安装测试运行Zeppelin情境导入

数据可视化可以使数据信息更清晰形象的展示给数据分析人员，也为数据分析结果汇报呈现提供了有效途径。Zeppelin是Apache基金会下的一个开源框架，它提供了一个数据可视化的框架。具有数据提取、数据发掘、数据分析、数据可视化展示与合作等功能。学习目标和要求1、会下载、安装、配置Zepplin可视化工具。2、会在Zepplin中配置Spark解释器。下载与安装Zepplin可视化工具的安装包可以直接通过其官网下载。通过下载地址：/download.html，选择zeppelin-0.10.1-bin-all.tgz包进行下载。下载与安装1、安装步骤（1）将Zepplin安装包上传到master节点的/root目录下。（2）解压并重命名目录。（3）修改配置文件zeppelin-env.sh。在文件最后添加以下配置内容：下载与安装2、Zeppelin控制台

启动Zeppelin后，通过浏览器访问30:9090可以打开Zeppelin控制台。下载与安装3、Zeppelin中的Spark解释器配置

目前，Zeppelin支持Scala、Python、Flink、SparkSQL、Hive、JDBC、Markdown、Shell等多种解释器。（1）anonymous用户按钮，打开解释器配置界面。（2）搜索spark解释器，进入spark解释器配置界面进行设置。测试运行Zeppelin1、测试自带的BasicFeature(Spark)案例

点击菜单栏中的Notebook按钮，找到并点击SparkBasicFeatures，本案例是对一份银行数据进行分析及可视化。测试运行Zeppelin2、测试运行自编代码（1）Zeppelin控制台首页，点击“Createnewnote”按钮，创建一个新的notebook。（2）在弹出的窗口中填写新文件的路径和名称，选择解释器，此处选择默认的spark解释器即可。最后点击Create按钮，创建文件。（3）在新建的notebook界面执行以下代码。女装电子商务评论情况数据可视化第二部分加载数据注册视图执行SQL数据可视化情境导入

在第五章中，我们利用SparkSQL在spark-shell中输入代码，对“女装电子商务评论情况”数据集进行了分析。但是通过此方法分析获得的结果都是二维数据表数据，数据特征不够直观生动，不利于总结分析汇报。因此，此女装线上商店负责人希望能对数据进行可视化，以帮助她们更好的进行决策。在Zeppelin的notebook中使用SQL解释器，可以将查询的结果以条状图、散点图、折线图等常见的方式进行可视化展示。学习目标和要求1、掌握在Zepplin中使用SparkSQL注册视图2、能使用Zepplin执行SparkSQL命令实现数据可视化。加载数据注册视图1、启动Hadoop集群、Spark集群和Zeppelin启动。在Zeppelin中创建一个新的notebook，并命名为“clothing_reviews”。2、读取HDFS文件系统中的CSV文件，自定义schema并转换为DataFrame。加载数据注册视图3、对“Clothing-Reviews.csv”女装电子商务评论数据集进行数据清洗，删除重复的数据，对有空缺数据的行进行删除。加载数据注册视图4、对“Clothing-Reviews.csv”女装电子商务评论数据集进行数据转换，将recommended_IND（是否推荐）列中的1替换为推荐，0替换为不推荐。加载数据注册视图5、使用createOrReplaceTempView()方法创建本地的临时视图clothing_review