Spark SQL与DataFrame详解:案例解析(第6天)

系列文章目录

• 1- Spark SQL的基本介绍（了解）
• 2- Spark SQL的入门案例（掌握）
• 3- DataFrame案例详解（掌握）
• 4- Spark SQL的综合案例分析（掌握）

---

---

前言

本文主要通过各种案例详解了Spark SQL以及DataFrame

---

一、Spark SQL 基本介绍（了解）

1. 什么是Spark SQL

Spark SQL是Spark多种组件中其中一个，主要是用于处理大规模的【结构化数据】

什么是结构化数据: 一份数据, 每一行都有固定的列, 每一列的类型都是一致的 我们将这样的数据称为结构化的数据
例如: mysql的表数据
    1 张三 20
    2 李四 15
    3 王五 18
    4 赵六 12
1
2
3
4
5
6

为什么要学习Spark SQL呢?

1- 会 SQL的人, 一定比会大数据的人多
2- Spark SQL 既可以编写SQL语句, 也可以编写代码, 甚至可以混合使用
3- Spark SQL 可以 和 HIVE进行集成, 集成后, 可以替换掉HIVE原有MR的执行引擎, 提升效率
1
2
3

Spark SQL特点:

1- 融合性: 既可以使用标准SQL语言, 也可以编写代码, 同时支持混合使用

2- 统一的数据访问: 可以通过统一的API来对接不同的数据源

3- HIVE的兼容性: Spark SQL可以和HIVE进行整合, 整合后替换执行引擎为Spark, 核心: 基于HIVE的metastore来处理

4- 标准化连接: Spark SQL也是支持 JDBC/ODBC的连接方式
1
2
3
4
5
6
7

2. Spark SQL 与 HIVE异同

相同点:

1- 都是分布式SQL计算引擎
2- 都可以处理大规模的结构化数据
3- 都可以建立Yarn集群之上运行
1
2
3

不同点:

1- Spark SQL是基于内存计算, 而HIVE SQL是基于磁盘进行计算的
2- Spark SQL没有元数据管理服务(自己维护), 而HIVE SQL是有metastore的元数据管理服务的
3- Spark SQL底层执行Spark RDD程序, 而HIVE SQL底层执行是MapReduce
4- Spark SQL可以编写SQL也可以编写代码，但是HIVE SQL仅能编写SQL语句
1
2
3
4

3. Spark SQL的数据结构对比

说明:
	pandas的DataFrame: 二维表  处理单机结构数据
	SparkCore的RDD: 处理任何的数据结构   处理大规模的分布式数据
	SparkSQL的DataFrame: 二维表  处理大规模的分布式结构数据
1
2
3
4

RDD（Resilient Distributed Dataset）是Spark中最基本的抽象，代表了一个不可变、分布式的数据集合。RDD支持并行操作，可以在集群中的多个节点上进行处理。RDD具有容错性，即使在节点故障时也能够自动恢复。但是RDD只提供了基本的功能，对于结构化数据的处理能力有限。

DataFrame是Spark SQL中的一个概念，它是一种以列为主的分布式数据集合，类似于关系型数据库中的表格。DataFrame具有数据结构化的特点，每一列都有相应的数据类型，而且可以使用SQL语句进行查询和操作。DataFrame也支持大部分RDD的操作，但是在处理结构化数据方面更加方便。

DataSet是Spark 2.0引入的一种新的API，它是DataFrame的一个扩展，提供了类型安全的数据操作。DataSet在编译时检查数据类型，可以避免一些运行时的错误。与DataFrame相比，DataSet更加适用于需要强类型支持的场景，但是在灵活性和易用性方面可能略逊于DataFrame。


由于Python不支持泛型, 所以无法使用Dataset类型, 客户端仅支持DataFrame类型
1
2
3
4
5
6
7
8

二、Spark SQL的入门案例（掌握）

SparkSession 和 SparkContext 是 Apache Spark 中两个重要的组件，它们在 Spark 应用程序中扮演着不同的角色。

SparkContext:
        SparkContext 是 Spark 1.x 版本中最重要的入口点，在 Spark 2.x 版本中，它已经被 SparkSession 取代，但在一些旧的代码和文档中仍然可能会看到它的存在。
        SparkContext 是 Spark 应用程序与 Spark 集群通信的主要入口点。它负责与集群管理器（如 YARN、Mesos 或 Spark 自带的 Standalone）通信，以便分配资源和执行任务。
        SparkContext 提供了创建 RDD（弹性分布式数据集）的功能，RDD 是 Spark 中基本的数据抽象，代表了分布在集群中的不可变的数据集。
        

SparkSession:
        在 Spark 2.x 中，SparkSession 被引入来取代 SparkContext，并提供了更多功能和简化的 API。，它是 Spark 应用程序中的入口点，封装了 SparkContext。
        SparkSession 提供了一种统一的入口点，用于读取数据、执行查询、进行数据处理等各种 Spark 任务。
        SparkSession 提供了 DataFrame 和 Dataset API，这两种 API 提供了更高级别、更易于使用的抽象，用于处理结构化数据。
        与 SparkContext 不同，SparkSession 可以与 Hive 集成，允许在 Spark 应用程序中执行 SQL 查询，并访问 Hive 中的表和数据。

总之，SparkContext 是 Spark 1.x 版本中的主要入口点，负责与集群通信和管理资源，而 SparkSession 是 Spark 2.x 中的主要入口点，提供了更多的功能和简化的 API，用于执行各种 Spark 任务，并且可以与 Hive 集成。还可以通过SparkSession对象还是可以得到SparkContext对象。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

