AWS Glue写入S3触发ExecutorLostFailure故障求助

核心问题定位 #

你的问题根源在于JDBC读取阶段默认生成单分区DataFrame，即使后续执行repartition，全量数据已经先加载到单个Executor内存中。当处理1400万行数据时，单Executor内存/资源超限被Kill，导致写入阶段出现ExecutorLostFailure错误；而600万行数据量较小，单Executor可以承载，因此任务正常。后续的repartition是在内存中重新分配数据，无法解决初始读取阶段的单节点过载问题。

针对性解决方案 #

1. 读取阶段直接并行分区（最关键） #

不要在读取后再做分区，而是在JDBC读取时就通过分区键拆分数据，让Spark从数据库并行拉取，避免单节点加载全量数据。针对PostgreSQL，配置如下：

• 选一个分布均匀的数值型/日期型字段（比如主键ID、创建时间）作为partitionColumn
• 设置该字段的lowerBound（最小值）和upperBound（最大值）
• 指定numPartitions（拆分的分区数，建议按数据量设置，比如100）

修改后的读取代码示例：

filter_query = f'SELECT * FROM {DB_TABLE}'
sql_query = '({}) as query'.format(filter_query)
spark_df = (glueContext.read.format('jdbc')
            .option('driver', 'org.postgresql.Driver')
            .option('url', JDBC_URL)
            .option('dbtable', sql_query)
            .option('user', DB_USERS)
            .option('password', DB_PASSWORD)
            # 新增并行分区配置
            .option('partitionColumn', 'id')  # 替换为你表中分布均匀的字段
            .option('lowerBound', 1)          # 该字段的最小值
            .option('upperBound', 14000000)   # 该字段的最大值
            .option('numPartitions', 100)     # 拆分的分区数
            .load()
            )

2. 优化写入阶段资源与配置 #

• 合理配置Executor资源：不要盲目调大spark.executor.memory，需配合spark.executor.cores调整。比如Glue标准Worker是4核16G，可设置--conf spark.executor.cores=4 --conf spark.executor.memory=12g（留足堆外内存给系统使用）。
• 减少IO压力：写入CSV时开启压缩，降低S3写入量和网络传输压力：

  spark_df.write.format("csv")\
      .option("header", "true")\
      .option("compression", "gzip")\
      .save(S3_PATH)
  

3. 调整Glue Worker配置 #

• 不要直接加到500个Worker，先根据数据量选合适的Worker类型：内存密集型任务优先选G.2X/G.4X（更大内存）。
• Worker数量可按分区数设置：比如100个分区配20-50个Worker，让每个Worker处理2-5个任务，避免资源浪费。

4. 排查数据倾斜 #

如果选的分区字段分布不均匀，会导致部分分区数据量过大，仍可能触发资源超限。此时可以：

• 换一个更均匀的分区字段
• 进一步增加numPartitions，缩小单个分区的数据量

修改后的完整代码 #

def read_from_db():
    logger.info(f'Starts Reading Data from {DB_TABLE} table')
    start = time.perf_counter()
    filter_query = f'SELECT * FROM {DB_TABLE}'
    sql_query = '({}) as query'.format(filter_query)
    # 读取阶段直接并行分区
    spark_df = (glueContext.read.format('jdbc')
                .option('driver', 'org.postgresql.Driver')
                .option('url', JDBC_URL)
                .option('dbtable', sql_query)
                .option('user', DB_USERS)
                .option('password', DB_PASSWORD)
                .option('partitionColumn', 'id')
                .option('lowerBound', 1)
                .option('upperBound', 14000000)
                .option('numPartitions', 100)
                .load()
                )
    end = time.perf_counter()
    logger.info(f'Count of records in DB is {spark_df.count()}')
    logger.info(f'Elapsed time for reading records from {DB_TABLE} table = {end - start:0.4f} seconds')
    logger.info(f'Finished Reading Data from {DB_TABLE} table')
    logger.info(f"Total no. of partitions - {spark_df.rdd.getNumPartitions()}")

    S3_PATH = f"{S3_BUCKET}/all-entities-update/{date}/{cur_time}"
    # 开启压缩优化写入
    spark_df.write.format("csv")\
        .option("header", "true")\
        .option("compression", "gzip")\
        .save(S3_PATH)
    return

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