基于条件向PySpark DataFrame新增行：按日期差复制对应数量的行

核心思路 #

利用PySpark内置的sequence函数生成日期偏移序列，再通过explode函数将序列炸开为多行，最后基于偏移量生成逐行的日期值即可完成需求。

前置依赖 #

PySpark 2.4及以上版本（内置sequence函数支持），低版本可使用UDF替代方案。

完整实现代码 #

步骤1：导入依赖包 #

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

步骤2：数据预处理（可选） #

如果你的日期字段是字符串类型，先转为DateType：

# 假设你的日期字段名分别为start_date、end_date
df = df.withColumn("start_date", F.to_date(F.col("start_date"))) \
       .withColumn("end_date", F.to_date(F.col("end_date")))

步骤3：生成偏移序列并炸开 #

# 1. 计算日期间差，生成0到间隔天数的偏移序列
df_with_offset = df.withColumn("day_diff", F.datediff(F.col("end_date"), F.col("start_date"))) \
                   .withColumn("offset_list", F.sequence(F.lit(0), F.col("day_diff")))

# 2. 炸开偏移序列为多行
df_exploded = df_with_offset.withColumn("offset", F.explode(F.col("offset_list")))

# 3. 生成每行对应的日期，删除中间字段
result_df = df_exploded.withColumn("date", F.date_add(F.col("start_date"), F.col("offset"))) \
                       .drop("day_diff", "offset_list", "offset")

低版本PySpark兼容方案（无sequence函数） #

用自定义UDF生成偏移序列：

from pyspark.sql.types import ArrayType, IntegerType

# 定义生成偏移列表的UDF
gen_offset_udf = F.udf(lambda diff: list(range(diff + 1)) if diff >=0 else [], ArrayType(IntegerType()))

# 替换sequence生成逻辑即可，其余步骤和上面一致
df_with_offset = df.withColumn("day_diff", F.datediff(F.col("end_date"), F.col("start_date"))) \
                   .withColumn("offset_list", gen_offset_udf(F.col("day_diff")))

注意事项 #

• 提前处理end_date < start_date的异常数据，避免生成空序列或者报错
• 若日期间隔超过1年以上，会生成大量行，建议提前评估数据量避免OOM
• 生成的date字段可根据需求重命名，原DataFrame的其他字段会完整保留在每一行

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