Spark高级用法-内置函数
目录
读取数据
# 内置数据集
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')
# print(df.show())
# 对字符串数据使用内置数据集进行处理
# 拼接
df_concat = df.select(df.id,df.name,df.gender,F.concat('name','gender').alias('fileds'),F.concat_ws(':','name','gender').alias('fileds2'))
df_concat.show()
1.字符串
1）拼接
df_concat = df.select(df.id,df.name,df.gender,F.concat('name','gender').alias('fileds'),F.concat_ws(':','name','gender').alias('fileds2'))
df_concat.show()
2）截取
df_substr = df.select(df.name,F.substring('name',1,2))
df_substr.show()
3）切割
df_split = df_concat.select(df_concat.fileds2,F.split('fileds2',":"))
df_split.show()
4）切割后取数据
df_split2 = df_concat.select(df_concat.fileds2,F.split('fileds2',":")[1])
df_split2.show() 
5）字符串替换
df_replace = df.select(df.name,F.regexp_replace('name','张','A'))
df_replace.show()
6）聚合函数
df_agg = df.groupby('gender').agg(F.sum('age').alias('sum'),F.avg('age').alias('avg'))
df_agg.show()
2.数值类
1）确定小数点位数
df_round = df_agg.select(df_agg.gender,df_agg.avg,F.round('avg',2))
df_round.show()
2）向上取值
df_ceil = df_agg.select(df_agg.gender,df_agg.avg,F.ceil('avg'))
df_ceil.show()
3）向下取值
df_floor = df_agg.select(df_agg.gender,df_agg.avg,F.floor('avg'))
df_floor.show()
4）从指定字段中取当前行最大的一个值
df_greatest = df_agg.select(df_agg.gender,df_agg.sum,df_agg.avg,F.greatest('sum','avg'))
df_greatest.show()
3.时间类型
1）获取当前的日期时间和unix时间
df_time = df.select(df.id,df.name,F.current_date().alias('dt'),F.current_timestamp().alias('tm'),F.unix_timestamp().alias('un'))
df_time.show()
2）将日期转为时间戳
df_unix = df_time.select(df_time.dt,F.unix_timestamp('dt'))
df_unix.show()
3）将时间戳转为日期
df_unix_time = df_time.select(df_time.un,F.from_unixtime('un','yyyy/MM/dd HH:mm:ss'))
df_unix_time.show()
4）日期的加减
df_add_dt = df_time.select(df_time.dt,F.date_add('dt',3))
df_add_dt.show()
df_add_dt = df_time.select(df_time.dt,F.date_add('dt',-3))
df_add_dt.show()
5）日期比较
df_diff_dt = df_time.select(df_time.dt,F.datediff('dt',F.date_add('dt',3)))
df_diff_dt.show()
6）日期取值
df_value_dt = df_time.select(df_time.tm,F.year('tm'),F.month('tm'),F.substring('tm',9,2),F.hour('tm'),F.second('tm'))
df_value_dt.show()
4.条件判断
1）实现if 效果判断
df_when = df.select(df.name,df.gender,F.when(df.gender == '女' , 1).otherwise(2))
df_when.show()
2）实现case when 效果判断
df_case_when = df.select(df.name,df.age,F.when(df.age > 30 , '中年').when((df.age >=18) & (df.age <= 30) , '青年').otherwise('青少年').alias('年龄层'))
df_case_when.show()
5.窗口函数
from pyspark.sql.window import Window
# 1-创建窗口
w = Window.partitionBy('gender').orderBy('age')
# 2-使用窗口函数
df_windows = df.select(df.id,df.name,df.gender,df.age,F.rank().over(w).alias('rank'))
df_windows.show()
原文地址：https://blog.csdn.net/weixin_58305115/article/details/142899338
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！