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() 3- 第三类: 其他函数 ntile() first_value() last_value() lead() lag() 三个排序函数的区别? row_number(): 巧记 1234 特点: 唯一且连续 rank(): 巧记 1224 特点: 并列不连续 dense_rank(): 巧记 1223 特点: 并列且连续
在Spark SQL中使用窗口函数案例：
已知数据如下:
cookie1,2018-04-10,1 cookie1,2018-04-11,5 cookie1,2018-04-12,7 cookie1,2018-04-13,3 cookie1,2018-04-14,2 cookie1,2018-04-15,4 cookie1,2018-04-16,4 cookie2,2018-04-10,2 cookie2,2018-04-11,3 cookie2,2018-04-12,5 cookie2,2018-04-13,6 cookie2,2018-04-14,3 cookie2,2018-04-15,9 cookie2,2018-04-16,7
需求: 要求找出每个cookie中pv排在前3位的数据，也就是分组取TOPN问题
# 导包
import os
from pyspark.sql import SparkSession,functions as F,Window as W
# 绑定指定的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 = spark.read.csv(
path='file:///export/data/spark_project/spark_sql/data/cookie.txt',
sep=',',
schema='cookie string,datestr string,pv int'
)
# 3.数据处理(切分,转换,分组聚合)
# 4.数据输出
etldf = df.dropDuplicates().dropna()
# SQL方式
etldf.createTempView('cookie_logs')
spark.sql(
"""
select cookie,datestr,pv
from (
select cookie,datestr,pv,
dense_rank() over(partition by cookie order by pv desc) as rn
from cookie_logs
) temp where rn <=3
"""
).show()
# DSL方式
etldf.select(
'cookie', 'datestr', 'pv',
F.dense_rank().over( W.partitionBy('cookie').orderBy(F.desc('pv')) ).alias('rn')
).where('rn <=3').select('cookie', 'datestr', 'pv').show()
# 5.关闭资源
spark.stop()
运行结果截图：
2、自定义函数背景
2.1 回顾函数分类标准:
SQL函数，主要分为以下三大类：
-
UDF函数：普通函数
-
特点：一对一，输入一个得到一个
-
例如：split() ...
-
-
UDAF函数：聚合函数
-
特点：多对一，输入多个得到一个
-
例如：sum() avg() count() min() max() ...
-
-
UDTF函数：表生成函数
-
特点：一对多，输入一个得到多个
-
例如：explode() ...
-
在SQL中提供的所有的内置函数，都是属于以上三类中某一类函数
2.2 自定义函数背景
思考：有这么多的内置函数，为啥还需要自定义函数呢?
为了扩充函数功能。在实际使用中，并不能保证所有的操作函数都已经提前的内置好了。很多基于业务处理的功能，其实并没有提供对应的函数，提供的函数更多是以公共功能函数。此时需要进行自定义，来扩充新的功能函数
在Spark SQL中，针对Python语言，对于自定义函数，原生支持的并不是特别好。目前原生仅支持自定义UDF函数，而无法自定义UDAF函数和UDTF函数。
在1.6版本后，Java 和scala语言支持自定义UDAF函数，但Python并不支持。
1- SparkSQL原生的时候，Python只能开发UDF函数 2- SparkSQL借助其他第三方组件(Arrow,pandas...)，Python可以开发UDF、UDAF函数,同时也提升效率
Spark SQL原生UDF函数存在的问题：大量的序列化和反序列
虽然Python支持自定义UDF函数，但是其效率并不是特别的高效。因为在使用的时候，传递一行处理一行，返回一行的方式。这样会带来非常大的序列化的开销的问题，导致原生UDF函数效率不好 早期解决方案: 基于Java/Scala来编写自定义UDF函数，然后基于python调用即可 目前主要的解决方案: 引入Arrow框架，可以基于内存来完成数据传输工作，可以大大的降低了序列化的开销，提供传输的效率，解决原生的问题。同时还可以基于pandas的自定义函数，利用pandas的函数优势完成各种处理操作