入门体验

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()
    sc = spark.sparkContext
    # print(spark,type(spark))
    # print(sc,type(sc))

    # 2.验证是否能生成rdd
    textRDD = sc.textFile('file:///export/data/spark_project/spark_sql/data/uniqlo.csv')
    # collect: 搜集数据触发任务展示数据  count:获取数据条数  type:查看类型
    # print(textRDD.collect())
    print(textRDD.count())
    print(type(textRDD)) # <class 'pyspark.rdd.RDD'>

    # 验证是否能生成DataFrame
    df = spark.read.csv('file:///export/data/spark_project/spark_sql/data/uniqlo.csv')
    # show: 展示数据  count:获取数据条数  type:查看类型
    # print(df.show())
    print(df.count())
    print(type(df)) # <class 'pyspark.sql.dataframe.DataFrame'>

    # 3.关闭资源
    sc.stop()
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

三、DataFrame详解（掌握）

1. DataFrame基本介绍

DataFrame表示的是一个二维的表。二维表，必然存在行、列等表结构描述信息

表结构描述信息(元数据Schema): StructType对象
字段: StructField对象，可以描述字段名称、字段数据类型、是否可以为空
行: Row对象
列: Column对象，包含字段名称和字段值

在一个StructType对象下，由多个StructField组成，构建成一个完整的元数据信息
1
2
3
4
5
6
7
8

如何构建表结构信息数据：

2. DataFrame的构建方式

方式1: 使用SparkSession的createDataFrame(data,schema)函数创建
    data参数
        1.基于List列表数据进行创建
        2.基于RDD弹性分布式数据集进行创建
        3.基于pandas的DataFrame数据进行创建
    schema参数
        1: 字符串
            格式一 :“字段名1 字段类型,字段名2 字段类型”
            格式二(推荐):“字段名1:字段类型,字段名2:字段类型”
        2: List
            格式: ["字段名1","字段名2"]  
        3: DataType（推荐，用的最多）
           格式一:schema=StructType().add('字段名1',字段类型).add('字段名2',字段类型)
           格式二:schema=StructType([StructField('字段名1',类型),StructField('字段名1',类型)])
 
方式2: 使用DataFrame的toDF(colNames)函数创建
	DataFrame的toDF方法是一个在Apache Spark的DataFrame API中用来创建一个新的DataFrame的方法。这个方法可以将一个RDD转换为DataFrame，或者将一个已存在的DataFrame转换为另一个DataFrame。在Python中，你可以使用toDF方法来指定列的名字。如果你不指定列的名字，那么默认的列的名字会是_1, _2等等。 
	格式: rdd.toDF([列名])


方式3: 使用SparkSession的read()函数创建
    在 Spark 中，SparkSession 的 read 是用于读取数据的入口点之一，它提供了各种方法来读取不同格式的数据并将其加载到 Spark 中进行处理。
    统一API格式: 
        spark.read
            .format('text|csv|json|parquet|orc|...')  : 读取外部文件的方式
            .option('k','v')   : 选项  可以设置相关的参数 (可选)
            .schema(StructType | String)  :  设置表的结构信息
            .load('加载数据路径')  : 读取外部文件的路径, 支持 HDFS 也支持本地
    简写API格式:
        注意: 以上所有的外部读取方式，都有简单的写法。spark内置了一些常用的读取方案的简写
        格式: spark.read.文件读取方式()

        注意: parquet:是Spark中常用的一种列式存储文件格式和Hive中的ORC差不多, 他俩都是列存储格式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

2.1 createDataFrame()创建

场景：一般用在开发和测试中。因为只能处理少量的数据

2.1.1 基于列表

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()

    # 2.创建DF对象
    data = [(1, '张三', 18), (2, '李四', 28), (3, '王五', 38)]
    df1 = spark.createDataFrame(data,schema=['id','name','age'])
    # 展示数据
    df1.show()
    # 查看结构信息
    df1.printSchema()

    print('---------------------------------------------------------')
    df2 = spark.createDataFrame(data,schema='id int,name string,age int')
    # 展示数据
    df2.show()
    # 查看结构信息
    df2.printSchema()

    print('---------------------------------------------------------')
    df3 = spark.createDataFrame(data,schema='id:int,name:string,age:int')
    # 展示数据
    df3.show()

    # 查看结构信息
    df3.printSchema()

    # 3.关闭资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

2.1.2 基于RDD普通方式

场景：RDD可以存储任意结构的数据；而DataFrame只能处理二维表数据。在使用Spark处理数据的初期，可能输入进来的数据是半结构化或者是非结构化的数据，那么我可以先通过RDD对数据进行ETL处理成结构化数据，再使用开发效率高的SparkSQL来对后续数据进行处理分析。

Schema选择StructType对象来定义DataFrame的“表结构”转换RDD

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
from pyspark.sql.types import StructType, StringType, StructField

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()
    sc = spark.sparkContext


    # 2.读取生成rdd
    textRDD = sc.textFile('file:///export/data/spark_project/spark_sql/data/data1.txt')
    print(type(textRDD)) # <class 'pyspark.rdd.RDD'>
    etlRDD = textRDD.map(lambda line:line.split(',')).map(lambda l:(l[0],l[1]))
    # 3.定义schema结构信息
    schema1 = StructType().add('name',StringType(),True).add('age',StringType(),True)
    schema2 = StructType([StructField('name',StringType(),True),StructField('age',StringType(),True)])
    schema3 = ['name','age']
    schema4 = 'name string,age string'
    schema5 = 'name:string,age:string'
    # 4.创建DF对象
    dfpeople = spark.createDataFrame(etlRDD,schema5)
    # 5.df展示结构信息
    dfpeople.show()
    dfpeople.printSchema()
    # 6.拓展: 创建临时视图,方便sql查询
    dfpeople.createTempView('peoples')
    r = spark.sql('select * from peoples')
    r.show()


    # 7.关闭资源
    sc.stop()
    spark.stop()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

