PySpark大表Join遇GC overhead等错误，求排查与解决方法

表信息 #

df1.count(): 9989352358(2 columns)
df2.count(): 64000000(1 columns)

报错现象 #

执行Join操作时，Spark UI显示1000个任务中固定有1个失败，报错类型包括：

• GC overhead limit exceeded
• java.util.concurrent.TimeoutException
• heartbeat timeout

关键日志（问题主因线索） #

23/03/28 07:10:13 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: mem size 134387820 > 134217728: flushing 4840100 records to disk.
23/03/28 07:10:13 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 133711235
23/03/28 07:10:30 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: mem size 134319884 > 134217728: flushing 4900100 records to disk.
23/03/28 07:10:30 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 134321506
23/03/28 07:10:45 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: mem size 134362428 > 134217728: flushing 4800100 records to disk.
23/03/28 07:10:45 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 133932237
23/03/28 07:10:52 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: mem size 134568280 > 134217728: flushing 4820100 records to disk.
23/03/28 07:10:52 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 133781816
23/03/28 07:11:00 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: mem size 134445336 > 134217728: flushing 4920100 records to disk.
23/03/28 07:11:00 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 134417873
23/03/28 07:11:08 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: mem size 134452136 > 134217728: flushing 4870100 records to disk.
23/03/28 07:11:08 INFO org.apache.parquet.hadoop.InternalParquetRecordWriter: Flushing mem columnStore to file. allocated memory: 134350885

当前Spark配置 #

spark_config["spark.executor.memory"] = "16G"
spark_config["spark.executor.memoryOverhead"] = "8G"
spark_config["spark.executor.cores"] = "8"
spark_config["spark.driver.memory"] = "10G"
spark_config["spark.network.timeout"] = "20000s"
spark_config["spark.executor.heartbeatInterval"] = "1000000"
spark_config["spark.dynamicAllocation.enabled"] = "true"
spark_config["spark.sql.execution.arrow.pyspark.enabled"] = "true"
spark_config["spark.shuffle.service.enabled"] = "true"
spark_config["spark.dynamicAllocation.minExecutors"] = "200"
spark_config["spark.dynamicAllocation.maxExecutors"] = "250"

已尝试方法 #

• 重分区操作，但未解决问题

问题根源分析 #

1. 数据倾斜：1000个任务仅1个失败是典型的Join键数据倾斜表现——某个键的记录量远高于其他键，导致对应任务需要处理远超平均的数据集，内存过载引发GC频繁触发，进而拖慢任务执行，最终触发心跳超时。日志中频繁的Parquet刷盘是内存不足的直接体现，即便集群整体资源充足，单任务内存也无法承载倾斜的数据量。
2. Parquet写缓存阈值限制：默认Parquet内存缓存阈值为128MB（134217728字节），倾斜任务处理大量数据时，内存中快速堆积记录，频繁触发刷盘，IO开销剧增进一步拖慢任务，加剧超时风险。

可行解决方法 #

1. 检测并处理数据倾斜 #

步骤1：定位倾斜键

统计Join键的频次，找出异常高频的键：

# 统计df1的Join键频次
skewed_keys_df1 = df1.groupBy("join_key").count().orderBy("count", ascending=False).limit(10)
skewed_keys_df1.show()

# 统计df2的Join键频次
skewed_keys_df2 = df2.groupBy("join_key").count().orderBy("count", ascending=False).limit(10)
skewed_keys_df2.show()

步骤2：拆分倾斜键处理

针对热点键单独用广播Join（df2仅6400万行，热点键子集数据量更小，适合广播）：

# 假设定位到的热点键为"hot_key_value"
hot_key = "hot_key_value"

# 提取热点数据并广播Join
df1_hot = df1.filter(df1.join_key == hot_key)
df2_hot = df2.filter(df2.join_key == hot_key)
joined_hot = df1_hot.join(broadcast(df2_hot), on="join_key", how="inner")

# 处理非热点数据的常规Join
df1_non_hot = df1.filter(df1.join_key != hot_key)
df2_non_hot = df2.filter(df2.join_key != hot_key)
joined_non_hot = df1_non_hot.join(df2_non_hot, on="join_key", how="inner")

# 合并结果
final_df = joined_hot.union(joined_non_hot)

步骤3：加盐法处理通用倾斜键

如果倾斜键是null、0这类通用值，通过加盐拆分数据：

from pyspark.sql.functions import rand, lit, concat, explode, array

# 给df1的倾斜键添加0-9随机后缀
df1_salted = df1.withColumn(
    "join_key_salted",
    when(df1.join_key == "skewed_value", concat(df1.join_key, lit("_"), (rand() * 10).cast("int")))
    .otherwise(df1.join_key)
)

# 给df2的倾斜键复制10份，匹配不同后缀
df2_expanded = df2.withColumn(
    "join_key_salted",
    when(df2.join_key == "skewed_value", concat(df2.join_key, lit("_"), explode(array([lit(i) for i in range(10)]))))
    .otherwise(df2.join_key)
)

# 用加盐后的键执行Join
joined_df = df1_salted.join(df2_expanded, on="join_key_salted", how="inner").drop("join_key_salted")

2. 调整Parquet写缓存参数 #

增大Parquet内存缓存阈值，减少刷盘频率：

spark_config["spark.sql.parquet.memory.threshold"] = "256MB"  # 可根据实际情况进一步调大
spark_config["spark.sql.parquet.block.size"] = "512MB"  # 增大块大小，减少小文件数量

3. 优化Executor GC配置 #

针对大内存Executor调整GC策略，减少GC停顿：

spark_config["spark.executor.extraJavaOptions"] = "-XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=4 -XX:ConcGCThreads=2"

• UseG1GC：适合大内存场景的垃圾收集器，可更好控制GC停顿时间
• MaxGCPauseMillis：设置最大GC停顿时间，避免长时间停顿触发超时
• ParallelGCThreads/ConcGCThreads：调整GC线程数，适配8核Executor的资源分配

4. 调整任务并行度 #

根据Executor核心数调整Shuffle并行度，让每个任务处理的数据量更均衡：

# 设置Shuffle并行度为Executor数量*核心数，例如200*8=1600
spark_config["spark.sql.shuffle.partitions"] = "1600"

# 也可先对df1按Join键重分区后再执行Join
df1_repartitioned = df1.repartition(1600, "join_key")
joined_df = df1_repartitioned.join(df2, on="join_key")

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