在PySpark中基于另一DataFrame拆分日期范围至指定周编号

步骤说明 #

要实现将df1的日期区间记录拆分对应到df2的周编号，核心是通过日期范围关联找到每条记录覆盖的所有周，再拆分生成多条记录。

代码实现 #

假设你的Spark环境已初始化，先确保日期字段类型正确，再执行以下步骤：

1. 统一日期类型
   把df1的BEGIN、END和df2的start_of_week转换为DateType（如果字段还不是日期类型的话）：

   from pyspark.sql import functions as F
   from pyspark.sql.types import DateType
   
   df1 = df1.withColumn("BEGIN", F.col("BEGIN").cast(DateType())) \
            .withColumn("END", F.col("END").cast(DateType()))
   
   df2 = df2.withColumn("start_of_week", F.col("start_of_week").cast(DateType()))
   

2. 计算周结束日期
   df2仅提供周起始日期，需计算对应周的结束日期（这里默认周为周一到周日，即起始日加6天；若你的业务是周日到周六，可调整为加5天）：

   df2_with_end = df2.withColumn("end_of_week", F.date_add(F.col("start_of_week"), 6))
   

3. 关联并筛选覆盖的周
   通过范围关联，匹配df1记录日期区间与df2周区间有重叠的所有周，筛选出有效关联结果：

   joined_df = df1.join(
       df2_with_end,
       (df1.BEGIN <= df2_with_end.end_of_week) & (df1.END >= df2_with_end.start_of_week),
       how="inner"
   )
   

4. 整理最终结果
   保留业务需要的字段，去掉原有的BEGIN、END，替换为week_no（若需要保留周的起止日期，可一并加入字段列表）：

   final_df = joined_df.select(
       "NAME", "X_NAME", "week_no", "A", "B", "C", "D", "E"
       # 可选：如需周起止日期，添加 "start_of_week", "end_of_week"
   )
   

示例验证 #

针对你提到的日期范围2021-12-07至2021-12-14的记录：

• 若df2中2021W49的start_of_week为2021-12-06、end_of_week为2021-12-12，该周与原区间重叠；
• 2021W50的start_of_week为2021-12-13、end_of_week为2021-12-19，同样与原区间重叠；
• 关联后会生成两条记录，分别对应2021W49和2021W50，完全符合需求。

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