2.1.3 基于RDD反射方式

Schema使用反射方法来推断Schema模式Spark SQL 可以将 Row 对象的 RDD 转换为 DataFrame，从而推断数据类型。

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
from pyspark.sql.types import StructType, StringType, StructField, Row

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()
    sc = spark.sparkContext


    # 2.读取生成rdd
    # 3.定义schema结构信息
    textRDD = sc.textFile('file:///export/data/spark_project/spark_sql/data/data1.txt')
    etlRDD_schema = textRDD.map(lambda line:line.split(',')).map(lambda l:Row(name=l[0],age=l[1]))

    # 4.创建DF对象
    dfpeople = spark.createDataFrame(etlRDD_schema)

    # 5.df展示结构信息
    dfpeople.show()
    dfpeople.printSchema()
    # 6.拓展: 创建临时视图,方便sql查询
    dfpeople.createTempView('peoples')
    r = spark.sql('select * from peoples')
    r.show()



    # 7.关闭资源
    sc.stop()
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

2.2 toDF()创建

schema模式编码在字符串中,toDF参数用于指定列的名字。如果你不指定列的名字，那么默认的列的名字会是_1, _2等等。

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
from pyspark.sql.types import StructType, StringType, StructField, Row

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()
    sc = spark.sparkContext


    # 2.读取生成rdd
    # 3.定义schema结构信息
    textRDD = sc.textFile('file:///export/data/spark_project/spark_sql/data/data1.txt')
    etlRDD = textRDD.map(lambda line:line.split(','))

    # 4.创建DF对象
    dfpeople = etlRDD.toDF(['name','age'])

    # 5.df展示结构信息
    dfpeople.show()
    dfpeople.printSchema()
    # 6.拓展: 创建临时视图,方便sql查询
    dfpeople.createTempView('peoples')
    r = spark.sql('select * from peoples')
    r.show()

    # 7.关闭资源
    sc.stop()
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

2.3 read读取外部文件

复杂API

统一API格式: 
spark.read
	.format('text|csv|json|parquet|orc|avro|jdbc|.....') # 读取外部文件的方式
	.option('k','v') # 选项  可以设置相关的参数 (可选)
	.schema(StructType | String) #  设置表的结构信息
	.load('加载数据路径') # 读取外部文件的路径, 支持 HDFS 也支持本地
1
2
3
4
5
6

简写API

请注意: 以上所有的外部读取方式，都有简单的写法。spark内置了一些常用的读取方案的简写
格式: 
	spark.read.读取方式()
	
例如: 
	df = spark.read.csv(
   		path='file:///export/data/_03_spark_sql/data/stu.txt',
        header=True,
        sep=' ',
        inferSchema=True,
        encoding='utf-8',
    )
1
2
3
4
5
6
7
8
9
10
11
12

2.3.1 Text方式读取

text方式读取文件：
    1- 不管文件中内容是什么样的，text会将所有内容全部放到一个列中处理
    2- 默认生成的列名叫value，数据类型string
    3- 我们只能够在schema中修改字段value的名称，其他任何内容不能修改
1
2
3
4

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
from pyspark.sql.types import StructType, StringType, StructField, Row

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()

    # 2.读取数据
    # 注意: 读取text文件默认只有1列,且列名交value,可以通过schema修改
    df = spark.read\
        .format('text')\
        .schema('info string')\
        .load('file:///export/data/spark_project/spark_sql/data/data1.txt')


    # 5.df展示结构信息
    df.show()
    df.printSchema()
    # 6.拓展: 创建临时视图,方便sql查询
    df.createTempView('peoples')
    r = spark.sql('select * from peoples')
    r.show()


    # 6.关闭资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

2.3.2 CSV方式读取

csv格式读取外部文件：
    1- 复杂API和简写API都必须掌握
    2- 相关参数作用说明：
        2.1- path：指定读取的文件路径。支持HDFS和本地文件路径
        2.2- schema：手动指定元数据信息
        2.3- sep：指定字段间的分隔符
        2.4- encoding：指定文件的编码方式
        2.5- header：指定文件中的第一行是否是字段名称
        2.6- inferSchema：根据数据内容自动推断数据类型。但是，推断结果可能不精确
1
2
3
4
5
6
7
8
9

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
from pyspark.sql.types import StructType, StringType, StructField, Row

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()

    # 2.读取数据
    # 注意: csv文件可以识别多个列,可以使用schema指定列名,类型
    # 原始方式
    # df = spark.read\
    #     .format('csv')\
    #     .schema('name string,age int')\
    #     .option('sep',',')\
    #     .option('encoding','utf8')\
    #     .option('header',False)\
    #     .load('file:///export/data/spark_project/spark_sql/data/data1.txt')
    # 简化方式
    df = spark.read.csv(
        schema='name string,age int',
        sep=',',
        encoding='utf8',
        header=False,
        path='file:///export/data/spark_project/spark_sql/data/data1.txt'
    )

    # 5.df展示结构信息
    df.show()
    df.printSchema()
    # 6.拓展: 创建临时视图,方便sql查询
    df.createTempView('peoples')
    r = spark.sql('select * from peoples')
    r.show()


    # 7.关闭资源
    spark.stop()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

2.3.3 JSON方式读取

