DynamoDB触发Lambda函数调用失败 求排查解决方案

问题根源 #

从日志及代码分析，核心问题有两点：

1. 数据类型不兼容：Kinesis Firehose的put_record接口要求Data参数必须是字节流，但当前代码直接传入字符串，触发类型错误。
2. 无异常捕获机制：未处理DynamoDB记录字段缺失、Firehose调用失败等场景，导致执行中断且无有效报错信息。

修复后的代码 #

import json
import boto3
from botocore.exceptions import ClientError

firehose_client = boto3.client('firehose')
DELIVERY_STREAM = "OrdersAuditFirehose"

def lambda_handler(event, context):
    batch_records = []
    for record in event['Records']:
        try:
            parsed = parse_dynamo_record(record['dynamodb'])
            # 转换为Firehose要求的字节类型，添加换行符便于S3批量文件解析
            batch_records.append({'Data': json.dumps(parsed).encode('utf-8') + b'\n'})
        except KeyError as e:
            print(f"解析记录失败：缺失字段 {str(e)}")
            continue

    if batch_records:
        try:
            # 用批量接口替代单条提交，适配DynamoDB触发器的多记录场景
            resp = firehose_client.put_record_batch(
                DeliveryStreamName=DELIVERY_STREAM,
                Records=batch_records
            )
            failed = resp.get('FailedPutCount', 0)
            if failed > 0:
                print(f"{failed}条记录提交失败：{resp['RequestResponses']}")
        except ClientError as e:
            print(f"Firehose调用失败：{e.response['Error']['Message']}")

def parse_dynamo_record(record):
    return {
        "orderId": record['NewImage']['orderId']['S'],
        "state": record['NewImage']['state']['S'],
        "lastUpdatedDate": record['NewImage']['lastUpdatedDate']['N']
    }

必要配置检查 #

• Lambda权限：确保执行角色包含以下权限：

  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Effect": "Allow",
              "Action": ["firehose:PutRecord", "firehose:PutRecordBatch"],
              "Resource": "arn:aws:firehose:你的AWS区域:你的账号ID:deliverystream/OrdersAuditFirehose"
          },
          {
              "Effect": "Allow",
              "Action": "dynamodb:DescribeStream",
              "Resource": "arn:aws:dynamodb:你的AWS区域:你的账号ID:table/你的表名/stream/你的流ID"
          }
      ]
  }
  

• DynamoDB触发器：确认流已启用，触发器关联正确的Lambda，批处理大小设置合理。
• Firehose投递流：确认OrdersAuditFirehose存在，目标S3桶权限允许Firehose写入。

内容的提问来源于stack exchange，提问作者BRE-ZUSES