DynamoDB batch_writer未向表中写入数据问题排查

1. 修正变量名错误 #

你的函数参数是invocations_table，但代码中使用了未定义的table变量调用batch_writer()——这是最可能导致无数据写入的原因。如果table不是指向目标表的全局变量，这段代码要么抛出NameError，要么写入到错误的表中。

修复后的代码：

def write_invocations_to_dynamodb(self, invocations_table):
    with invocations_table.batch_writer() as batch:
        for invocation in self.invocations:
            item = {
                "conversationId": invocation.conversation_ulid,
                "invocationId": invocation.invocation_ulid,
                "modelId": invocation.model_id,
                "modelVersionId": invocation.model_version_id,
                "systemPrompt": invocation.system_prompt,
                "isContextTrimmed": invocation.is_context_trimmed,
                "userInput": invocation.user_input,
                "modelOutput": invocation.model_output
            }
            logger.info(f"Inserting item:{item}")
            batch.put_item(Item=item)

2. 检查数据源是否为空 #

查看CloudWatch日志中是否有Inserting item:的输出。如果没有，说明self.invocations是空列表，没有数据可写入。需要确认调用该函数前，self.invocations已正确填充调用记录。

3. 验证表主键匹配 #

确保你的item包含目标DynamoDB表定义的分区键（及排序键，若有）。如果缺失主键字段，DynamoDB会抛出ValidationException，但如果代码未捕获并打印异常，可能看不到错误。可以添加异常捕获排查：

def write_invocations_to_dynamodb(self, invocations_table):
    try:
        with invocations_table.batch_writer() as batch:
            for invocation in self.invocations:
                item = {
                    "conversationId": invocation.conversation_ulid,
                    "invocationId": invocation.invocation_ulid,
                    "modelId": invocation.model_id,
                    "modelVersionId": invocation.model_version_id,
                    "systemPrompt": invocation.system_prompt,
                    "isContextTrimmed": invocation.is_context_trimmed,
                    "userInput": invocation.user_input,
                    "modelOutput": invocation.model_output
                }
                logger.info(f"Inserting item:{item}")
                batch.put_item(Item=item)
    except Exception as e:
        logger.error(f"Error writing to DynamoDB: {str(e)}")
        raise

4. 确认IAM权限配置 #

确保执行该函数的IAM角色拥有dynamodb:BatchWriteItem权限，且资源ARN正确指向目标表。权限策略示例：

{
    "Effect": "Allow",
    "Action": "dynamodb:BatchWriteItem",
    "Resource": "arn:aws:dynamodb:你的区域:你的账号ID:table/你的目标表名"
}

5. 检查目标表是否正确 #

如果table是全局变量，确认它指向的是你正在检查的DynamoDB表，而非测试环境或其他无关表。

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