day06_Spark SQL

王柳 · 2025-1-14 12:40:19

day06_Spark SQL课程笔记

一、今日课程内容

1- DataFrame详解（掌握）
2- Spark SQL的综合案例（掌握）
3- Spark SQL函数定义（掌握）

今日目的：掌握DataFrame详解

二、DataFrame详解（掌握）

5.清洗相干的API

API是什么

简单来说：API（应用程序编程接口）就像是“软件之间的翻译官”，定义了不同系统或组件之间如何交互和通讯，让开发者能够更方便地利用外部功能或服务。
具体而言：
- 定义：API是一组预定义的函数、协议和工具，用于构建软件应用程序。它规定了如何哀求服务、通报数据以及吸收结果。
- 类型：
  - Web API：基于HTTP协议，用于Web服务之间的通讯，如RESTful API、GraphQL。
  - 库API：编程语言提供的函数库，如Python的NumPy、Pandas。
  - 操作系统API：操作系统提供的接口，如Windows API、Linux系统调用。
- 工作原理：
  - 哀求：客户端发送哀求（如HTTP哀求）到API。
  - 处理：服务器吸收哀求并处理。
  - 响应：服务器返回结果（如JSON数据）给客户端。
实际生产场景：
- 在移动应用中，利用Web API获取天气数据或支付服务。
- 在数据分析中，利用Python库API（如Pandas）处理数据。
- 在系统开发中，利用操作系统API管理文件或进程。
总之：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最常用的是传递字典的形式

复制代码

代码演示:

# 导包
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()

复制代码

6.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.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")

复制代码

# 导包
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()

复制代码

7.数据写出操作

写出到文件

同一的输出语法:

对应的简写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: 输出到文件中 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()

复制代码

写出到数据库

将结果数据基于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',
sql={ 'user' : 'root', 'password' : '123456' }
)
# 5- 释放资源
spark.stop()

复制代码

运行结果截图：

可能出现的错误一:

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

复制代码

可能出现的错误二:

原因: 缺少连接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- 数据库连接要加上: ?useUnicode=true&characterEncoding=utf-8
2- 创建数据库的时候需要指定编码: character set utf8

复制代码

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

复制代码

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

1、常见DSL代码整理

