Spark内置函数:字符串、日期和时间函数、聚合函数、数值函数、条件判断函数、窗口函数

文章目录

1. 内置函数

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()