1.1 字符串
concat(col1,col2,......):连接多个字符串列。
例如：concat(‘Hello’, ’ ‘, ‘World’)将返回’Hello World’。
substring(col,start,length):从给定字符串列中提取子字符串。
例如：substring(‘HelloWorld’, 6, 5) 的结果是 ‘World’。
trim(col):去除字符串两端空格。
若有字符串列 str_col，使用 trim(str_col) 可以去除其中每个字符串的前后空格。
# 内置函数使用
# 导入内置函数模块
from pyspark.sql import SparkSession,functions as F

ss = SparkSession.builder.getOrCreate()

# 读取文件数据转为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')

# 对字符串数据使用内置函数进行处理
# 拼接 F.concat('name','gender')
df_concat = df.select(df.id,df.name,df.gender,F.concat('name','gender').alias('field1'),F.concat_ws(':','name','gender').alias('field2'))
df_concat.show()
# 截取
df_substr = df.select(df.name,F.substring('name',1,2))
df_substr.show()
# 切割
df_split = df_concat.select(df_concat.field2,F.split('field2',':'))
df_split.show()
# 切割后的数据取值
df_split2 = df_concat.select(df_concat.field2,F.split('field2',':')[0])
df_split2.show()
# 字符串替换
df_replace = df.select(df.name,F.regexp_replace('name','王','a'))
df_replace.show()


1.2 日期和时间函数
current_date()：返回当前日期。
无需参数，直接调用可获取当前系统日期。
current_timestamp()：返回当前时间戳。
调用后得到当前的日期和时间。
year(col)、month(col)、day(col)：分别提取日期列中的年、月、日部分。
# 读取文件数据转为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')


# 日期时间处理函数
#获取当前的日期时间及unix时间（时间戳）
#将日期转为时间戳
df_dt = df.select(df.id,df.name,F.current_date().alias('dt'),F.current_timestamp().alias('tm'),F.unix_timestamp().alias('un'))
df_dt.show()

# 时间戳和日期之间的转化
#将时间戳转为日期
df_unix = df_dt.select(df_dt.dt,F.unix_timestamp('dt'))
df_unix.show()

# 日期加减
df_add_dt = df_dt.select(df_dt.dt,F.date_add('dt',-5))
df_add_dt.show()

#日期比较 比较相差多少
df_diff_dt= df_dt.select(df_dt.dt,F.datediff('dt',F.date_add('dt',-5)))
df_diff_dt.show()

#日期取值
df_value_dt = df_dt.select(df_dt.tm,F.year('tm'),F.month('tm'),F.substring('tm',9,2),F.hour('tm'),F.second('tm'))
df_value_dt.show()


1.3 聚合函数
count(col)：计算非空值的数量。
例如：SELECT count(*) FROM table_name; 计算表中的行数。
sum(col)：计算给定数值列的总和。
如：对数值列 amount 使用 SELECT sum(amount) FROM sales_table;。
avg(col)：计算给定数值列的平均值。
例如：SELECT avg(salary) FROM employees_table; 计算员工平均工资。
# 内置函数使用
# 导入内置函数模块
from pyspark.sql import SparkSession,functions as F

ss = SparkSession.builder.getOrCreate()

# 读取文件数据转为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_agg = df.groupby('gender').agg(F.sum('age').alias('sum_data'),F.avg('age').alias('avg_data'))
df_agg.show()

运行结果：



1.4 数值函数
abs(col)：返回给定列的绝对值。
例如，对于数值列 num_col，使用abs(num_col)可以得到该列每个值的绝对值。
ceil(col)：返回大于或等于给定列值的最小整数。
比如，ceil(4.2) 的结果是 5。
floor(col)：返回小于或等于给定列值的最大整数。
例如，floor(4.8) 的结果是 4。
# 读取文件数据转为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_agg = df.groupby('gender').agg(F.sum('age').alias('sum_data'),F.avg('age').alias('avg_data'))
df_agg.show()

# 数值处理函数
# 指定小数位数
df_round = df_agg.select(df_agg.gender,df_agg.avg_data,F.round('avg_data',2),F.round('avg_data'))
df_round.show()
# 向下取整数
df_floor = df_agg.select(df_agg.gender,df_agg.avg_data,F.floor('avg_data'))
df_floor.show()
# 向下取整数
df_ceil = df_agg.select(df_agg.gender,df_agg.avg_data,F.ceil('avg_data'))
df_ceil.show()
# 从指定的字段中取出当前行的最大的一个值
df_greatest = df_agg.select(df_agg.gender,df_agg.sum_data,df_agg.avg_data,F.greatest('sum_data','avg_data'))
df_greatest.show()


运行结果：



1.5 条件判断函数
case when... then... else... end：类似于 SQL 中的 CASE WHEN 语句，根据不同条件返回不同的值。

例如，case when col1 > 10 then 'High' else 'Low' end 根据 col1 的值返回不同的字符串结果。
coalesce(col1, col2,...)：返回第一个非空值。

如果 col1 为 null，则返回 col2 的值，以此类推。
# 内置函数使用
# 导入内置函数模块
from pyspark.sql import SparkSession,functions as F

ss = SparkSession.builder.getOrCreate()

# 读取文件数据转为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_when = df.select(df.name,df.gender,F.when(df.gender == '男',1).otherwise(2))
df_when.show()

#实现case when 判断不同年龄 返回请少年青年中年
df_case_when = df.select(df.name,df.age,F.when(df.age>30,'中年').when((df.age>20) & (df.age<=30),'青年').when((df.age>15) & (df.age<=20),'少年'))
df_case_when.show()


运行结果：




1.6 窗口函数
聚合窗口函数
这些函数对窗口内的行进行聚合操作，类似于常规的聚合函数，但在窗口范围内应用。例如 SUM、AVG、MAX、MIN 等。
示例：计算每个部门员工的工资总和，并按照部门进行分组。可以使用 SUM 窗口函数对工资列在部门窗口内进行求和。
排名窗口函数
RANK：为分区中的每一行分配一个排名，排名可能不连续，如果有并列的情况，会占用下一个名次的位置。
DENSE_RANK：也为分区中的每一行分配一个排名，但排名是连续的，不会出现排名间断的情况。
ROW_NUMBER：为分区中的每一行分配一个唯一的连续整数序号。
示例：对学生的考试成绩进行排名，可以使用 RANK 或 DENSE_RANK 函数按照成绩从高到低进行排名。
分析窗口函数
LEAD 和 LAG：可以访问当前行之前或之后的行的值。LEAD 函数获取当前行之后指定偏移量的行的值，LAG 函数获取当前行之前指定偏移量的行的值。
FIRST_VALUE 和 LAST_VALUE：分别返回窗口中第一行和最后一行的值。
示例：分析股票价格的变化趋势，可以使用 LAG 函数获取前一天的股票价格，然后计算价格的涨跌情况。
# 内置函数使用
# 导入内置函数模块
from pyspark.sql import SparkSession,functions as F

ss = SparkSession.builder.getOrCreate()

# 读取文件数据转为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')

#窗口函数 排序生成序号
from pyspark.sql.window import Window

#1-创建窗口
w = Window.partitionBy('gender').orderBy('age')

#2- 使用窗口函数
df_window = df.select(df.id,df.name,df.age,df.gender,F.rank().over(w).alias('rn'))
df_window.show()

————————————————

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

原文链接：https://blog.csdn.net/m0_70882914/article/details/142939580

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！
《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu
《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1
想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs
同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack