AWS Lambda生产环境监控：如何汇总查看DynamoDB处理记录及统计数量？

阿华AIGC实验室

2026-5-20

解决方案：汇总Lambda处理的DynamoDB记录 & X-Ray统计记录数

刚好我之前在生产环境处理过类似的需求，给你分享几个实用的方案，覆盖日志汇总、X-Ray追踪和实时监控三个场景：

一、用CloudWatch日志洞察做记录汇总

既然你已经在CloudWatch打印了每一步执行日志，那可以利用日志洞察的查询能力，快速汇总处理过的记录明细和数量。假设你的日志里会输出类似Processing record: abc123的内容（包含记录的唯一标识，比如DynamoDB主键），可以用下面的查询语句：

fields @timestamp, @message
| filter @message like /Processing record: [A-Za-z0-9-]+/  # 替换成你日志里的记录标识格式
| stats 
    count() as total_processed_records, 
    count_distinct(@message) as unique_records_processed 
  by bin(1h)  # 按小时聚合，也可以改成1m/1d等
| sort @timestamp desc

操作步骤：

打开CloudWatch控制台，找到你的Lambda函数对应的日志组
切换到「日志洞察」标签，粘贴上面的查询语句，选择时间区间后执行
结果会展示每个时间段内处理的总记录数，以及去重后的唯一记录数，还可以导出结果做进一步分析

二、用X-Ray添加自定义注解/元数据，统计记录数

既然你已经在用X-Ray排查耗时和错误，那可以直接在Lambda代码里添加X-Ray的自定义注解和元数据，既能追踪单次调用处理的记录数，也能在X-Ray控制台做聚合统计：

代码示例（Python）：

import boto3
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

# 初始化X-Ray
patch_all()

def lambda_handler(event, context):
    processed_count = 0
    
    # 遍历DynamoDB记录并处理
    for record in event.get('Records', []):
        # 你的业务处理逻辑
        processed_count += 1
        
        # 可选：记录每条处理的记录ID到X-Ray元数据（方便排查）
        record_id = record['dynamodb']['Keys']['id']['S']  # 替换成你的主键字段
        xray_recorder.put_metadata(
            key='processed_record_id',
            value=record_id,
            namespace='DynamoDB_Records'
        )
    
    # 添加自定义注解（可被X-Ray索引，用于聚合统计）
    xray_recorder.put_annotation('processed_records_count', processed_count)
    
    return {
        'statusCode': 200,
        'body': f"Successfully processed {processed_count} records"
    }

怎么在X-Ray里查看统计：

打开X-Ray控制台，进入「服务地图」找到你的Lambda函数
点击函数进入「追踪」页面，选择时间区间后，在「分析」标签里可以添加统计维度：
- 选择「聚合依据」为processed_records_count，就能看到每次调用处理的记录数分布
- 还可以结合调用次数、错误率等指标，分析记录数和性能的关联

三、自定义CloudWatch指标，做实时监控

如果需要在CloudWatch仪表盘实时展示处理记录数，或者设置告警（比如处理记录数骤降/骤升），可以在Lambda里发送自定义指标：

代码示例（Python）：

import boto3
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

patch_all()
cloudwatch = boto3.client('cloudwatch')

def lambda_handler(event, context):
    processed_count = 0
    for record in event.get('Records', []):
        # 业务处理逻辑
        processed_count += 1
    
    # 发送自定义指标到CloudWatch
    cloudwatch.put_metric_data(
        Namespace='Lambda/DynamoDB_Processor',
        MetricData=[
            {
                'MetricName': 'Processed_Records_Count',
                'Dimensions': [
                    {
                        'Name': 'FunctionName',
                        'Value': context.function_name
                    }
                ],
                'Value': processed_count,
                'Unit': 'Count'
            }
        ]
    )
    
    # 同时添加X-Ray注解
    xray_recorder.put_annotation('processed_records_count', processed_count)
    
    return {
        'statusCode': 200,
        'body': f"Processed {processed_count} records"
    }