PySpark实现：从周数字段提取每周首个周一日期（避免UDF）

给定如下Spark DataFrame：

from pyspark.sql.types import StructType,StructField, StringType, IntegerType
data2 = [("James",202245),
    ("Michael",202133),
    ("Robert",202152),
    ("Maria",202252),
    ("Jen",202201)
  ]

schema = StructType([ \
    StructField("firstname",StringType(),True), \
    StructField("Week", IntegerType(), True) \
  ])
 
df = spark.createDataFrame(data=data2,schema=schema)
df.printSchema()
df.show(truncate=False)

其中Week字段格式为年份+周数，例如202245代表2022年第45周，需要提取该周对应的周一日期（如2022年第45周的周一为2022年11月7日）。

已通过Python的datetime结合UDF实现需求，代码如下：

def get_monday_from_week(x: int) -> datetime.date:
    """
    Converts fiscal week to datetime of first Monday from that week

    Args:
        x (int): fiscal week


    Returns:
        datetime.date: datetime of first Monday from that week
    """
    x = str(x)
    r = datetime.datetime.strptime(x + "-1", "%Y%W-%w")
    return r

现希望使用Spark内置函数实现该功能，避免使用UDF。

方法一：利用to_date直接解析格式 #

Spark的to_date函数支持按指定格式解析日期字符串，我们可以构造年份周数-1的格式（其中1代表周一），使用yyyyww-u格式符直接解析出目标周一日期，代码简洁高效：

from pyspark.sql import functions as F

df_result = df.withColumn("week_date_str", F.concat(F.col("Week").cast("string"), F.lit("-1"))) \
    .withColumn("target_monday", F.to_date("week_date_str", "yyyyww-u")) \
    .select("firstname", "Week", "target_monday")

df_result.show(truncate=False)

格式说明：yyyyww-u中，yyyy匹配4位年份，ww匹配2位周数，u代表一周中的第几天（1=周一，7=周日），因此yyyyww-1会被解析为对应年份第ww周的周一。

方法二：分步计算（适配特殊周定义） #

如果需要适配自定义的周规则（比如不同的周起始日或第1周定义），可以通过拆分年份周数、计算当年基准周一再偏移的方式实现：

from pyspark.sql import functions as F

df_result = df.withColumn("week_str", F.col("Week").cast("string")) \
    # 拆分年份和周数
    .withColumn("year", F.substring("week_str", 1, 4).cast("int")) \
    .withColumn("week_num", F.substring("week_str", 5, 2).cast("int")) \
    # 生成当年1月1日
    .withColumn("year_start", F.make_date("year", 1, 1)) \
    # 找到当年第一个周一
    .withColumn("first_monday", F.next_day(F.col("year_start"), "Monday")) \
    # 计算周偏移量：适配1月1日所在周是否为当年第1周的情况
    .withColumn("week_of_year_start", F.weekofyear("year_start")) \
    .withColumn("offset_weeks", 
                F.when(F.col("week_of_year_start") == 1, F.col("week_num") - 1)
                .otherwise(F.col("week_num"))) \
    # 偏移得到目标周周一
    .withColumn("target_monday", F.date_add("first_monday", F.col("offset_weeks") * 7)) \
    .select("firstname", "Week", "target_monday")

df_result.show(truncate=False)

内容的提问来源于stack exchange，提问作者Tom J Muthirenthi