如何在DynamoDB中监听特定数值变更并触发应用函数

核心用DynamoDB Streams + Lambda 原生服务组合实现，无需额外部署中间件：

1. 开启DynamoDB流 #

进入目标DynamoDB表的「导出和流」配置页，启用DynamoDB流，流查看类型选择新旧映像（New and old images），只有该类型能同时拿到变更前后的完整数据，用于判断目标字段是否发生变更。

2. 配置流触发器 #

在表的「触发器」页新建触发器，关联用于逻辑判断的Lambda函数，按需配置触发批次大小：对延迟要求高设为1，普通业务场景设10~100降低调用频次。

3. 编写Lambda判断逻辑 #

Lambda会自动接收流推送的变更事件，你只需要在代码中过滤目标字段的变更，匹配后调用你的应用函数即可，以下是Python示例代码：

def lambda_handler(event, context):
    # 替换为你需要监听的字段名
    MONITOR_FIELD = "your_target_field"
    for record in event["Records"]:
        # 仅处理数据插入、更新事件
        if record["eventName"] not in ("INSERT", "MODIFY"):
            continue
        new_img = record["dynamodb"].get("NewImage", {})
        old_img = record["dynamodb"].get("OldImage", {})
        # 按字段类型取值，以下为数字类型示例，字符串类型替换N为S即可
        new_val = new_img.get(MONITOR_FIELD, {}).get("N")
        old_val = old_img.get(MONITOR_FIELD, {}).get("N") if old_img else None
        # 匹配规则：新插入数据包含该字段 / 已有数据该字段值发生变更
        if (old_val is None and new_val is not None) or (new_val != old_val):
            # 此处触发你的应用函数：如果应用部署在公网可直接发HTTP请求，内部应用可走专线/VPN调用，若逻辑可迁移直接写在Lambda内也可
            trigger_your_app_func(new_val, record["dynamodb"]["Keys"])

可选优化 #

• 若你的应用可用性无法保证，可在Lambda匹配到变更后先将事件写入SQS队列，由你的应用主动消费队列触发逻辑，避免事件丢失
• 若字段变更需要做幂等处理，可将DynamoDB的事件ID作为幂等键传入你的应用，避免重复触发

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