如何用PySpark(Glue)高效批量加载DynamoDB记录？替代PutItem方案

核心结论：放弃单条PutItem，改用批量写入API或Spark DynamoDB连接器，这是解决性能问题的关键。 #

1. 首选方案：Spark DynamoDB官方连接器 #

AWS提供的Spark-DynamoDB连接器底层自动封装了BatchWriteItem API（一次最多处理25条记录/400KB数据），能大幅降低网络开销，同时自动处理节流重试、并行写入等问题，是Glue环境下批量写入的最优选择。

实现步骤： #

• 添加依赖：在Glue作业的依赖包中加入对应版本的连接器，例如com.amazonaws:dynamodb-spark-connector_2.11:1.0.0（需匹配Spark版本，Glue 2.0+对应Spark 2.4，Glue 3.0+对应Spark 3.1）。
• 代码示例（Scala）：

// 从GlueContext获取SparkSession
val spark = glueContext.sparkSession

// 假设你已通过Glue读取数据源得到DynamicFrame，转换为DataFrame
val sourceDf = glueContext.createDynamicFrame.fromCatalog(...)
  .toDF()

// 配置DynamoDB参数
val dynamoOpts = Map(
  "dynamodb.tableName" -> "your-target-table",
  "dynamodb.writeThroughputPercent" -> "0.7", // 控制占用表的预留写入吞吐量比例，避免节流
  "aws.region" -> "us-east-1"
)

// 批量写入
sourceDf.write
  .format("dynamodb")
  .options(dynamoOpts)
  .mode("append") // 或"overwrite"，根据业务需求选择
  .save()

2. 手动封装BatchWriteItem API #

如果需要更精细的控制（比如自定义重试逻辑），可以直接调用AWS SDK的BatchWriteItem接口，手动将数据分批次处理：

关键要点： #

• 按25条记录/400KB的限制拆分数据批次
• 处理API返回的UnprocessedItems，添加指数退避重试逻辑（避免触发DynamoDB节流）
• 利用Spark的并行化能力，将批次分配到不同Executor执行

代码示例（Scala）： #

import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder
import com.amazonaws.services.dynamodbv2.model.{PutRequest, WriteRequest}
import scala.collection.JavaConverters._

// 初始化DynamoDB客户端
val ddbClient = AmazonDynamoDBClientBuilder.defaultClient()

// 将DataFrame转换为DynamoDB的PutRequest列表
val putRequests = sourceDf.rdd.map { row =>
  val item = // 将Row转换为DynamoDB Item格式（比如用Item.fromMap）
  new WriteRequest().withPutRequest(new PutRequest().withItem(item))
}.collect()

// 分批次处理
putRequests.grouped(25).foreach { batch =>
  val request = new com.amazonaws.services.dynamodbv2.model.BatchWriteItemRequest()
    .withRequestItems(Map("your-target-table" -> batch.asJava).asJava)
  
  var result = ddbClient.batchWriteItem(request)
  // 处理未成功写入的项目，添加指数退避重试
  while (!result.getUnprocessedItems.isEmpty) {
    Thread.sleep(1000) // 可根据情况调整等待时间，实现指数退避
    result = ddbClient.batchWriteItem(new BatchWriteItemRequest().withRequestItems(result.getUnprocessedItems))
  }
}

3. 额外优化建议 #

• 调整Spark并行度：根据DynamoDB表的预留写入吞吐量设置合理的分区数，比如每1000写入吞吐量对应2-3个Spark分区，避免写入压力集中。
• Glue资源配置：增加Worker数量或选用更大规格的Worker（比如G.2X），提升并行处理能力。
• 避免热点：如果写入的记录集中在某个分区键值，会导致DynamoDB热点，需调整分区键设计或添加随机后缀分散写入。

为什么PutItem速度极慢？ #

PutItem是单条请求模式，每条记录都要单独建立HTTP连接、发送请求、等待响应，3万条记录就会产生3万次独立请求，网络延迟和服务端开销叠加后，性能会急剧下降。而批量API能将多次请求合并，大幅减少网络往返次数，效率提升可达10-100倍。

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