DynamoDB 流 - 将数据写回表中

Question

考虑以下架构：

write -> DynamoDB table -> stream -> Lambda -> write metadata item to same table 

它可以用于很多很多糟糕的情况，例如表格和项目级别的聚合。我已经看到官方 AWS 工程师在几次技术讲座中推广了这种架构。

---

但是写入元数据项目不会将新项目添加到流并再次运行 Lambda 吗？

如何避免死循环？有没有办法避免元数据写入出现在流中？

或者这种架构不可避免地要花费 2 个流和 Lambda 请求？ （我们按请求收费）即如果它是元数据项，请提前退出 Lambda 函数。

Answer 1

由于从 DynamoDB 流触发 AWS Lambda 函数是一个二元选项（开/关），因此不可能仅针对某些表写入触发 AWS Lambda 函数。因此，您的 AWS Lambda 函数将针对它刚刚写入 DynamoDB 表的项目再次调用。重要的是在您的 AWS Lambda 函数中设置逻辑，以检测它是否写入了该数据，并且在这种情况下不再写入数据。否则你会得到上面提到的无限循环，这将是一个非常不幸的情况，特别是如果它被忽视的话。

Answer 2

目前 dynamo DB 不提供基于条件的流订阅，所以是的，Dynamo DB 将在无限循环中执行您的 lambda 函数，目前唯一的解决方案是限制您的 lambda 函数执行时间，您可以使用多个 lambda 函数，一个 lambda 函数只是为了检查元数据是否被写入，我正在分享一个云架构图，说明如何实现它，

Answer 3

有点晚了，但希望寻找更具示范性答案的人们会发现这很有用。

假设您要处理要添加到某个阈值的项目的记录，您可能有一个 if 条件来检查并处理或跳过记录，例如

此代码假定您的每个实体/对象类型都有一个属性“类型” - 这是 Rick Houlihan 本人向我推荐的，但您也可以检查属性是否存在，即"<your-attribute>" in record["dynamodb"]["NewImage"] - 您正在设计PK 和 SK 作为通用主键和排序键名。

threshold = (os.environ.get("THRESHOLD"))


def get_value():
    response = table.query(KeyConditionExpression=Key('PK').eq(<your-pk>))
    value = response['Items']['<your-attribute>'] if 'Items' in response else 0
    return value


def your_aggregation_function():
    # Your aggregation logic here
    # Write back to the table with a put_item call once done


def lambda_handler(event, context):
    for record in event['Records']:
        if record['eventName'] != "REMOVE" and record["dynamodb"]["NewImage"]["Type'] == <your-entity-type>:
            # Query the table to extract the attribute value
            attribute_value = get_value(record["dynamodb"]["Keys"]["PK"]["S"])
            if attribute_value < threshold:
                # Send to your aggregation function 

在 lambda 处理程序中设置条件（或者您可以根据需要更改位置）可以防止提到的无限循环。

您可能需要在更新表达式中进行额外检查，以确保两个（或更多）并发 lambda 不会写入同一个对象。我建议您使用 date = # timestamp defined in the lambda 并将其添加到 SK 中，或者如果您不能，请在您的项目中添加“EventDate”属性，以便您可以添加 ConditionExpression 或 UpdateExpression SET if_not_exists(#attribute, :date)

以上将保证您的 lambda 为 idempotent。