DynamoDB 未收到整个 SQS 消息正文答案

【问题标题】：DynamoDB not receiving the entire SQS message bodyDynamoDB 未收到整个 SQS 消息正文
【发布时间】：2021-04-15 04:05:23
【问题描述】：

我从 API 中批量提取数据并将其发送到 SQS 队列。我遇到的问题是处理消息以便将数据发送到 DynamoDB。数据集中应该有 147,689 条记录。但是，在运行代码时，有时会少于 147,689 条记录放入 DynamoDB，有时超过 147,689 条记录会放入 DynamoDB，有时会放入 147,689 条记录到 DynamoDB。将 147,689 条记录放入数据库并不一致。

我已经尝试了所有我能想到的方法来尝试解决这个问题，包括（使用 Fifo 队列而不是标准队列、增加可见性超时、增加传递超时、使用 uuid.uuid1() 而不是 uuid.uuid4 ()) 我正在循环浏览“记录”列表，所以不确定它为什么不处理整个批次。以下是我处理消息并将数据发送到 DynamoDB 的最新代码：

import boto3
import json
import uuid
import time

dynamo = boto3.client("dynamodb", "us-east-1")

def lambda_handler(event, context):
    for item in json.loads(event["Records"][0]["body"]):
        item["id"] = uuid.uuid1().bytes
        for key, value in item.items():
            if key == "id":
                item[key] = {"B": bytes(value)}
            elif key == "year":
                item[key] = {"N": str(value)}
            elif key == "amt_harvested":
                item[key] = {"N": str(value)}
            elif key == "consumed":
                item[key] = {"N": str(value)}
            else:
                item[key] = {"S": str(value)}

     
            time.sleep(0.001)
        
        dynamo.put_item(TableName="TableOne", Item=dict(item))

【问题讨论】：

检查 CloudWatch 中是否有任何 lambda 错误。
什么是 Lambda EventSourceMapping 中的 batchSize？您似乎只是从 SQS 读取第一条记录 Records"][0]，除非您的 batchSize 为 1，否则这可能不起作用。
batchSize 应该为 1，或者我们需要修改代码来循环 Records 数组，因为每次 Lambda 函数调用，最多将传递 10 条记录。
如果您决定将最大批量大小配置为 1（我假设您还没有），那么您还应该断言 event["Records"] 的长度为 1。否则，当有人更改批量大小时，您可能会再次遇到此类问题。最好的解决方案几乎总是简单地遍历event["Records"]，而不是假设它只有一个事件。
我会添加某种断言长度为 1 或简单地打印一条错误消息，这样您就可以确定是否有超过 1 个事件被呈现给 Lambda 函数。

标签： python amazon-web-services aws-lambda amazon-dynamodb amazon-sqs

【解决方案1】：

Lambda Event Source Mapping for SQS 将轮询消息并根据默认为 10 的批量大小为一批记录调用 Lambda 函数。处理该批应通过循环 event["Records"] 数组来完成。

设置批量大小应考虑的关键因素。

如果 lambda 处理失败，整个批次将被重新发送并由 AWS 重试。如果函数不能接受处理重复记录，batchsize 应该设置为 1。
如果在 lambda 中处理一条记录需要 20 毫秒，AWS 仍然会收取 100 毫秒（这是最低限度）的费用，我们只需将批量大小设置为 5 即可轻松降低 5 倍成本。

一直推荐

将批处理大小设置得更高，并将 lambda 编码为幂等。
使用 Lambda 代码处理所有记录，无论批次大小如何。

【讨论】：