json读取数据：
1- 需要手动指定schema信息。如果手动指定的时候，字段名称与json中的key名称不一致，会解析不成功，以null值填充
2- csv/json中schema的结构，如果是字符串类型，那么字段名称和字段数据类型间，只能以空格分隔
1
2
3

json的数据内容

{'id': 1,'name': '张三','age': 20}
{'id': 2,'name': '李四','age': 23,'address': '北京'}
{'id': 3,'name': '王五','age': 25}
{'id': 4,'name': '赵六','age': 29}
1
2
3
4

代码实现

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
from pyspark.sql.types import StructType, StringType, StructField, Row

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()

    # 2.读取数据
    # 注意: json的key和schema指定的字段名不一致,会用null补充,如果没有数据也是用null补充
    # 简化方式
    df = spark.read.json(
        schema='id int,name string,age int,address string',
        encoding='utf8',
        path='file:///export/data/spark_project/spark_sql/data/data2.txt'
    )

    # 5.df展示结构信息
    df.show()
    df.printSchema()
    # 6.拓展: 创建临时视图,方便sql查询
    df.createTempView('peoples')
    r = spark.sql('select * from peoples')
    r.show()

    # 关闭资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

四、DataFrame的相关API

操作DataFrame一般有二种操作方案：一种为【SQL方式】，另一种为【DSL方式】

SQL方式: 通过编写SQL语句完成统计分析操作
DSL方式: 特定领域语言，使用DataFrame特有的API完成计算操作，也就是代码形式

从使用角度来说: SQL可能更加的方便一些，当适应了DSL写法后，你会发现DSL要比SQL更好用
从Spark角度来说: 更推荐使用DSL方案，此种方案更加利于Spark底层的优化处理
1
2
3
4
5

1. SQL相关的API

• 创建一个视图/表

df.createTempView('视图名称'): 创建一个临时的视图(表名)
注意: 临时视图仅能在当前这个Spark Session的会话中使用

df.createOrReplaceTempView('视图名称'): 创建一个临时的视图(表名)，如果视图存在，直接替换
注意: 临时视图仅能在当前这个Spark Session的会话中使用

df.createGlobalTempView('视图名称'): 创建一个全局视图，运行在一个Spark应用中多个spark会话中都可以使用。
注意: 在使用的时候必须通过 global_temp.视图名称 方式才可以加载到。了解即可,较少使用
1
2
3
4
5
6
7
8

• 执行SQL语句

spark.sql('书写SQL')
1

2. DSL相关的API

官网链接: https://spark.apache.org/docs/3.1.2/api/python/reference/pyspark.sql.html#dataframe-apis

• select()：类似于SQL中select, SQL中select后面可以写什么, 这样同样也一样

• distinct(): 去重后返回一个新的DataFrame

• withColumn(参数1,参数2)：用来产生新列。参数1是新列的名称；参数2是新列数据的来源

• withColumnRenamed(参数1,参数2)：给字段重命名操作。参数1是旧字段名，参数2是新字段名

• alias(): 返回设置了别名的新DataFrame

• agg()：执行聚合操作。如果有多个聚合，聚合之间使用逗号分隔即可,比较通用

• where()和filter()：用于对数据进行过滤操作, 一般在spark SQL中主要使用where

• groupBy()：使用指定的列对DataFrame进行分组，方便后期对它们进行聚合

• orderBy()：返回按指定列排序的新DataFrame

• limit() : 返回指定数目的结果集

• show()：用于展示DF中数据, 默认仅展示前20行

  • 参数1：设置默认展示多少行 默认为20
  • 参数2：是否为阶段列, 默认仅展示前20个字符数据, 如果过长, 不展示(一般不设置)

• printSchema()：用于打印当前这个DF的表结构信息

DSL主要支持以下几种传递的方式:  str | Column对象 | 列表
	str格式:  '字段'
	Column对象:  
		DataFrame含有的字段  df['字段']
		执行过程新产生:  F.col('字段')
	列表: 
		['字段1','字段2'...]
		[df['字段1'],df['字段2']]
1
2
3
4
5
6
7
8

为了能够支持在编写Spark SQL的DSL时候，在DSL中使用SQL函数，专门提供一个SQL的函数库。直接加载使用

链接: https://spark.apache.org/docs/3.1.2/api/sql/index.html

导入这个函数库: import pyspark.sql.functions as F
通过F调用对应的函数即可,常见函数如下:
    F.explode()
    F.split()
    F.count()
    F.sum()
    F.avg()
    F.max()
    F.min()
    ...
1
2
3
4
5
6
7
8
9
10

3. Spark SQL的词频统计案例分析

准备一个words.txt的文件,words.txt文件的内容如下:

hadoop hive hadoop sqoop hive
sqoop hadoop zookeeper hive hue
hue sqoop hue zookeeper hive
spark oozie spark hadoop oozie
hive oozie spark hadoop
1
2
3
4
5

需求分析：

1- 扫描文件将每行内容切分得到单个的单词

2- 组织DataFrame的数据结构,分别利用SQL风格和DSL风格完成每个单词个数统计

3- 要求最后结果有两列:一列是单词，一列是次数

代码实现：