分类格式含义示例API/方法select查询字段select(‘id1’, ‘id2’)where对数据过滤where(‘avg_score>3’)groupBy对数据分组groupBy(‘userid’)orderBy对数据排序orderBy(‘cnt’, ascending=False)limit取前几条数据orderBy(‘cnt’, ascending=False).limit(1)agg聚合操作，里面可以写多个聚合表达式agg(F.round(F.avg(‘score’), 2).alias(‘avg_score’))show打印数据init_df.show()printSchema打印数据的schema信息，也就是元数据信息init_df.printSchema()alias对字段取别名F.count(‘movieid’).alias(‘cnt’)join关联2个DataFrameetl_df.join(avg_score_dsl_df, ‘movieid’)withColumn基于目前的数据产生一个新列init_df.withColumn(‘word’,F.explode(F.split(‘value’, ’ ')))dropDuplicates删除重复数据init_df.dropDuplicates(subset=[“id”,“name”])dropna删除缺失值init_df.dropna(thresh=2,subset=[“name”,“age”,“address”])fillna替换缺失值init_df.fillna(value={“id”:111,“name”:“未知姓名”,“age”:100,“address”:“北京”})first取DataFrame中的第一行数据over创建一个窗口列窗口partitionBy对数据分区orderBy对数据排序orderBy(F.desc(‘pv’))函数row_number行号。从1开始编号desc降序排序avg计算均值count计数round保存小数位col将字段包装成Column对象，一般用于对新列的包装

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

复制代码

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、电影分析案例

需求说明:

数据集的介绍:

数据说明 : userid,movieid,score,datestr
字段的分隔符号为: \t

复制代码

需求分析:

需求一: 查询用户平均分
需求分析：
维度：用户
指标：平均分
需求二: 查询每部电影的平均分（课后作业，将自己对需求的分析步调以文字的形式放在代码中）
需求三: 查询大于平均分的电影的数量
需求分析：
1- 统计全部打分的平均分，这个结果就是一个数字
2- 统计每部电影各自的平均分
3- 查询大于平均分的电影的数量
需求四: 查询高分电影中(电影平均分大于3)打分次数最多的用户, 并且求出此人全部的打分记载中, 打的平均分是多少
需求分析：
1- 筛选出高分电影。统计每部电影的平均分，再过滤出>3分的电影信息
2- 从高分电影中，统计每个用户分别打了多少次分。再选择TOP1的用户
3- 统计该用户全部打分记载的平均分
需求五: 查询每个用户的平均打分, 最低打分, 最高打分（课后作业）
需求六: 查询被评分超过100次的电影的平均分排名 TOP10（课后作业）

一三四需求实现代码:

# 导包
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()

复制代码

附录: 题目
可能出现的错误一：

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

复制代码

可能出现的错误二：

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

解决办法：

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

复制代码

可能遇到的错误三：

缘故原由：对于在计算过程中临时产生的字段，需要利用F.col封装成Column对象
解决办法：F.col(‘avg_score’)
四、Spark SQL函数定义

1、窗口函数

回首之前学习过的窗口函数：

分析函数 over(partition by xxx order by xxx [asc|desc] [rows between xxx and xxx])
分析函数可以大致分成如下3类：
1- 第一类: 聚合函数 sum() count() avg() max() min()
2- 第二类: row_number() rank() dense_rank() ntile()
3- 第三类: first_value() last_value() lead() lag()

复制代码

在Spark SQL中利用窗口函数案例：
需求是找出每个cookie中pv排在前3位的数据，也就是分组取TOPN题目

from pyspark import SparkConf, SparkContext
import os
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql import Window as win
# 绑定指定的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__':
# 1- 创建SparkSession对象
spark = SparkSession.builder\
.config('spark.sql.shuffle.partitions',1)\
.appName('sparksql_win_function')\
.master('local[*]')\
.getOrCreate()
# 2- 数据输入
init_df = spark.read.csv(
path='file:///export/data/gz16_pyspark/02_spark_sql/data/cookie.txt',
schema='cookie string,datestr string,pv int',
sep=',',
encoding='UTF-8'
)
init_df.createTempView('win_data')
init_df.show()
init_df.printSchema()
# 3- 数据处理
# SQL
spark.sql("""
select
cookie,datestr,pv
from (
select
cookie,datestr,pv,
row_number() over (partition by cookie order by pv desc) as rn
from win_data
) tmp where rn<=3
""").show()
# DSL
"""
select：注意点，结果中需要看到哪几个字段，就要明确写出来
"""
init_df.select(
"cookie","datestr","pv",
F.row_number().over(win.partitionBy('cookie').orderBy(F.desc('pv'))).alias('rn')
).where('rn<=3').select("cookie","datestr","pv").show()
# 4- 数据输出
# 5- 释放资源
spark.stop()

复制代码

运行结果截图：

2、SQL函数分类

SQL函数，主要分为以下三大类：

UDF函数：用户自定义函数
- 特点：一对一，输入一个得到一个
- 比方：split() substr()
- 在Hive中继承UDF类，方法名称为evaluate，返回值不能为void，实在就是实现一个方法；
UDAF函数：用户自定义聚合函数
- 特点：多对一，输入多个得到一个
- 比方：sum() avg() count() min()
UDTF函数：用户自定义表数据生成函数
- 特点：一对多，输入一个得到多个
- 比方：explode()

简单来说：SQL函数中的UDF、UDAF和UDTF就像是“数据处理的三剑客”，分别善于“单点突破”、“团队协作”和“多点开花”，满足不同的数据处理需求。
具体而言：
- UDF（用户自定义函数）：
  - 比喻：像是“单兵作战”，一对一处理数据。
  - 特点：输入一个值，输出一个值。
  - 示例：split()（字符串分割）、substr()（字符串截取）。
  - 实现：在Hive中继承UDF类，实现evaluate方法。
- UDAF（用户自定义聚合函数）：
  - 比喻：像是“团队协作”，多对一聚合数据。
  - 特点：输入多行数据，输出一个聚合结果。
  - 示例：sum()（求和）、avg()（平均值）、count()（计数）。
  - 实现：在Hive中继承UDAF类，实现init()、iterate()、merge()等方法。
- UDTF（用户自定义表生成函数）：
  - 比喻：像是“多点开花”，一对多展开数据。
  - 特点：输入一行数据，输出多行数据。
  - 示例：explode()（将数组或Map展开为多行）。
  - 实现：在Hive中继承GenericUDTF类，实现initialize()、process()等方法。
实际生产场景：
- 在数据清洗中，利用UDF对字符串进行格式化处理。
- 在数据分析中，利用UDAF计算复杂的业务指标，如加权平均值。
- 在数据展开中，利用UDTF将嵌套的JSON数据拆分为多行，便于后续分析。
总之：UDF、UDAF和UDTF是SQL中强大的扩展工具，分别适用于“一对一”、“多对一”和“一对多”的数据处理场景，为业务需求提供了机动的支持。

在SQL中提供的全部的内置函数，都是属于以上三类中某一类函数
思考：有这么多的内置函数，为啥还需要自定义函数呢?

为了扩充函数功能。在实际使用中，并不能保证所有的操作函数都已经提前的内置好了。很多基于业务处理的功能，其实并没有提供对应的函数，提供的函数更多是以公共功能函数。此时需要进行自定义，来扩充新的功能函数

复制代码

1- SparkSQL原生的时候，Python只能开发UDF函数
2- SparkSQL借助其他第三方组件，Python可以开发UDF、UDAF函数

复制代码

在Spark SQL中，针对Python语言，对于自定义函数，原生支持的并不是特别好。目前原生仅支持自定义UDF函数，而无法自定义UDAF函数和UDTF函数。
在1.6版本后，Java 和scala语言支持自定义UDAF函数，但Python并不支持。
Spark SQL原生存在的题目：大量的序列化和反序列

虽然Python支持自定义UDF函数，但是其效率并不是特别的高效。因为在使用的时候，传递一行处理一行，返回一行的方式。这样会带来非常大的序列化的开销的问题，导致原生UDF函数效率不好
早期解决方案: 基于Java/Scala来编写自定义UDF函数，然后基于python调用即可
目前主要的解决方案: 引入Arrow框架，可以基于内存来完成数据传输工作，可以大大的降低了序列化的开销，提供传输的效率，解决原生的问题。同时还可以基于pandas的自定义函数，利用pandas的函数优势完成各种处理操作

复制代码

3、Spark原生自定义UDF函数

自定义函数流程：

第一步: 在PySpark中创建一个Python的函数，在这个函数中书写自定义的功能逻辑代码即可
第二步: 将Python函数注册到Spark SQL中
注册方式一: udf对象 = sparkSession.udf.register(参数1,参数2,参数3)
参数1: 【UDF函数名称】，此名称用于后续在SQL中使用，可以任意取值，但是要符合名称的规范
参数2: 【自定义的Python函数】，表示将哪个Python的函数注册为Spark SQL的函数
参数3: 【UDF函数的返回值类型】。用于表示当前这个Python的函数返回的类型
udf对象: 返回值对象，是一个UDF对象，可以在DSL中使用
注意: 如果通过方式一来注册函数, 【可以用在SQL和DSL】
注册方式二: udf对象 = F.udf(参数1,参数2)
参数1: Python函数的名称，表示将那个Python的函数注册为Spark SQL的函数
参数2: 返回值的类型。用于表示当前这个Python的函数返回的类型
udf对象: 返回值对象，是一个UDF对象，可以在DSL中使用
注意: 如果通过方式二来注册函数，【仅能用在DSL中】
注册方式三: 语法糖写法 @F.udf(returnType=返回值类型) 放置到对应Python的函数上面
说明: 实际是方式二的扩展。如果通过方式三来注册函数，【仅能用在DSL中】
第三步: 在Spark SQL的 DSL/ SQL 中进行使用即可

复制代码

自定义演示一:
请自定义一个函数，完成对数据同一添加一个后缀名的操作

from pyspark import SparkConf, SparkContext
import os
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
# 绑定指定的Python解释器
from pyspark.sql.types import StringType
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("请自定义一个函数，完成对数据统一添加一个后缀名的操作_itheima")
# 1- 创建SparkSession对象
spark = SparkSession.builder\
.config("spark.sql.shuffle.partitions",1)\
.appName('sparksql_udf_basetype')\
.master('local[*]')\
.getOrCreate()
# 2- 数据输入
init_df = spark.createDataFrame(
data=[(1,'张三','广州'),(2,'李四','深圳')],
schema='id int,name string,address string'
)
init_df.printSchema()
init_df.show()
init_df.createTempView('tmp')
# 3- 数据处理
# 3.1- 创建自定义的Python函数
def add_suffix(address):
return address + "_itheima"
# 3.2- 将Python函数注册到Spark SQL
# 注册方式一
dsl_add_suffix = spark.udf.register('sql_add_suffix',add_suffix,StringType())
# 3.3- 在SQL/DSL中调用
# SQL
spark.sql("""
select
id,name,address,
sql_add_suffix(address) as new_address
from tmp
""").show()
# DSL
init_df.select(
"id",
"name",
"address",
dsl_add_suffix("address").alias("new_address")
).show()
print("-"*30)
# 在错误的地方调用了错误的函数。spark.udf.register参数1取的函数名只能在SQL中使用，不能在DSL中用。
# spark.sql("""
# select
# id,name,address,
# dsl_add_suffix(address) as new_address
# from tmp
# """).show()
# 注册方式二：UDF返回值类型传值方式一
dsl2_add_suffix = F.udf(add_suffix,StringType())
# DSL
init_df.select(
"id",
"name",
"address",
dsl2_add_suffix("address").alias("new_address")
).show()
# 注册方式二：UDF返回值类型传值方式二
dsl3_add_suffix = F.udf(add_suffix, 'string')
# DSL
init_df.select(
"id",
"name",
"address",
dsl3_add_suffix("address").alias("new_address")
).show()
# 注册方式三：语法糖/装饰器
@F.udf(returnType=StringType())
def add_suffix_candy(address):
return address + "_itheima"
# DSL
init_df.select(
"id",
"name",
"address",
add_suffix_candy("address").alias("new_address")
).show()
# 4- 数据输出
# 5- 释放资源
spark.stop()

复制代码

运行结果截图：

可能遇到的题目：

原因: 在错误的地方调用了错误的函数。spark.udf.register参数1取的函数名只能在SQL中使用，不能在DSL中用。

复制代码

自定义演示二:
自定义UDF函数，让其返回值类型为复杂类型: 字典列表元组

1- 返回列表和元组，表现一致
2- 返回字典。需要注意，字典中的key需要和schema中的字段名保持一致，否则获取不到值，以null填充

复制代码

from pyspark import SparkConf, SparkContext
import os
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
# 绑定指定的Python解释器
from pyspark.sql.types import StringType, StructType
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__':
# 1- 创建SparkSession对象
spark = SparkSession.builder\
.config("spark.sql.shuffle.partitions",1)\
.appName('sparksql_udf_compaxtype')\
.master('local[*]')\
.getOrCreate()
# 2- 数据输入
init_df = spark.createDataFrame(
data=[(1,'张三广州'),(2,'李四深圳')],
schema='id int,name_address string'
)
init_df.printSchema()
init_df.show()
init_df.createTempView('tmp')
# 3- 数据处理
# 3.1- 创建自定义的Python函数
def my_split(name_address):
data_list = name_address.split(' ')
# 返回列表
# return data_list
# 返回元组。与列表表现一致
# return tuple(data_list)
# 返回字典。需要注意，字典中的key需要和schema中的字段名保持一致，否则获取不到值，以null填充
# return {"n": data_list[0], "a": data_list[1]}
return {"name":data_list[0],"address":data_list[1]}
# 3.2- 将Python函数注册到Spark SQL
schema = StructType().add("name",StringType()).add("address",StringType())
dsl_my_split = spark.udf.register('sql_my_split',my_split,schema)
# 3.3- 在SQL/DSL中调用
# SQL方式
sql_result = spark.sql("""
select
id,name_address,
sql_my_split(name_address) as new_field,
sql_my_split(name_address)['name'] as name,
sql_my_split(name_address)['address'] as address
from tmp
""")
sql_result.printSchema()
sql_result.show()
# DSL方式
init_df.select(
"id",
"name_address",
dsl_my_split("name_address").alias("new_field"),
dsl_my_split("name_address")['name'].alias("name"),
dsl_my_split("name_address")['address'].alias("address")
).show()
# 4- 数据输出
# 5- 释放资源
spark.stop()

复制代码

运行结果截图：

01_数据清洗的dropDuplicates函数.py

dropDuplicates 行来看,保存第一个重复值,有subset也是

# 导包
import os
from pyspark.sql import SparkSession
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 注意: 如果操作sparksql直接用spark对象即可,如果还有操作rdd的操作,需要获取sc对象
sc = spark.sparkContext
# 先读取数据生成df对象
df = spark.read.csv(
path='file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/clear_data.csv',
sep=',',
header=True
)
# 测试是否有数据
print(df.count())
df.show()
print('-------------------------------------------------------------')
# 对df数据清洗
# dropDuplicates(): 删除重复数据,默认整行一样就删除
df1 = df.dropDuplicates(subset=['id','name'])
# 测试数据是否清洗
print(df1.count())
df1.show()
print('-------------------------------------------------------------')
# dropDuplicates(): 删除重复数据,默认整行一样就删除
# 参数subset: 指定删除哪几列数据重复,默认整行一样就删除
# 测试数据是否清洗
# 注意: 最后一定释放资源
sc.stop()
spark.stop()
### dropDuplicates ==行来看,保留第一个重复值,有subset也是==

复制代码

结果

02_数据清洗的dropna函数.py

dropna 行来看,how默认=any行中有空就删行,how='all’行中全部都是空才删行,subset=[‘name’]列中有空删行,thresh=3最少有三个正常值(非空值)

# 导包
import os
from pyspark.sql import SparkSession
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 注意: 如果操作sparksql直接用spark对象即可,如果还有操作rdd的操作,需要获取sc对象
sc = spark.sparkContext
# 先读取数据生成df对象
# 注意: 如果读取的文件首行是表头,可以使用header=True,把首行作为表头使用
df = spark.read.csv(
path='file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/clear_data.csv',
sep=',',
header=True
)
# 测试是否有数据
print(df.count())
df.show()
print('-------------------------------------------------------------')
# 对df数据清洗
# dropna() : 默认删除有空值的行
df1 = df.dropna()
# 测试是否数据被清洗
print(df1.count())
df1.show()
print('-------------------------------------------------------------')
# 对df数据清洗
# dropna() : 默认删除有空值的行
# how参数: how默认=any行中有空就删行,how='all'行中全部都是空才删行
df2 = df.dropna(how='all')
# 测试是否数据被清洗
print(df2.count())
df2.show()
print('-------------------------------------------------------------')
# 对df数据清洗
# dropna() : 默认删除有空值的行
# subset参数: 指定哪几列有空值就删除对应行
df3 = df.dropna(subset=['name'])
# 测试是否数据被清洗
print(df3.count())
df3.show()
print('-------------------------------------------------------------')
# 对df数据清洗
# dropna() : 默认删除有空值的行
# thresh参数: 表示至少有thresh个非空值才保留,默认是1
df3 = df.dropna(thresh=3)
# 测试是否数据被清洗
print(df3.count())
df3.show()
# 注意: 最后一定释放资源
sc.stop()
spark.stop()
### dropna ==行来看,how默认=any行中有空就删行,how='all'行中全部都是空才删行,subset=['name']列中有空删行,thresh=3最少有三个正常值(非空值)==

复制代码

结果

03_数据清洗的fillna函数.py

fillna 列来看,value用指定值添补空值

# 导包
import os
from pyspark.sql import SparkSession
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 注意: 如果操作sparksql直接用spark对象即可,如果还有操作rdd的操作,需要获取sc对象
sc = spark.sparkContext
# 读取文件生成df对象
df = spark.read.csv(
path="file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/clear_data.csv",
sep=',',
header=True
)
df.show()
print('--------------------------------------------------------------------------------')
# fillna(): 默认用指定值填充所有空值
# value: 制定单值
df2 = df.fillna(value='未知')
df2.show()
print('--------------------------------------------------------------------------------')
# value: 制定单值
# subset: 指定哪几列有空值就填充对应列
df3 = df.fillna(value='未知', subset=['name', 'age'])
df3.show()
print('--------------------------------------------------------------------------------')
# value: 制定字典(每个字段都能单独指定值)
# 通过字典方式填充空值,k就是列名,v就是要填充的值
df4 = df.fillna(value={'name': '未知', 'age': 0, 'address': '深圳'})
df4.show()
# 注意: 最后一定释放资源
sc.stop()
spark.stop()
### fillna ==列来看,value用指定值填充空值==

复制代码

结果

04_清洗后数据存储到文件.py

# 导包
import os
from pyspark.sql import SparkSession
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 注意: 如果操作sparksql直接用spark对象即可,如果还有操作rdd的操作,需要获取sc对象
sc = spark.sparkContext
# 读取原始未清洗文件生成df对象
df = spark.read.csv(
path="file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/clear_data.csv",
sep=',',
header=True
)
# 清洗df数据
etl_df = df.dropDuplicates().fillna('未知')
# 输出清洗后的数据方式1: 各种文件方式2: 数据库
# 先演示输出到文件
# 注意: 路径可以hdfs可以linux本地,分隔符也可以自定义
etl_df.write.csv(
path='hdfs://node1:8020/output',
sep='\001',
header=True,
mode='overwrite',
encoding='utf-8'
)
data = "John\00125\001Male\001New York"
fields = data.split('\001')
print(fields) # 输出：['John', '25', 'Male', 'New York']
# 注意: 最后一定释放资源
sc.stop()
spark.stop()

复制代码

结果

05_清洗后数据存储到mysql数据库.py

# 导包
import os
from pyspark.sql import SparkSession
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 注意: 如果操作sparksql直接用spark对象即可,如果还有操作rdd的操作,需要获取sc对象
sc = spark.sparkContext
# 读取原始未清洗文件生成df对象
df = spark.read.csv(
path="file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/clear_data.csv",
sep=',',
header=True
)
# 清洗df数据
etl_df = df.dropDuplicates().fillna('未知')
# 输出清洗后的数据方式1: 各种文件方式2: 数据库
# 再演示输出到数据库
# 注意: 输出到数据库,需要先在spark中配置jdbc驱动,否则会报错
etl_df.write.jdbc(
url="jdbc:mysql://node1:3306/数据库名字?useUnicode=true&characterEncoding=utf-8",
table="clear_data_etl_tb2",
mode="append",
properties={
"user": "root",
"password": "123456"
}
)
###############################################################################################
etl_df.write.jdbc(
url="jdbc:mysql://{host}:{port}/{db_name}?useUnicode=true&characterEncoding=utf-8",
table="表名", # 替换表名，例如 "clear_data_etl_tb2"
mode="模式", # 替换模式，例如 "append" 或 "overwrite"
properties={
"user": "用户", # 替换用户名，例如 "root"
"password": "密码" # 替换密码，例如 "123456"
}
)
################################################################################################
# 注意: 最后一定释放资源
sc.stop()
spark.stop()

复制代码

结果

07_清洗后数据存储到文件_优化分区数.py

# 导包
import os
import time
from pyspark.sql import SparkSession
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# TODO: 任务前: 修改saprkSQL的分区数
spark.conf.set("spark.sql.shuffle.partitions", '1')
# TODO: 获取SparkSQL的分区数
print(spark.conf.get("spark.sql.shuffle.partitions"))
# 读取原始未清洗文件生成df对象
df = spark.read.csv(
path="file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/clear_data.csv",
sep=',',
header=True
)
# 清洗df数据
etl_df = df.dropDuplicates().fillna('未知')
# 输出清洗后的数据方式1: 各种文件方式2: 数据库
# 先演示输出到文件
# 注意: path路径可以hdfs可以linux本地,sep分隔符也可以自定义
etl_df.write.csv(
path='hdfs://node1:8020/output1',
sep='\001',
header=True,
mode='overwrite',
encoding='utf-8'
)
# 为了查看分区效果,可以先让程序休息会儿
time.sleep(500)
# 注意: 最后一定释放资源
spark.stop()

复制代码

结果

08_电影数据分析案例.py

# 导包
import os
from pyspark.sql import SparkSession, functions as F
# 解决JAVA_HOME 未设置问题
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 get_user_avg_score():
"""
需求一: 统计每个用户的平均评分
"""
# 方式1: SQL风格
spark.sql("""
select user_id,round(avg(score),2) as avg_score
from movie_tb
group by user_id
""").show()
# 方式2: DSL风格
movie_df.groupby("user_id").agg(
F.round(F.avg("score"), 2).alias("avg_score")
).show()
def get_big_avg_score_movie_cnt():
"""
需求: 获取大于平均分电影数量
1- 统计所有打分的平均分，这个结果就是一个数字
2- 统计每部电影各自的平均分
3- 查询大于平均分的电影的数量
"""
# 方式1: SQL风格
spark.sql("""
with t as (
select movie_id,avg(score) from movie_tb group by movie_id
having avg(score) > (select avg(score) as all_movie_avg_score from movie_tb)
)
select count(1) as cnt from t
""").show()
# 方式2: DSL风格
movie_df.groupby("movie_id").agg(
F.avg("score").alias("movie_avg_score")
).where(
F.col("movie_avg_score") > movie_df.select(F.avg("score").alias("all_movie_avg_score")).first()[
'all_movie_avg_score']
).agg(
F.count('movie_avg_score').alias("cnt")
).show()
def get_big_score_top1_user_avg_score_sql():
"""
需求四: 查询高分电影中(电影平均分大于3)打分次数最多的用户, 并且求出此人所有的打分记录中, 打的平均分是多少
1- 筛选出高分电影。统计每部电影的平均分，再过滤出>3分的电影信息
2- 从高分电影中，统计每个用户分别打了多少次分。再选择TOP1的用户
3- 统计该用户所有打分记录的平均分
"""
# 1 - 筛选出高分电影。统计每部电影的平均分，再过滤出 > 3分的电影信息
spark.sql("""
select movie_id,avg(score) as movie_avg_score
from movie_tb
group by movie_id
having movie_avg_score > 3
""").createTempView("high_score_movie_tb")
# 2 - 从高分电影中，统计每个用户分别打了多少次分。再选择TOP1的用户(不考虑并列)
spark.sql("""
select user_id,count(*) as cnt
from high_score_movie_tb h
join movie_tb m on h.movie_id = m.movie_id
group by user_id
order by cnt desc
limit 1
""").createTempView("top1_user_tb")
# 3 - 统计该用户所有打分记录的平均分
spark.sql("""
select user_id,round(avg(score),2) as avg_score
from movie_tb where user_id = (select user_id from top1_user_tb)
group by user_id
""").show()
def get_big_score_top1_user_avg_score_dsl():
"""
需求四: 查询高分电影中(电影平均分大于3)打分次数最多的用户, 并且求出此人所有的打分记录中, 打的平均分是多少
1- 筛选出高分电影。统计每部电影的平均分，再过滤出>3分的电影信息
2- 从高分电影中，统计每个用户分别打了多少次分。再选择TOP1的用户
3- 统计该用户所有打分记录的平均分
"""
# 1 - 筛选出高分电影。统计每部电影的平均分，再过滤出 > 3分的电影信息
high_score_movie_df = movie_df.groupby("movie_id").agg(
F.avg("score").alias('movie_avg_score')
).where(
F.col("movie_avg_score") > 3
)
# 2 - 从高分电影中，统计每个用户分别打了多少次分。再选择TOP1的用户
top1_user_df = high_score_movie_df.join(movie_df, on="movie_id", how="inner").groupby("user_id").agg(
F.count("user_id").alias("cnt")
).orderBy("cnt", ascending=False).limit(1)
# 3 - 统计该用户所有打分记录的平均分
movie_df.where(
F.col("user_id") == top1_user_df.first()['user_id']
).groupby("user_id").agg(
F.round(F.avg("score"), 2).alias("avg_score")
).show()
if __name__ == '__main__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 1.数据收集
df = spark.read.csv(
path="file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/u.data",
sep="\t",
schema="user_id string, movie_id string, score int, dt string"
)
df.show()
df.printSchema()
print(df.count())
# 2.数据处理
movie_df = df.dropDuplicates().dropna()
print(movie_df.count())
# 3.数据分析
# 因为多个需求都会用到临时表,所以提前创建临时表
movie_df.createTempView("movie_tb")
# 需求一: 统计每个用户的平均评分
# get_user_avg_score()
# 需求三: 查询大于平均分的电影的数量
# get_big_avg_score_movie_cnt()
# 需求四: 查询高分电影中(电影平均分大于3)打分次数最多的用户, 并且求出此人所有的打分记录中, 打的平均分是多少
# get_big_score_top1_user_avg_score_sql()
# get_big_score_top1_user_avg_score_dsl()
# 4.数据可视化
# 可视化本次略
# 注意: 最后一定释放资源
spark.stop()

复制代码

09_sparkSQL中应用开窗函数.py

# 导包
import os
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.window import Window
# 解决JAVA_HOME 未设置问题
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__':
# 先创建spark session对象
spark = SparkSession.builder.appName("spark_demo").master("local[*]").getOrCreate()
# 注意: 如果操作sparksql直接用spark对象即可,如果还有操作rdd的操作,需要获取sc对象
sc = spark.sparkContext
# 数据收集
df = spark.read.csv(
path="file:///export/data/spark_project/06_spark_sql_数据清洗_导出以及综合案例/resources/cookie.txt",
sep=',',
schema="cid string,dt string,pv int"
)
df.show()
# 数据处理
etldf = df.dropDuplicates().fillna('未知')
# 数据分析
# 需求: 统计每个cookie的top3的pv值
# 方式1: SQL方式
etldf.createTempView("cookie_tb")
spark.sql("""
with t as (
select cid,dt,pv,
row_number() over(partition by cid order by pv desc) as rn
from cookie_tb
)
select * from t where rn <=3
""").show()
# 方式2: dsl方式
etldf.select(
"cid",
"dt",
"pv",
F.row_number().over(
Window.partitionBy("cid").orderBy(F.desc("pv"))
).alias("rn")
).where("rn <= 3").show()
# 数据可视化: 略
# 注意: 最后一定释放资源
sc.stop()
spark.stop()

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

用户名		自动登录	找回密码
密码			立即注册

day06_Spark SQL

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

day06_Spark SQL

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

登录参与点评抽奖加入IT实名职场社区