PYSPARK中Spark作业读取PostgreSQL遇内存溢出及GC超限问题求助

针对你遇到的情况——已经调整spark.driver.memory解决了堆空间OOM，但仍存在GC overhead超限问题，我整理了几个实用的解决方案，按优先级排序：

1. 调整Executor的内存与资源配置 #

Driver内存只是解决了驱动端的问题，而Spark作业的大部分计算逻辑是在Executor节点执行的，GC问题往往出在这里：

• 增加Executor内存：在spark-defaults.conf中添加或修改spark.executor.memory，比如设置为2g或更高（根据集群资源情况调整）；同时配套调整spark.executor.cores，合理分配CPU核心数，避免单Executor承载过多任务导致内存压力过载。
• 优化Executor内存分配比例：Spark默认会给Executor堆内存分配一部分用于存储（Storage）和执行（Execution），可以通过spark.memory.fraction和spark.memory.storageFraction调整，比如将spark.memory.fraction设为0.8（默认0.6），让更多内存用于计算和缓存，减少内存碎片化。

2. 优化PostgreSQL数据读取逻辑 #

不合理的JDBC读取方式会导致大量数据一次性加载到内存，触发频繁GC：

• 拆分读取分区：默认情况下Spark读取JDBC数据源可能只用少量分区，导致单分区数据量过大。可以通过partitionColumn、lowerBound、upperBound、numPartitions参数拆分数据，示例代码如下：

  val df = spark.read.format("jdbc")
    .option("url", "jdbc:postgresql://host:port/db")
    .option("dbtable", "your_target_table")
    .option("user", "db_user")
    .option("password", "db_pass")
    .option("partitionColumn", "id")
    .option("lowerBound", "1")
    .option("upperBound", "1000000")
    .option("numPartitions", "10")
    .load()
  

  这样会把数据分成10个分区并行读取，每个分区的数据量更小，减轻单Executor的内存压力。
• 提前过滤数据：在读取时通过dbtable参数传入带WHERE条件的SQL语句，只加载聚合需要的字段和数据，比如option("dbtable", "(SELECT id, agg_field FROM your_table WHERE date >= '2024-01-01') AS filtered_data")，直接减少加载到Spark中的数据总量。

3. 优化聚合操作逻辑 #

聚合操作是GC高发场景，优化逻辑能大幅减少内存占用：

• 优先使用reduceByKey/aggregateByKey替代groupByKey：groupByKey会先把相同key的数据全部shuffle到同一个节点再聚合，而reduceByKey会先在每个Executor本地做预聚合，减少shuffle的数据量和内存占用。
• 用广播变量处理小表关联：如果聚合过程涉及表关联，且其中一张表数据量很小，可以将其转为广播变量，避免shuffle操作，降低内存开销。
• 避免不必要的缓存：如果聚合后的数据不需要重复使用，不要盲目调用cache()或persist()，缓存的数据会持续占用Executor内存，增加GC压力。

4. 调整JVM GC参数 #

针对Spark的Executor和Driver，定制JVM GC参数优化垃圾回收：

• 在spark-defaults.conf中添加以下配置：

  spark.executor.extraJavaOptions -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:G1HeapRegionSize=16m
  spark.driver.extraJavaOptions -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:G1HeapRegionSize=16m
  

  G1GC是针对大内存场景优化的垃圾收集器，能更好地控制GC停顿时间，避免出现GC overhead超限的情况。

5. 排查并解决数据倾斜 #

如果某个key的数据量远大于其他key，会导致单个Executor处理海量数据，引发GC问题：

• 通过Spark UI的Stages页面，观察每个Task的处理数据量和执行时间，定位数据倾斜的key。
• 解决办法包括：对倾斜key加盐拆分、将倾斜key单独处理、给倾斜key添加随机前缀分散到多个Executor等。

内容的提问来源于stack exchange，提问作者Jaya Sree Meruga