PySpark DataFrame中将周末日期转换为最近工作日的实现

没问题，我来帮你搞定这个需求！你的场景是要把DataFrame里的周末日期调整为最近的工作日，下面是具体的实现步骤和代码：

首先，我们得先把字符串格式的日期转换成PySpark能识别的日期类型，然后判断日期是否属于周末，最后根据需求调整到前一个或后一个工作日。

完整代码实现 #

from pyspark.sql.session import SparkSession
from pyspark.sql.functions import to_date, dayofweek, when, date_add, date_sub

# 初始化SparkSession
spark = SparkSession.builder.getOrCreate()

# 创建示例DataFrame
columns = ['Date', 'id', 'dogs', 'cats']
vals = [('04-05-2018',1, 2, 0), ('05-05-2018',2, 0, 1), ('06-05-2018',2, 0, 1)]
df = spark.createDataFrame(vals, columns)

# 第一步：将字符串日期转为日期类型（注意这里的格式是dd-MM-yyyy，根据你的实际格式调整）
df = df.withColumn("date_type", to_date(df["Date"], "dd-MM-yyyy"))

# 第二步：判断并调整周末日期
# 说明：PySpark的dayofweek函数中，周日=1，周六=7
# 方案1：周六转前一个工作日（周五），周日转后一个工作日（周一）
df_adjusted = df.withColumn(
    "adjusted_date",
    when(dayofweek("date_type") == 7, date_sub("date_type", 1))  # 周六减1天
    .when(dayofweek("date_type") == 1, date_add("date_type", 1))  # 周日加1天
    .otherwise("date_type")  # 工作日保持不变
)

# 如果你想统一调整为**前一个工作日**（比如周日也转周五），可以用这个逻辑：
# df_adjusted = df.withColumn(
#     "adjusted_date",
#     when(dayofweek("date_type") == 7, date_sub("date_type", 1))
#     .when(dayofweek("date_type") == 1, date_sub("date_type", 2))  # 周日减2天到周五
#     .otherwise("date_type")
# )

# 查看结果
df_adjusted.show()

代码解释 #

• to_date(df["Date"], "dd-MM-yyyy")：把字符串日期转换成PySpark的DateType，这里的格式参数要和你实际的日期格式匹配（如果你的日期是MM-dd-yyyy，就改成"MM-dd-yyyy"）。
• dayofweek：返回日期对应的星期数，方便我们识别周末。
• when...otherwise：条件判断函数，根据星期数对日期进行调整，date_add和date_sub分别用于增减天数。

执行结果 #

运行上面的代码后，示例中的日期会被调整：

• 04-05-2018（周五，工作日）保持不变
• 05-05-2018（周六）转为2018-05-04（周五）
• 06-05-2018（周日）转为2018-05-07（周一）

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