# 导包
import os
from pyspark.sql import SparkSession,functions as F

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建spark对象
    # appName:应用程序名称  master:提交模式
    # getOrCreate:在builder构建器中获取一个存在的SparkSession，如果不存在，则创建一个新的
    spark = SparkSession.builder.appName('sparksql_demo').master('local[*]').getOrCreate()

    # 2.通过read读取外部文件方式创建DF对象
    df = spark.read\
        .format('text')\
        .schema('words string')\
        .load('file:///export/data/spark_project/spark_sql/data/data3.txt')

    print(type(df))

    # 需求: 从data3.txt读取所有单词,然后统计每个单词出现的次数

    # 3.SQL风格
    # 方式1: 使用子查询方式
    # 先创建临时视图,然后通过sql语句查询展示
    df.createTempView('words_tb')
    qdf = spark.sql(
        "select words,count(1) as cnt from (select explode(split(words,' ')) as words from words_tb) t group by words"
    )
    print(type(qdf))
    qdf.show()
    
    # # 方式2: 使用侧视图
    # qdf = spark.sql(
    #     "select t.words,count(1) as cnt from words_tb lateral view explode(split(words,' ')) t as words  group by t.words"
    # )
    print(type(qdf))
    qdf.show()

    # 4.DSL风格
    # 方式1: 分组后直接用count()统计
    df.select(
        F.explode(F.split('words', ' ')).alias('words')
    ).groupBy('words').count().show()
    # 方式1升级版:通过withColumnRenamed修改字段名
    df.select(
        F.explode(F.split('words', ' ')).alias('words')
    ).groupBy('words').count().withColumnRenamed('count','cnt').show()

    # 方式2: 分组后用agg函数
    df.select(
        F.explode(F.split('words', ' ')).alias('words')
    ).groupBy('words').agg(
        F.count('words').alias('cnt')
    ).show()

    # 方式3: 直接使用withColum
    df.withColumn(
        'words',
        F.explode(F.split('words', ' '))
    ).groupBy('words').agg(
        F.count('words').alias('cnt')
    ).show()

    # 5.释放资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

4. 清洗相关的API

总结:
1- dropDuplicates(subset):用来删除重复数据。
	1.1- 如果没有指定参数subset，那么要比对行中的所有字段内容，如果全部相同，就认为是重复数据，会被删除；
	1.2- 如果有指定参数subset，那么只比对subset中指定的字段范围内删除重复数据
            
2- dropna(thresh,subset):删除缺失值数据.
   2.1- 如果不传递任何参数，只要有任意一个字段值为null,那么就删除整行数据
   2.2- 如果只指定了subset,那么空值的检查,就只会限定在subset指定的范围内
   2.3- 如果只指定了thresh,那么空值检查的这些字段中,至少需要有thresh个字段的值不为空,才不会被删除
            
3- fillna(value,subset): 替换缺失值数据
   3.1- value: 必须要传递参数.是用来填充缺失值的,默认填充所有的缺失值
   3.1- subset: 如果有指定参数subset，那么只比对subset中指定的字段范围内替换
   
   注意:value最常用的是传递字典的形式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

代码演示:

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'


def demo1_dropDuplicates():
    # dropDuplicates: 如果有重复就会去重,保留第一个,当然也可以指定参考的列
    df.dropDuplicates().show()
    df.dropDuplicates(['name']).show()


def demo2_dropna():
    # dropna: 默认去除带null每行数据,当然也可以指定参考的列
    df.dropna().show()
    df.dropna(thresh=2).show()
    df.dropna(thresh=1, subset=['id', 'name']).show()


def demo3_fillna():
    # fillna: 默认只要带null就补充指定内容,当然也可以指定参考列
    df.fillna('空').show()
    df.fillna('空', subset=['address']).show()
    df.fillna({'name': '未知', 'address': '广州'}).show()


# 创建main函数
if __name__ == '__main__':
    # 1.创建spark对象
    # appName:应用程序名称  master:提交模式
    # getOrCreate:在builder构建器中获取一个存在的SparkSession，如果不存在，则创建一个新的
    spark = SparkSession.builder.appName('sparksql_demo').master('local[*]').getOrCreate()

    # 2.通过read读取外部文件方式创建DF对象
    df = spark.read \
        .format('csv') \
        .option('header', True) \
        .load('file:///export/data/spark_project/spark_sql/data/clear_data.csv')

    # 3.show直接展示
    df.show()

    # 4.清洗数据
    # dropDuplicates: 如果有重复就会去重,保留第一个,当然也可以指定参考的列
    # demo1_dropDuplicates()

    print('------------------------------------------------------')
    # dropna: 默认去除带null每行数据,当然也可以指定参考的列
    # demo2_dropna()

    print('------------------------------------------------------')
    # fillna: 默认只要带null就补充指定内容,当然也可以指定参考列
    demo3_fillna()

    # 5.释放资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

5. Spark SQL的Shuffle分区设置

补充：

如果运行sparksql，发现Shuffle分区每次都是1,或者后续count_distinct找不到，那么是因为pyspark版本原因导致。解决办法如下：

1- 检查自己3台机器的pyspark版本是否是3.1.2版本

pip list | grep pyspark

2-如果不是3.1.2版本，那么先卸载pyspark

命令: pip uninstall pyspark

3- 再按照【Spark课程阶段_部署文档.doc】中【3.5.1章节】重新安装3.1.2版本pyspark

命令: pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark==3.1.2

​

​ Spark SQL底层本质上还是Spark的RDD程序，认为 Spark SQL组件就是一款翻译软件，用于将SQL/DSL翻译为Spark RDD程序, 执行运行

​ Spark SQL中同样也是存在shuffle的分区的,在执行shuffle分区后, shuffle分区数量默认为 200个,但是实际中, 一般都是需要调整这个分区的, 因为当数据量比较少的数据, 200个分区相对来说比较大一些, 但是当数据量比较大的时候, 200个分区显得比较小

如何调整shuffle分区数量呢? spark.sql.shuffle.partitions

方案一（不推荐）:  直接修改spark的配置文件spark-defaults.conf。全局设置，默认值为200。设置为: 
		spark.sql.shuffle.partitions     20

