计算两日期间工作日返回Null值，求PySpark/Spark SQL优化方案

核心思路 #

放弃循环逐行处理的方式，利用Spark的分布式关联+过滤+聚合能力，这是适配Spark引擎的最优处理方案，既能解决Null值问题，又能大幅提升性能。

PySpark DataFrame API 实现 #

基础版（仅统计有工作日的Key） #

from pyspark.sql import functions as F

# 关联两个DataFrame，筛选出每个Key对应的日期区间内的记录
joined_df = dfDates.join(
    dfDimDate,
    (dfDimDate.DateKey >= dfDates.StartDateKey) & (dfDimDate.DateKey <= dfDates.EndDateKey)
)

# 按Key聚合，统计工作日数量
result_df = joined_df.filter(F.col("WorkDayFlag") == 1) \
    .groupBy("Key") \
    .agg(F.count("DateKey").alias("WorkDayCount"))

# 查看结果
result_df.show()

增强版（保留所有Key，无工作日时返回0） #

如果需要确保所有Key都出现在结果中（即使区间内无工作日），使用左关联+条件求和避免Null值：

from pyspark.sql import functions as F

# 左关联确保所有Key被保留
joined_df = dfDates.join(
    dfDimDate,
    (dfDimDate.DateKey >= dfDates.StartDateKey) & (dfDimDate.DateKey <= dfDates.EndDateKey),
    how="left"
)

# 按Key聚合，用条件求和统计工作日，无工作日则返回0
result_df = joined_df.groupBy("Key") \
    .agg(
        F.sum(F.when(F.col("WorkDayFlag") == 1, 1).otherwise(0))
        .alias("WorkDayCount")
    )

# 查看结果
result_df.show()

Spark SQL 实现 #

先将DataFrame注册为临时视图，再通过SQL语句完成统计：

-- 注册临时视图
CREATE OR REPLACE TEMP VIEW df_dates AS SELECT * FROM dfDates;
CREATE OR REPLACE TEMP VIEW df_dim_date AS SELECT * FROM dfDimDate;

-- 计算每个Key对应的工作日总数
SELECT
    d.Key,
    SUM(CASE WHEN dd.WorkDayFlag = 1 THEN 1 ELSE 0 END) AS WorkDayCount
FROM df_dates d
LEFT JOIN df_dim_date dd
    ON dd.DateKey BETWEEN d.StartDateKey AND d.EndDateKey
GROUP BY d.Key;

性能说明 #

原循环方案慢的原因是逐行单节点处理，完全没有利用Spark的分布式计算能力。而关联+聚合的方式会将任务拆分到集群节点并行执行，能大幅缩短运行时间。

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