PySpark：生成最大与最小日期间的缺失日期（不含首尾）

步骤1：准备示例数据 #

先创建带日期列的示例DataFrame，方便演示逻辑：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, min, max, sequence, explode, to_date, date_add, lit

spark = SparkSession.builder.appName("MissingDates").getOrCreate()

data = [("2024-01-01",), ("2024-01-03",), ("2024-01-05",), ("2024-01-01",)]
df = spark.createDataFrame(data, ["date_col"]).withColumn("date_col", to_date(col("date_col")))

步骤2：计算日期范围并生成完整序列 #

先获取去重后日期列的最小、最大值，再生成该范围内的所有日期：

# 提取去重后的最小、最大日期
date_bounds = df.select(
    min("date_col").alias("min_date"),
    max("date_col").alias("max_date")
).first()

min_d = date_bounds["min_date"]
max_d = date_bounds["max_date"]

# 生成从最小到最大日期的完整序列
full_dates = spark.range(0, (max_d - min_d).days + 1) \
    .withColumn("full_date", date_add(lit(min_d), col("id"))) \
    .select("full_date")

步骤3：筛选缺失日期并排除两端 #

用左反连接找出不在原去重日期中的记录，再过滤掉最小和最大日期：

# 获取原数据去重后的日期
distinct_dates = df.select("date_col").distinct()

# 筛选缺失日期，同时排除最小、最大日期
missing_dates = full_dates.join(distinct_dates, full_dates.full_date == distinct_dates.date_col, "left_anti") \
    .filter((col("full_date") != min_d) & (col("full_date") != max_d)) \
    .withColumnRenamed("full_date", "missing_date")

missing_dates.show()

简化链式写法 #

如果偏好更紧凑的代码，可以合并为单链调用：

date_stats = df.select(min("date_col").alias("min_d"), max("date_col").alias("max_d")).first()

result = spark.range(0, (date_stats["max_d"] - date_stats["min_d"]).days + 1) \
    .withColumn("missing_date", date_add(lit(date_stats["min_d"]), col("id"))) \
    .join(df.select("date_col").distinct(), on="missing_date", how="left_anti") \
    .filter((col("missing_date") != date_stats["min_d"]) & (col("missing_date") != date_stats["max_d"]))

result.show()

关键说明 #

• left_anti连接直接筛选出序列中不存在于原去重日期的记录，比左连接后过滤null更高效
• 使用date_add结合range生成日期序列，避免了SQL字符串拼接的安全风险

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