如何在PySpark中基于当前日期过滤前一日的数据

核心思路 #

先获取执行查询当天的前一天日期，再将DataFrame中的Date列转换为匹配的日期类型后进行等值过滤。

代码实现 #

根据Date列的实际数据类型，选择对应处理方式：

1. 当Date列已是date类型

from pyspark.sql import functions as F

# 计算执行当天的前一天日期
yesterday = F.date_sub(F.current_date(), 1)

# 过滤出Date列等于前一天的数据
filtered_df = df.filter(F.col('Date') == yesterday)

2. 当Date列是timestamp类型（含时分秒）

需要先将时间戳转为日期类型，再进行过滤：

from pyspark.sql import functions as F

yesterday = F.date_sub(F.current_date(), 1)

filtered_df = df.filter(F.to_date(F.col('Date')) == yesterday)

3. 当Date列是字符串类型

先按实际格式将字符串转为日期类型，再过滤：

from pyspark.sql import functions as F

yesterday = F.date_sub(F.current_date(), 1)

# 假设字符串格式为'yyyy-MM-dd'，需根据实际格式调整第二个参数
filtered_df = df.filter(F.to_date(F.col('Date'), 'yyyy-MM-dd') == yesterday)

时区注意事项 #

current_date()默认使用Spark的时区设置，如果数据时区与Spark时区不一致，需先统一时区：

# 示例：指定时区为Asia/Shanghai
yesterday = F.date_sub(F.to_date(F.current_timestamp().tz_convert("Asia/Shanghai")), 1)
filtered_df = df.filter(F.to_date(F.col('Date').tz_convert("Asia/Shanghai")) == yesterday)

内容的提问来源于stack exchange，提问作者n179911a