PySpark 如何按脚本执行日的上一自然月过滤Spark DataFrame

核心实现逻辑 #

不需要硬编码时间范围，核心是通过PySpark内置的日期函数动态计算脚本运行当日的上一自然月的首尾边界，再用范围匹配完成数据过滤，自动兼容大小月、闰年的日期差异。

完整实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 初始化Spark环境（实际业务中可直接复用已创建的Spark会话）
spark = SparkSession.builder.appName("FilterLastNaturalMonth").getOrCreate()

# 构造示例数据集
raw_data = [
    ("2020-12-31", 1),
    ("2020-11-02", 2),
    ("2020-10-01", 3),
    ("2021-02-01", 4),
    ("2021-01-05", 5),
    ("2021-02-10", 6),
    ("2021-02-11", 7),
    ("2021-02-26", 8),
    ("2021-03-01", 9)
]
df = spark.createDataFrame(raw_data, schema=["Column_A", "Column_B"])

# 1. 将字符串格式的日期列转换为Date类型，避免格式匹配错误
df = df.withColumn("Column_A", F.to_date(F.col("Column_A"), "yyyy-MM-dd"))

# 2. 动态计算上一个自然月的起止时间边界
run_date = F.current_date()  # 获取脚本执行时的系统当日日期
# 上一月起始：当月1号往前推1个月
last_month_start = F.date_trunc("month", F.add_months(run_date, -1))
# 上一月截止：当月1号往前减1天，即上一月最后一天
last_month_end = F.date_trunc("month", run_date) - F.expr("INTERVAL 1 DAY")

# 3. 按边界范围过滤数据
result_df = df.filter(
    (F.col("Column_A") >= last_month_start) &
    (F.col("Column_A") <= last_month_end)
)

result_df.show()

运行效果说明 #

当脚本在2021年3月1日执行时，代码计算得到的过滤边界为2021-02-01至2021-02-28，最终输出结果与预期完全匹配：

+----------+--------+
|  Column_A|Column_B|
+----------+--------+
|2021-02-01|       4|
|2021-02-10|       6|
|2021-02-11|       7|
|2021-02-26|       8|
+----------+--------+

注意事项 #

• 该逻辑无需手动维护硬编码时间，脚本在当月任意一天运行，都会自动锁定上一个完整自然月的范围
• 内置日期函数会自动适配闰年、大小月的天数差异，不会出现2月跨月匹配错误
• 如果业务中Column_A是带时分秒的Timestamp类型，可将上一月截止时间调整为last_month_end = F.date_trunc("month", run_date) - F.expr("INTERVAL 1 SECOND")，覆盖上一月最后一天23:59:59的所有记录，避免漏数。

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