AWS Glue处理十万份S3文件写入PostgreSQL性能优化求助

看起来你现在在处理十万份S3小文件的Glue任务，耗时超一小时确实让人头疼！我结合你的代码逻辑，给你几个针对性的优化方向，应该能帮你大幅提速：

1. 优化S3小文件读取逻辑，减少分区开销 #

你当前用wholeTextFiles读取十万份小文件，这会直接生成十万个Spark分区——每个文件对应一个分区。过多的分区会导致任务调度、启动的开销暴增，而且后续shuffle阶段的压力也会很大。

优化方案： #

• 调整Spark分区大小参数：在作业开头添加以下配置，让Spark自动将多个小文件合并到同一个分区（建议设置为128MB或256MB，对应S3和Spark的最佳实践）：

  spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728")  # 128MB
  spark.conf.set("spark.sql.files.openCostInBytes", "134217728")
  

• 改用Glue原生读取方式：替代wholeTextFiles，用Glue的create_dynamic_frame_from_options读取文本文件，它会更智能地处理小文件合并：

  from awsglue.dynamicframe import DynamicFrame
  ind_dyf = glueContext.create_dynamic_frame_from_options(
      connection_type="s3",
      connection_options={"paths": [f"s3a://{source_bucket}/{source_prefix}/"], "recurse": True},
      format="text"
  )
  # 转成DataFrame继续处理
  whole_df = ind_dyf.toDF().withColumnRenamed("text", "content")
  

• 提前合并S3小文件：如果这些小文件是长期存在的，建议用S3 Batch Operations或一个前置Glue任务，将大量小文件合并成几个大文件（比如每个100MB以上），从根源上解决小文件问题。

2. 简化数据转换逻辑，减少重复计算 #

你的转换代码中存在重复的split操作（在生成struct时两次调用split(x, ':', 2)），这会额外消耗CPU资源。同时，循环调用withColumn提取字段也会让查询计划变得冗余。

优化后的转换代码： #

from pyspark.sql.functions import col, expr, map_from_entries

# Step1: 分割内容为键值对数组
whole_df = whole_df.withColumn("kv_pairs", split(col("content"), "\\|"))

# Step2: 优化transform逻辑，避免重复split
whole_df = whole_df.withColumn(
    "fields_map",
    map_from_entries(
        expr("transform(kv_pairs, x -> let kv = split(x, ':', 2) in struct(kv[0] as key, kv[1] as value))")
    )
)

# Step3: 一次性提取所有字段，替代循环withColumn
fields_to_extract = [
    "SSN_TIN", "TIDTYPE", "FNAME", "MNAME", "LNAME", "ENTNAME",
    "BRK_ACCT", "DIR_ACCT", "NONBRKAC", "SPONSOR", "REP_ID", "RR2",
    "REG_TY", "LOB", "PROC_DTE", "DT_REC", "SCANDTS", "XTRACID",
    "NASU_ID", "DC_SRCRF", "CHK_AMT", "CHK_NUML", "DEPOSDT", "RCVDDATE",
    "STK_CNUM", "STK_SHR", "DOC-TY-ID"
]
# 一次性生成所有字段的表达式
extract_exprs = [col("fields_map").getItem(field).alias(field) for field in fields_to_extract]
# 直接select生成最终DataFrame，保留path字段（如果需要的话）
whole_df = whole_df.select("path", *extract_exprs).drop("content", "kv_pairs", "fields_map")

3. 优化PostgreSQL写入性能 #

JDBC写入是常见的性能瓶颈，默认配置下批量大小小、并行度不足，会导致写入速度极慢。

优化写入配置： #

# 去掉不必要的cache()！count()会触发一次全量计算，写入又会再算一次，完全浪费资源
# client_df.cache()
# print(f"Total rows in client_df: {client_df.count()}")

client_dynamic_frame = DynamicFrame.fromDF(client_df, glueContext, "dynamic_frame")

# 调整JDBC连接参数，开启批量写入
glueContext.write_dynamic_frame.from_options(
    frame=client_dynamic_frame,
    connection_type="JDBC",
    connection_options={
        "connectionName": connection_name,
        "database": tgt_database,
        "dbtable": "staging.bpm_migration_client1",
        "useConnectionProperties": "true",
        # 关键优化参数
        "batchsize": "10000",  # 每次批量插入10000条
        "rewriteBatchedStatements": "true",  # PostgreSQL开启批量插入优化
        "numPartitions": "20",  # 并行写入的分区数，根据数据库承载能力调整（建议10-50）
        "truncate": "false"  # 如果是覆盖写入可以设为true，比delete更快
    }
)

注意：numPartitions不要设置过大，避免超出PostgreSQL的最大连接数（默认是100），可以根据数据库的max_connections参数调整。

4. 调整Glue作业资源配置 #

最后，确保你的Glue作业有足够的资源来并行处理数据：

• Worker类型与数量：改用G.1X或G.2X类型的Worker（比标准型Worker有更多CPU和内存），并增加Worker数量（比如从默认2个增加到10-20个，根据数据量调整）。
• Spark参数优化：在Glue作业的“作业参数”中添加以下配置：

  --conf spark.sql.shuffle.partitions=40  # 建议设置为Worker数量*4
  --conf spark.default.parallelism=40
  --conf spark.driver.memory=8g  # 根据Worker类型调整
  

这些优化点你可以逐个尝试，优先从调整S3读取和JDBC写入参数开始，应该能看到明显的提速效果！

内容来源于stack exchange