使用boto3从EMR集群向DynamoDB加载海量数据过慢，求提速方案

嘿，我来帮你搞定这个批量加载的性能问题——2000万条要数分钟内完成，核心是把你的EMR集群算力用起来，当前的单进程代码完全浪费了5节点的优势，这才是慢的根源。下面分几个方向给你拆解优化方案：

1. 用EMR Spark实现分布式批量写入（最推荐） #

EMR的核心价值就是分布式处理，你现在用单进程跑Python代码，相当于只用到了集群里的一个节点的一个核，完全暴殄天物。用PySpark来读取S3数据并并行写入DynamoDB，才能真正发挥5节点的算力。

步骤&示例代码 #

from pyspark.sql import SparkSession

# 初始化SparkSession，加载Hadoop-AWS依赖来对接S3和DynamoDB
spark = SparkSession.builder \
    .appName("DynamoDB_Bulk_Load") \
    .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.3.4") \
    .getOrCreate()

# 读取S3上的JSON文件，Spark会自动拆分文件为多个分区，分布式处理
# 如果是单个大文件，建议先拆成100MB左右的小文件，或者用gzip压缩
df = spark.read.json("s3://mybucket/emp-rec.json")

# 配置DynamoDB写入参数，优化批量性能
df.write \
    .format("dynamodb") \
    .option("dynamodb.tableName", "EMP") \
    .option("dynamodb.output.batch.write.size", "25")  # 单次批量写入的条数，默认25，可根据吞吐量调整
    .option("dynamodb.throughput.write.percent", "0.9")  # 使用表90%的写吞吐量，避免触发限流
    .option("dynamodb.endpoint", "dynamodb.us-west-2.amazonaws.com")  # 替换成你的区域端点
    .mode("append") \
    .save()

关键注意点 #

• 文件拆分：如果你的emp-rec.json是单个超大文件，先在S3里拆分（比如用split命令或者AWS Glue拆分），Spark的分区效率会更高
• 压缩数据：把JSON文件用gzip压缩后上传到S3，减少读取时的网络传输时间和解压开销

2. 临时调高DynamoDB表的写入吞吐量 #

不管用哪种方式，DynamoDB的吞吐量是硬限制。2000万条要在5分钟（300秒）内完成，大概需要的写吞吐量（WCU）是：20000000 / 300 ≈ 66666 WCU（每条按1KB计算，1WCU=每秒写入1条1KB的项目）。

• 如果你的表是预配置吞吐量模式：临时把写吞吐量调到7万左右，写完之后再调回原来的数值，避免不必要的成本
• 如果是按需模式：DynamoDB会自动扩容，但按需的成本比预配置高，适合临时大流量场景
• 开启自动扩缩容：可以让DynamoDB根据负载自动调整吞吐量，不用手动操作

3. 优化原生Python代码（如果暂时不想用Spark） #

如果坚持用原生Python，至少要改成多进程/多线程，利用EMR节点的CPU核心：

多进程版示例代码 #

import boto3
import json
import decimal
from concurrent.futures import ProcessPoolExecutor

def write_records_batch(batch_records):
    # 每个进程初始化自己的DynamoDB客户端
    dynamodb = boto3.resource('dynamodb', region_name='us-west')
    table = dynamodb.Table('EMP')
    with table.batch_writer() as batch:
        for rec in batch_records:
            batch.put_item(Item=rec)

def main():
    s3 = boto3.client('s3')
    obj = s3.get_object(Bucket='mybucket', Key='emp-rec.json')
    records = json.loads(obj['Body'].read().decode('utf-8'), parse_float=decimal.Decimal)
    
    # 拆分记录为多个批次，每个批次1万条（可根据内存调整）
    batch_size = 10000
    record_batches = [records[i:i+batch_size] for i in range(0, len(records), batch_size)]
    
    # 进程数设置为EMR集群的总CPU核心数（比如5节点，每个节点4核，就设20）
    with ProcessPoolExecutor(max_workers=20) as executor:
        executor.map(write_records_batch, record_batches)

if __name__ == "__main__":
    main()

注意点 #

• batch_writer会自动处理重试和节流，但多进程要注意不要超过DynamoDB的吞吐量上限
• 如果S3文件太大，一次性加载到内存会OOM，建议分块读取（比如用ijson流式读取JSON）

4. 其他小技巧 #

• 过滤冗余字段：如果JSON里有不需要写入DynamoDB的字段，提前过滤掉，减少每条记录的大小，降低WCU消耗
• 用DynamoDB批量导入工具：AWS官方的dynamodb-import工具，适合从S3直接批量导入，不需要写代码，速度也很快

总结一下：优先用Spark+EMR的方案，这是处理海量数据最成熟的方式，配合临时调高DynamoDB吞吐量，2000万条数分钟内完全可以搞定。如果是一次性加载，用完记得把吞吐量调回去省成本哦！

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