如何每日将S3数据增量同步至已有DynamoDB表？

要实现你说的每日自动增量同步，核心就是对比当日S3里的ID列表和DynamoDB现有数据的差异，只处理新增和删除的部分，不用全量覆盖。下面是具体的落地步骤：

1. 触发同步任务 #

用AWS EventBridge（原CloudWatch Events）设置每日定时规则，到点就触发Lambda函数执行同步逻辑——这是最省心的无服务器触发方式，不用管服务器运维。

2. 读取当日S3的ID列表 #

在Lambda里用AWS SDK读取对应日期文件夹下的文件，把逗号分隔的字符串转成集合（方便后续对比）：

import boto3
from datetime import datetime

s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your-table-name')

def lambda_handler(event, context):
    # 生成当日日期格式，要和S3文件夹命名一致（比如YYYY-MM-DD）
    today = datetime.today().strftime('%Y-%m-%d')
    bucket = 'your-s3-bucket'
    file_key = f'{today}/id_list.txt'  # 替换成你的实际文件路径

    # 读取并解析S3文件
    s3_response = s3.get_object(Bucket=bucket, Key=file_key)
    raw_ids = s3_response['Body'].read().decode('utf-8').strip()
    current_ids = set(raw_ids.split(','))
    # 清理掉空字符串（防止文件里有多余逗号）
    current_ids = {id.strip() for id in current_ids if id.strip()}

3. 获取DynamoDB的现有ID集合 #

如果你的DynamoDB表规模不大，直接全量扫表拿所有ID就行：

existing_ids = set()
# 用分页器处理大量数据，避免单次扫描返回不全
paginator = table.get_paginator('scan')
for page in paginator.paginate(ProjectionExpression='Identifier'):  # 主键字段名改成你表的实际主键
    existing_ids.update(item['Identifier'] for item in page['Items'])

要是表数据量很大，全量扫描太费资源，建议每次同步完把当日的ID集合存成S3快照（比如snapshots/latest_ids.txt），下次同步直接读这个快照，不用扫表：

existing_ids = set()
try:
    snapshot_response = s3.get_object(Bucket=bucket, Key='snapshots/latest_ids.txt')
    existing_ids = set(snapshot_response['Body'].read().decode('utf-8').splitlines())
except s3.exceptions.NoSuchKey:
    # 首次同步时快照不存在，默认现有ID为空
    pass

4. 计算差异并批量同步 #

拿到两个集合后，直接用集合运算算出要新增和删除的ID，然后用DynamoDB的批量写入接口一次性处理：

# 算差异：新增当日有但DB没有的，删除DB有但当日没有的
to_add = current_ids - existing_ids
to_delete = existing_ids - current_ids

# 批量操作DynamoDB
with table.batch_writer() as batch:
    # 新增ID
    for id in to_add:
        batch.put_item(Item={'Identifier': id})  # 可以加同步时间等额外字段，比如'SyncDate': today
    # 删除失效ID
    for id in to_delete:
        batch.delete_item(Key={'Identifier': id})

# 如果用了快照机制，更新S3里的最新快照
if to_add or to_delete:
    s3.put_object(
        Bucket=bucket,
        Key='snapshots/latest_ids.txt',
        Body='\n'.join(current_ids).encode('utf-8')
    )

return {
    'statusCode': 200,
    'body': f'完成同步：新增{len(to_add)}个ID，删除{len(to_delete)}个ID'
}

关键注意点 #

• 主键要对应：确保DynamoDB表的主键就是你同步的Identifier字段，不然没法准确匹配和操作。
• 错误处理：加个异常捕获，比如S3文件不存在、DynamoDB写入超时，还可以给Lambda配个死信队列，同步失败了能留记录排查。
• 大数量适配：如果单次同步的ID超过25个，Lambda的batch_writer会自动拆分请求，不用自己处理，但要是百万级别的ID，建议分批次处理，避免超时。
• 幂等性：就算任务重复触发，因为是对比集合差异，重复执行也不会有问题，不会重复新增或删除。

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