PySpark SQL关联大表并按时间条件聚合多列的优化方法

基础实现逻辑 #

核心需求是对表1的每行(id_product, id_customer, start_date)，统计同产品同客户下所有stop_date < start_date的各数值字段总和，基础SQL实现如下：

SELECT
  t1.id_product,
  t1.id_customer,
  t1.start_date,
  SUM(COALESCE(t2.duration, 0)) AS sum_duration,
  SUM(COALESCE(t2.col1, 0)) AS sum_col1,
  -- 剩余19个求和字段以此类推
  SUM(COALESCE(t2.col20, 0)) AS sum_col20
FROM table1 t1
LEFT JOIN table2 t2
  ON t1.id_product = t2.id_product
  AND t1.id_customer = t2.id_customer
  AND t2.stop_date < t1.start_date
GROUP BY t1.id_product, t1.id_customer, t1.start_date

大体积数据优化方案 #

1. 提前裁剪数据 #

• 字段裁剪：表2只保留参与计算的字段：id_product、id_customer、stop_date+20个需要求和的数值字段，排除多余字段减少shuffle数据量
• 分区过滤：如果两张表有日期分区，提前过滤掉明显不符合逻辑的分区，比如表2过滤掉stop_date大于表1最大start_date的分区，表1过滤不需要统计的开通日期分区
• 类型优化：提前把start_date和stop_date转为timestamp类型，避免字符串比较的性能损耗和格式错误

2. 替换非等值关联为窗口预聚合（性能提升最明显） #

直接写非等值JOIN（带stop_date < start_date关联条件）会导致大量无效数据比对，大表场景下性能极差，推荐用窗口函数方案，仅需一次shuffle即可完成计算：

WITH union_data AS (
  -- 合并表1和表2的事件，统一按日期排序
  SELECT
    id_product,
    id_customer,
    start_date AS event_date,
    0 AS duration,
    0 AS col1,
    -- 剩余19个数值字段补0
    0 AS col20,
    1 AS row_type, -- 标记为表1的结果输出行
    start_date
  FROM table1

  UNION ALL

  SELECT
    id_product,
    id_customer,
    stop_date AS event_date,
    duration,
    col1,
    -- 剩余19个数值字段直接取原值
    col20,
    0 AS row_type, -- 标记为表2的指标行
    NULL AS start_date
  FROM table2
),
window_calc AS (
  SELECT
    id_product,
    id_customer,
    start_date,
    row_type,
    -- 按同产品同客户分区，按日期升序滚动求和
    SUM(duration) OVER(PARTITION BY id_product, id_customer ORDER BY event_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum_duration,
    SUM(col1) OVER(PARTITION BY id_product, id_customer ORDER BY event_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum_col1,
    -- 剩余19个字段滚动求和以此类推
    SUM(col20) OVER(PARTITION BY id_product, id_customer ORDER BY event_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS sum_col20
  FROM union_data
)
-- 仅筛选表1的输出行即为最终结果
SELECT id_product, id_customer, start_date, sum_duration, sum_col1, sum_col20
FROM window_calc
WHERE row_type = 1

3. 集群参数优化 #

• 开启自适应执行：设置spark.sql.adaptive.enabled=true、spark.sql.adaptive.skewJoin.enabled=true，自动处理关联倾斜、动态调整并行度
• 广播小表：如果其中一张表数据量小于10MB（可通过spark.sql.autoBroadcastJoinThreshold调整阈值），自动广播小表避免shuffle，比如表1是开通记录远小于停用记录表2的场景，可直接广播表1

内容的提问来源于stack exchange，提问作者agreüs