用户可以根据需求自己封装计算的逻辑，对字段数据进行计算
内置函数，是spark提供的对字段操作的方法 ，split(字段) 对字段中的数进行切割，F.sum(字段) 会将该字段下的数据进行求和
实际业务中又能内置函数不满足计算需求，此时就需要自定义行数，完成字段数据的业务处
函数分类
- udf
- 自定义
- 一进一出
- udaf
- 聚合
- 自定义
- 多进一出
- udtf
- 爆炸
- 一进多出
UDF函数
对每一行数据以此进行计算，返回每一行的结果
1）不带装饰器
# UDF函数
from pyspark.sql import SparkSession,functions as F
from pyspark.sql.types import *ss = SparkSession.builder.getOrCreate()# 读取文件数据转为df
df = ss.read.csv('hdfs://node1:8020/data/students.csv',header=True,sep=',')df.show()# 自定义字符串长度计算函数
# @F.udf(returnType=IntegerType()) # 使用装饰器注册函数，只能在DSL方法中使用，不能在SQL中使用
def len_func(field):"""自定义函数，函数名可以自己指定:param field: 是用来结构处理的字段数据，可以定义多个。根据实际处理的字段数量决定定义多少个接收参数:return: 返回处理后的数据"""if field is None:return 0else:data = len(field)return data# 将自定义的函数注册到spark中使用
# 参数一 指定spark中使用函数的名
# 参数二 指定自定义函数的名
# 参数三 指定函数的返回值类型
# 接收参数 定义和函数名一样的值
len_func = ss.udf.register('len_func',len_func,returnType = IntegerType())# 在spark中使用
df2 = df.select('id','name','gender',len_func('name'))
df2.show()# 使用sql语句
df.createTempView('stu')df3 = ss.sql('select * ,len_FUNC(name) from stu')
df3.show()
2）带有装饰器
# UDF函数
from pyspark.sql import SparkSession,functions as F
from pyspark.sql.types import *ss = SparkSession.builder.getOrCreate()# 读取文件数据转为df
df = ss.read.csv('hdfs://node1:8020/data/students.csv',header=True,sep=',')df.show()# 自定义字符串长度计算函数
@F.udf(returnType=IntegerType()) # 使用装饰器注册函数，只能在DSL方法中使用，不能在SQL中使用
def len_func(field):"""自定义函数，函数名可以自己指定:param field: 是用来结构处理的字段数据，可以定义多个。根据实际处理的字段数量决定定义多少个接收参数:return: 返回处理后的数据"""if field is None:return 0else:data = len(field)return data# 在spark中使用
df2 = df.select('id','name','gender',len_func('name'))
df2.show()装饰器注册
-
只能在DSL方法中使用，在sql语句中无法使用
UDAF函数
多进一出 主要是聚合
使用pandas中的series实现，可以读取一列数据存储在pandas的seriess中进行数据的聚合
# 读取文件数据转为df
df = ss.read.csv('hdfs://node1:8020/data/students.csv',header=True,sep=',',schema='id int,name string,gender string,age int,cls string')df.show()# 自定义udaf函数
# 装饰器注册
@F.pandas_udf(returnType=IntegerType())
# 自定义udaf函数
# fileds:pd.Series 给数据字段指定一个类型
# -> float 指定返回值类型
# udaf函数注册需要两步，第一步现指定装饰器
def sub(filed:pd.Series) -> int:"""自定义udaf函数，实现累减:param field: 接收的字段列数据 pd.Series声明字段数据的类型,接收一列数据可以使用pandas的series类型:return:"""# field是series，就按照series方式操作n = filed[0] # 取出第一个值作为初始值for i in filed[1::]:n-=ireturn n# regidter方法注册
sub = ss.udf.register('sub',sub)# 使用udaf函数 缺少 PyArrow pandas中series类型交个spark程序无法识别，spark是有scala实现，scala中没有对应的series类型
# 可以使用 PyArrow框架将series转为scale能识别的数据类型
df2 = df.select(sub('age'))
df2.show()
- arrow框架 pyarrow
-
Apache Arrow 是一种内存中的列式数据格式，用于Spark中，以在JVM和Python进程之间有效地传输数据。目前这对使用 Pandas/NumPy 数据的 Python 用户最有益，提升传输速度。
-
在线安装 三台机器安装
-
进入虚拟环境 conda activate base
-
在线安装 pip install pyspark[sql] -i Verifying - USTC Mirrors
-
-
离线安装 三台机器安装
-
pip install pyarrow-10.0.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl
-
-
安装pyarrow
conda activate base
pip install pyspark[sql] -i https://pypi.mirrors.ustc.edu.cn/simple/