PySpark中多表同条件Join优化方案咨询

问题背景 #

现有30亿行40列的数据集A，需与3000万行的B、6000万行的C分别做左关联，从B取1列、从C取2列数据。A与B、A与C均为多对多关系，关联条件均基于user_id和log_id匹配：

join_conditions_A_B = [
    A.user_id == B.b_user_id,
    A.log_id == B.b_log_id
]

join_conditions_A_C = [
    A.user_id == C.c_user_id,
    A.log_id == C.c_log_id
]

当前两次独立关联的逻辑已运行23小时未完成，核心痛点是A被重复shuffle两次，需通过合并关联操作减少shuffle开销。

核心优化思路 #

两次关联的分区键完全一致（user_id+log_id），因此可以先将B、C按相同键合并为一个临时表，再与A做一次左关联，避免A被多次shuffle。同时提前对所有数据集按关联键分区，进一步降低shuffle成本。

优化后代码实现 #

import pyspark.sql.functions as F

# 1. 预处理B：统一关联键名称，重命名待添加列
B_processed = B.select(
    F.col("user_id"),
    F.col("log_id"),
    F.col("column_to_be_added_to_A").alias("b_target_col")
)

# 2. 预处理C：统一关联键名称，重命名待添加列
C_processed = C.select(
    F.col("user_id"),
    F.col("log_id"),
    F.col("column_to_be_added_to_A_1").alias("c_target_col1"),
    F.col("column_to_be_added_to_A_2").alias("c_target_col2")
)

# 3. 合并B、C：按关联键做全外关联，保留双方所有匹配项
# 全外关联确保B和C中不重叠的user_id+log_id组合都能被保留
BC_combined = B_processed.join(
    C_processed,
    on=["user_id", "log_id"],
    how="full"
)

# 4. 提前分区：按关联键重分区，避免关联时重复shuffle
# 分区数根据集群资源调整，30亿行建议设置为2000-5000
target_partitions = 3000
A_partitioned = A.repartition(target_partitions, "user_id", "log_id")
BC_partitioned = BC_combined.repartition(target_partitions, "user_id", "log_id")

# 5. 单次左关联完成数据合并
A_final = A_partitioned.join(
    BC_partitioned,
    on=["user_id", "log_id"],
    how="left"
)

额外优化建议 #

• 避免广播小表：B、C数据量过大（千万级），远超Spark广播阈值，强行广播会导致Driver内存溢出，因此不适用broadcast join。
• 处理数据倾斜：检查user_id或log_id是否存在热点值，若有倾斜可拆分热点键单独关联，再合并结果。
• 调整Spark参数：调大spark.sql.shuffle.partitions至2000+（默认200），同时根据集群配置提升spark.executor.memory和spark.driver.memory，增强并行处理能力。

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