如何获取当前日期后指定周数内所有周日的PySpark DataFrame

要高效生成指定范围内的所有周日，应该利用Spark的批量处理能力，避免在Python中循环构造Column表达式。以下是具体实现：

步骤说明 #

1. 计算第一个目标周日：如果当前日期已是周日，直接使用当前日期；否则取当前日期之后的第一个周日。
2. 生成周偏移序列：创建0到MAX_WEEKS-1的整数序列，代表从第一个周日开始往后的周数偏移。
3. 计算所有周日：给第一个周日加上偏移量×7天，得到连续的周日日期。

完整代码 #

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

# 初始化SparkSession（如果未初始化）
spark = SparkSession.builder.appName("GenerateSundays").getOrCreate()

MAX_WEEKS = 5

# 计算第一个目标周日（处理当前日期已是周日的情况）
first_sunday = f.when(
    f.date_format(f.current_date(), "EEEE") == "Sunday",
    f.current_date()
).otherwise(
    f.next_day(f.current_date(), "Sunday")
)

# 生成0到MAX_WEEKS-1的偏移量序列
offset_df = spark.range(MAX_WEEKS).withColumnRenamed("id", "offset")

# 计算所有周日并构造结果DataFrame
sundays_df = offset_df.withColumn(
    "Date",
    f.date_add(first_sunday, f.col("offset") * 7)
).select("Date")

# 查看结果
sundays_df.show()

代码解释 #

• spark.range(MAX_WEEKS)：生成包含0到MAX_WEEKS-1的整数序列，对应需要生成的周日数量。
• date_add(first_sunday, offset*7)：通过偏移量计算每个后续周日，确保日期间隔为一周。
• 边界处理：用when/otherwise判断当前日期是否为周日，避免跳过当天（如果需要包含的话）。

原代码问题分析 #

你之前的循环代码只是不断嵌套next_day的Column表达式，并没有实际执行日期计算。Spark的Column对象是延迟计算的，这种循环方式会生成复杂且冗余的表达式，既不高效也不符合Spark的批量处理模式。

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