方案二（常用，推荐使用）: 在客户端通过submit命令提交的时候, 动态设置shuffle的分区数量。部署、上线的时候、基于spark-submit提交运行的时候
	./spark-submit --conf "spark.sql.shuffle.partitions=20"

方案三（比较常用）: 在代码中设置。主要在测试环境中使用, 但是一般在部署上线的时候, 会删除。优先级也是最高的。一般的使用场景是，当你的数据量未来不会发生太大的波动。
设置shuffle分区的数量方式1:SparkSession.builder.config('spark.sql.shuffle.partitions', 1)
设置shuffle分区的数量方式2:spark.conf.set("spark.sql.shuffle.partitions",'1')
	
获取shuffle分区的数量:spark.conf.get("spark.sql.shuffle.partitions")
1
2
3
4
5
6
7
8
9
10
11

# 导包
import os
import time
from pyspark.sql import SparkSession, functions as F

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 7.1 TODO: 记录程序开始时间
    start = time.time()

    # 1.创建spark对象
    # appName:应用程序名称  master:提交模式
    # getOrCreate:在builder构建器中获取一个存在的SparkSession，如果不存在，则创建一个新的
    spark = SparkSession.builder \
        .config('spark.sql.shuffle.partitions', 1)\
        .appName('sparksql_demo')\
        .master('local[*]')\
        .getOrCreate()

    # 获取shuffle分区的数量
    shuffle_partitions = spark.conf.get("spark.sql.shuffle.partitions")
    print("Shuffle partitions:", shuffle_partitions)
    
    # 2.通过read读取外部文件方式创建DF对象
    df = spark.read \
        .format('text') \
        .schema('words string') \
        .load('file:///export/data/spark_project/spark_sql/data/data3.txt')

    print(type(df))

    # 需求: 从data3.txt读取所有单词,然后统计每个单词出现的次数

    # 3.SQL风格
    # 方式1: 使用子查询方式
    # 先创建临时视图,然后通过sql语句查询展示
    df.createTempView('words_tb')
    qdf = spark.sql(
        "select words,count(1) as cnt from (select explode(split(words,' ')) as words from words_tb) t group by words"
    )
    print(type(qdf))
    qdf.show()

    # 4.DSL风格
    # 方式2: 分组后用agg函数
    df.select(
        F.explode(F.split('words', ' ')).alias('words')
    ).groupBy('words').agg(
        F.count('words').alias('cnt')
    ).show()

    # 7.2 TODO:记录结束时间
    end = time.time()
    # 7.3 计算运行时间
    # 结论: 合理设置分区数,效率会提高!
    t = end - start
    print(f"程序运行了{t}秒")

    # 6.TODO:为了方便查看web页面可以让程序多睡会儿
    # time.sleep(1000)

    # 5.释放资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

6. 数据写出操作

6.2 写出到文件

统一的输出语法:

对应的简写API格式如下，以CSV为例：
init_df.write.csv(
    path='存储路径',
    mode='模式',
    header=True,
    sep='\001',
    encoding='UTF-8'
)
常用参数说明：
    1- path：指定结果数据输出路径。支持本地文件系统和HDFS文件系统
    2- mode：当输出目录中文件已经存在的时候处理办法
        2.1- append：追加。如果文件已经存在，那么继续在该目录下产生新的文件
        2.2- overwrite：覆盖。如果文件已经存在，那么就先将已有的文件清除，再写入进去
        2.3- ignore：忽略。如果文件已经存在，那么不执行任何操作
        2.4- error：报错。如果文件已经存在，那么直接报错。会报错AnalysisException: path 	
        			file:xxx already exists.
        
    3- sep：字段间的分隔符
    4- header：数据输出的时候，是否要将字段名称输出到文件的第一行。推荐设置为True
    5- encoding：文件输出的编码方式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

• 演示1: 输出到文件中 json csv orc text …

# 导包
import os
from pyspark.sql import SparkSession

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()

    spark.conf.set("spark.sql.shuffle.partitions", '1')
    # 2.数据输入
    df = spark.read\
        .format('csv')\
        .option('header',True)\
        .load('file:///export/data/spark_project/spark_sql/data/clear_data.csv')
    # 3.数据处理(切分,转换,分组聚合)
    etldf = df.dropDuplicates().dropna()
    etldf.show()
    # 4.数据输出
    # 原始API
    etldf.write\
        .format('csv')\
        .option('sep',',')\
        .option('header',True)\
        .mode('overwrite')\
        .save('file:///export/data/spark_project/spark_sql/data/output')
    # 简化API
    etldf.write.csv(
        sep=',',
        header=True,
        mode='overwrite',
        path='file:///export/data/spark_project/spark_sql/data/output2'
    )

    # 5.关闭资源
    spark.stop()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

6.1 写出到数据库

• 将结果数据基于JDBC方案, 输出到关系型数据库, 例如说: MySql

from pyspark import SparkConf, SparkContext
import os
from pyspark.sql import SparkSession

# 绑定指定的Python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    print("数据输出到数据库")

    # 1- 创建SparkSession对象
    spark = SparkSession.builder\
        .config("spark.sql.shuffle.partitions","1")\
        .appName('sparksql_database')\
        .master('local[*]')\
        .getOrCreate()

    # 2- 数据输入
    init_df = spark.read.csv(
        path='file:///export/data/gz16_pyspark/02_spark_sql/data/stu.txt',
        sep=' ',
        encoding='UTF-8',
        header="True",
        inferSchema=True
    )

    # 3- 数据处理
    result = init_df.where('age>=20')

    # 4- 数据输出
    result.show()
    result.printSchema()

    # 数据输出到数据
    """
        创建数据库命令：create database day06 character set utf8;
    """
    result.write.jdbc(
        url='jdbc:mysql://node1:3306/day06?useUnicode=true&characterEncoding=utf-8',
        table='student',
        mode='append',
        properties={ 'user' : 'root', 'password' : '123456' }
    )

    # 5- 释放资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

