如何在不超出吞吐量上限的情况下批量更新DynamoDB多条数据

我需要对列表中的每个条目执行DynamoDB的update_item操作，列表包含id和total_sales字段，total_sales每小时更新一次。我的需求是遍历列表，根据id更新对应条目的stats.revenue字段（值为total_sales），但列表条目超过5000条，已超出吞吐量上限。请问有没有比以下代码更高效的实现方式？

def write_to_dynamo(id, total_sales):
    response = table.update_item(
            Key={
                'id': id
            },
            UpdateExpression="set stats.revenue = :r",
            ExpressionAttributeValues={
                ':r': total_sales
            },
            ReturnValues="UPDATED_NEW"
        )
    return response

def main():
    for item in lst:
        write_to_dynamo(item.id, item.total_sales)

1. 使用DynamoDB批量写入器（batch_writer） #

DynamoDB的batch_writer会自动处理批量请求的拆分、重试和指数退避，每个批量请求最多包含25个操作，刚好符合DynamoDB的限制。用它包装更新操作能大幅减少API调用次数：

def update_batch(items):
    with table.batch_writer() as batch:
        for item in items:
            batch.update_item(
                Key={'id': item.id},
                UpdateExpression="set stats.revenue = :r",
                ExpressionAttributeValues={':r': item.total_sales}
            )

def main():
    update_batch(lst)

内置的重试机制能自动处理吞吐量不足的临时报错，比单条循环高效得多。

2. 控制并发请求数 #

如果需要更快的处理速度，可以用线程池并发执行更新，但要根据表的读写容量单位（RCU/WCU）调整并发数，避免触发限流：

from concurrent.futures import ThreadPoolExecutor

def write_to_dynamo(item):
    table.update_item(
        Key={'id': item.id},
        UpdateExpression="set stats.revenue = :r",
        ExpressionAttributeValues={':r': item.total_sales}
    )

def main():
    # 根据表吞吐量调整max_workers，建议从50开始测试
    with ThreadPoolExecutor(max_workers=50) as executor:
        executor.map(write_to_dynamo, lst)

3. 切换到按需容量模式 #

如果更新是每小时一次的突发负载，可以把表的容量模式从预配置切换到按需模式。DynamoDB会自动扩容应对突发请求，无需手动调整吞吐量，适合这种非持续高负载的场景（注意成本会略高于预配置模式）。

4. 分批处理+指数退避 #

如果必须用预配置吞吐量，可以手动拆分批次，配合指数退避处理限流报错：

import time
from botocore.exceptions import ClientError

def write_to_dynamo(id, total_sales):
    retry_count = 0
    max_retries = 5
    while retry_count < max_retries:
        try:
            return table.update_item(
                Key={'id': id},
                UpdateExpression="set stats.revenue = :r",
                ExpressionAttributeValues={':r': total_sales}
            )
        except ClientError as e:
            if e.response['Error']['Code'] == 'ProvisionedThroughputExceededException':
                retry_count += 1
                time.sleep(2 ** retry_count)  # 指数退避等待
            else:
                raise

def main():
    batch_size = 100  # 批次大小根据吞吐量调整
    for i in range(0, len(lst), batch_size):
        batch = lst[i:i+batch_size]
        for item in batch:
            write_to_dynamo(item.id, item.total_sales)
        time.sleep(1)  # 批次间等待降低负载

额外优化点 #

• 不需要返回更新后数据的话，删除ReturnValues="UPDATED_NEW"，减少数据传输量。
• 确保id是表的主键，否则update_item无法直接定位条目。

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