PySpark3大小表左Join执行缓慢，重分区后仍有长尾问题如何解决

云上老周

2026-10-7

问题根因分析

非等值Join的隐性计算膨胀：你的Join包含两个条件，除等值键匹配外还有时间差大于5小时的非等值条件，即使分区的总输入行数均匀，也可能出现单个分区内部分等值键（a/a2）的时间范围重叠度极高的情况，比如某一个a值对应100条事件表数据、1万条明细表数据，满足时间差条件的匹配结果可达100万条，计算量随匹配对数指数级上升，和分区输入行数无关。
加盐分区逻辑错误：你仅给明细表加了盐分区，但Join时Shuffle还是会按等值Join键a/a2重新分发数据，前面加盐的重分区完全无效，还额外多了一次Shuffle，从执行计划也能看到明细表侧有两次Exchange操作，属于多余开销。
Join条件的重复计算开销：你在Join条件中每次匹配都要实时执行unix_timestamp字符串转时间戳的计算，每一行配对都要重复跑两次转换逻辑，CPU开销极高，放大了慢任务的耗时。
可选排查点：可去Spark UI查看慢任务的Shuffle读写、GC耗时、输入字节数，也可能是单任务GC停顿过长、源Parquet文件压缩比不均匀导致的读倾斜、S3对象访问热点的问题。

对应解决方案

提前预计算时间戳列，避免Join时重复计算

# 提前把日期转成时间戳数值，Join时直接比较
events = events.withColumn("ts_event", unix_timestamp("date").cast("long"))
details = details.withColumn("ts_detail", unix_timestamp("date").cast("long"))

join_condition = [
    details["a"] == events["a2"],
    events["ts_event"] - details["ts_detail"] > 5 * 3600
]

修正加盐Join逻辑，两边同时加盐打散等值键数据

nSaltBins = 200 # 可根据数据倾斜程度调整
# 事件表每一行复制nSaltBins份，每份对应一个盐值
events = events.withColumn("salt", explode(array([lit(i) for i in range(nSaltBins)])))
# 明细表每一行随机分配一个盐值
details = details.withColumn("salt", (rand(seed=42) * nSaltBins).cast("int"))

# 按等值键+盐共同分区，避免同个a值的所有数据落到同一个分区
details = details.repartition(1770, "a", "salt")
events = events.repartition(1770, "a2", "salt")

join = events.join(
    details,
    ["a2", "salt"] + join_condition[1:],
    "left"
).drop("salt", "a")

调整Spark配置规避异常优化
修改Spark配置，先关闭AQE的分区自动合并功能，避免小分区被合并后计算量过大，同时将Shuffle分区数调整为总核心数的2~3倍，降低单任务计算量：

spark = SparkSession.builder.appName('Test')\
    .config("spark.driver.memory", "108g")\
    .config("spark.executor.instances", "59")\
    .config("spark.executor.memoryOverhead", "13312")\
    .config("spark.executor.memory", "108g")\
    .config("spark.executor.cores", "15")\
    .config("spark.driver.cores", "15")\
    .config("spark.default.parallelism", "1770")\
    .config("spark.sql.adaptive.enabled", "true")\
    .config("spark.sql.adaptive.skewJoin.enabled", "true")\
    # 关闭分区自动合并，避免合并后单分区计算量过大
    .config("spark.sql.adaptive.coalescePartitions.enabled", "false")\
    # 调整Shuffle分区数为总核心数的2倍左右
    .config("spark.sql.shuffle.partitions", "2000")\
    .getOrCreate()

源文件优化：如果排查到是源Parquet文件大小不均匀导致的读倾斜，可提前将未分区的Parquet文件重写为固定128M~256M大小的分片，再做后续计算。

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