运行结果截图：

可能出现的错误一:

    etldf.write.jdbc(
       url='jdbc:mysql://node1:3306/库名',
       table='表名',
       mode='append',
       # 解决方法: 给密码123456加上引号
       properties={'user':'root','password':'123456'}
    )
1
2
3
4
5
6
7

可能出现的错误 二:

原因:  缺少连接MySQL数据库的驱动
解决方法如下:
数据库的驱动包, 一般都是一些Jar包,放置【mysql-connector-java-5.1.41.jar】驱动包到以下位置:
	1- 放置位置一: 当spark-submit提交的运行环境为Spark集群环境的时候,以及运行模式为local, 默认从 spark的jars目录下加载相关的jar包,
		目录位置: /export/server/spark/jars
	
	2- 放置位置二: 当我们使用pycharm运行代码的时候, 基于python的环境来运行的, 需要在python的环境中可以加载到此jar包
		目录位置: 
			/root/anaconda3/lib/python3.8/site-packages/pyspark/jars/
	
	3- 放置位置三: 当我们提交选择的on yarn模式 需要保证此jar包在HDFS上对应目录下
		hdfs的spark的jars目录下:  hdfs://node1:8020/spark/jars
		
	请注意: 以上三个位置, 主要是用于放置一些 spark可能会经常使用的jar包, 对于一些不经常使用的jar包, 在后续spark-submit 提交运行的时候, 会有专门的处理方案:  spark-submit --jars  ....
1
2
3
4
5
6
7
8
9
10
11
12
13
14

可能出现的错误三：

原因：将中文输出到了数据表中
解决办法：
1- 数据库连接要加上:  ?useUnicode=true&characterEncoding=utf-8
2- 创建数据库的时候需要指定编码: character set utf8
1
2
3
4

    # 最终连接数据代码如下:
    etldf.write.jdbc(
       url='jdbc:mysql://node1:3306/spark_db1?useUnicode=true&characterEncoding=utf-8',
       table='student',
       mode='append',
       properties={'user':'root','password':'123456'}
    )
1
2
3
4
5
6
7

五、Spark SQL的综合案例（掌握）

1. 常见DSL代码整理

1- 什么使用使用select()，什么时候使用groupBy()+agg()/select()实现聚合?:如果不需要对数据分组，那么可以直接使用select()实现聚合；如果有分组操作，需要使用groupBy()+agg()/select()，推荐使用agg()
        
2- first(): 如果某个DataFrame中只有一行数据，并且不使用join来对比数据，那么一般需要使用first()明确指定和第一行进行比较
    
3- F.col(): 对于在计算过程中临时产生的字段，需要使用F.col()封装成Column对象,然后去使用
1
2
3
4
5

• API/方法：是由DataFrame来调用
• 函数：需要先通过import pyspark.sql.functions as F导入，使用F调用。Spark SQL内置提供的函数https://spark.apache.org/docs/3.1.2/api/sql/index.html
• 窗口：需要先通过from pyspark.sql import Window导入

2. 电影分析案例

2.1 需求说明

数据集的介绍:

数据说明 :  userid,movieid,score,datestr

字段的分隔符号为:  \t
1
2
3

资料已经上传

2.2 需求分析

• 需求一: 查询用户平均分

  需求分析：

  维度：用户
  指标：平均分

• 需求二: 查询每部电影的平均分（课后作业，将自己对需求的分析步骤以文字的形式放在代码中）

• 需求三: 查询大于平均分的电影的数量

  需求分析：

  1- 统计所有打分的平均分，这个结果就是一个数字

  2- 统计每部电影各自的平均分

  3- 查询大于平均分的电影的数量

• 需求四: 查询高分电影中(电影平均分大于3)打分次数最多的用户, 并且求出此人所有的打分记录中, 打的平均分是多少

  需求分析：

  1- 筛选出高分电影。统计每部电影的平均分，再过滤出>3分的电影信息

  2- 从高分电影中，统计每个用户分别打了多少次分。再选择TOP1的用户

  3- 统计该用户所有打分记录的平均分

一三四需求实现代码:

# 导包
import os
from pyspark.sql import SparkSession,functions as F

# 绑定指定的python解释器
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'


def demo1_get_user_avg_score():
    # 方式1: SQL方式
    spark.sql(
        'select userid,round(avg(score),3) as user_avg_score from movie group by userid'
    ).show()
    # 方式2: DSL方式
    etldf.groupBy('userid').agg(
        F.round(F.avg('score'), 3).alias('user_avg_score')
    ).show()


def demo2_get_lag_avg_movie_cnt():
    # 方式1: SQL
    spark.sql(
        """
        select count(1) as cnt from (
            select movieid,avg(score) as movie_avg_score from movie group by movieid
            having movie_avg_score > (select avg(score) as all_avg_score from movie)  
        ) t
        """
    ).show()
    # 方式2: DSL
    # col():把临时结果作为新列使用   first():取第一个值
    etldf.groupBy('movieid').agg(
        F.avg('score').alias('movie_avg_score')
    ).where(
        F.col('movie_avg_score') > etldf.select(F.avg('score').alias('all_avg_score')).first()['all_avg_score']
    ).agg(
        F.count('movieid').alias('cnt')
    ).show()


def demo3_get_top1_user_avg_sql():
    # 方式1: SQL
    # ①先查询高分电影:
    spark.sql(
        "select movieid,avg(score) as movie_avg_score from movie group by movieid having movie_avg_score > 3"
    ).createTempView('hight_score_tb')
    # ②再求打分次数最多的用户(先不考虑并列,只取最大1个)
    spark.sql(
        "select userid,count(1) as cnt from hight_score_tb h join movie m on h.movieid = m.movieid group by userid order by cnt desc limit 1"
    ).createTempView('top1_user_tb')
    # ③最后求此人所有打分的平均分
    spark.sql(
        "select avg(score) as top1_user_avg from movie where userid = (select userid from top1_user_tb)"
    ).show()


def demo3_get_top1_user_avg_dsl():
    # ①先查询高分电影:
    hight_score_df = etldf.groupBy('movieid').agg(
        F.avg('score').alias('movie_avg_score')
    ).where('movie_avg_score>3')
    # ②再求打分次数最多的用户(先不考虑并列,只取最大1个)
    top1_user_df = hight_score_df.join(etldf, on=hight_score_df['movieid'] == etldf['movieid']) \
        .groupBy('userid').agg(F.count('userid').alias('cnt')) \
        .orderBy('cnt', ascending=False).limit(1)
    # ③最后求此人所有打分的平均分
    etldf.where(
        etldf['userid'] == top1_user_df.first()['userid']
    ).agg(
        F.avg('score').alias('top1_user_avg')
    ).show()


# 创建main函数
if __name__ == '__main__':
    # 1.创建SparkContext对象
    spark = SparkSession.builder.appName('pyspark_demo').master('local[*]').getOrCreate()

    # 2.数据输入
    df = spark.read.csv(
        schema='userid string,movieid string,score int,datestr string',
        sep='\t',
        path='file:///export/data/spark_project/spark_sql/data/u.data'
    )
    print(df.count())
    # 3.数据处理(切分,转换,分组聚合)
    etldf = df.dropDuplicates().dropna()
    print(etldf.count())
    # 4.数据分析
    # 方便后续所有SQL方式使用,提前创建临时视图作为表
    etldf.createTempView('movie')
    # 需求1: 查询用户的平均分
    # demo1_get_user_avg_score()

    # 需求3: 查询大于平均分的电影的数量
    # demo2_get_lag_avg_movie_cnt()

    # 需求4: 查询高分电影(平均分>3)中,打分次数最多的用户,并求出此人所有打分的平均分
    # 方式1: SQL
    demo3_get_top1_user_avg_sql()
    # 方式2: DSL
    demo3_get_top1_user_avg_dsl()

    # 5.数据输出

    # 6.关闭资源
    spark.stop()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

附录: 问题

可能出现的错误一：

原因: 是使用withColumn产生新列，但是表达式中有聚合的操作。缺少groupBy调用
1

可能出现的错误二：

错误原因：DataFrame结果是单行的情况，列值获取错误

解决办法：

将df_total_avg_score['total_avg_score']改成df_total_avg_score.first()['total_avg_score']
1

可能遇到的错误三：

原因：对于在计算过程中临时产生的字段，需要使用F.col封装成Column对象

解决办法：F.col(‘avg_score’)

六、常见面试题

1. 你对Spark优化有了解吗？

1)避免创建重复的RDD
2）尽可能复用同一个RDD
3）对多次使用的RDD进行持久化
4）尽量避免使用shuffle类算子
5）使用map-sid预聚合的shuffle操作
所谓的map-side预聚合，说的是在每个节点本地对相同的key进行一次聚合操作，类似于MapReduce中
的本地combiner。map-side预聚合之后，每个节点本地就只会有一条相同的key，因为多条相同的key
都被聚合起来了。其他节点在拉取所有节点上的相同key时，就会大大减少需要拉取的数据数量，从而也
就减少了磁盘IO以及网络传输开销。如reduceByKey或者aggregateByKey代替groupByKey。
6）使用高性能算子
比如使用reduceByKey/aggregateByKey替代groupByKey
使用mapPartitions替代普通map
使用foreachPartitions替代foreach
使用filter之后进行coalesce操作
7）广播大变量
有时在开发过程中，会遇到需要在算子函数中使用外部变量的场景（尤其是大变量，比如100M以上的大
集合），那么此时就应该使用Spark的广播（Broadcast）功能来提升性能。
8）调整参数
比如num-executors，executor-memory，executor-cores，driver-memory 等参数

2. SparkSQL与Hive语法差异

• 相同函数差异
  （1）Spark运行时用到的hash函数，与Hive的哈希算法不同，如果使用hash()，结果和Hive的hash()会
  有差异。
  （2）Hive和SparkSQL使用grouping sets生成的GROUPING_ID不一致。
  （3）regexp_extract未匹配上的话，在Hive里回是null，但在Spark里返回是空字符。
  （4）SparkSQL中row_number的over中不能省略sort by 或order by
  （5）grouping_id()函数生成的数据不同
  （6）reflect()函数中，如果入参有非法数据或者null，Hive会返回null，而Spark会抛出异常
• 仅Hive支持
  （1）SparkSQL关联on条件不支持函数rand()。
  （2）创建临时表时,Spark不支持直接赋值null。
  （3）SparkSQL无法读取字段类型为void的表。
  （4）SparkSQL中如果表达式没有指定别名，SparkSQL会将整个表达式作为别名。如果表达式中包含特殊符号（如逗号），则CTAS建表会失败。
• 仅Spark支持
  （1）SparkSQL允许在join on条件中使用or等不等值关联语句，Hive中不允许，只能用等值关联。