如何统计指定时段DynamoDB经Get/BatchGet查询的唯一条目数

DynamoDB原生提供的CloudWatch默认指标仅包含请求总次数、返回条目总次数，不支持直接统计被查询的唯一条目量，可通过以下两类落地方案实现需求：

方案1：CloudTrail日志 + 流计算（无业务侵入） #

适合已经上线、不方便修改业务代码的场景，步骤如下：

• 开启CloudTrail针对目标DynamoDB表的数据事件记录：注意CloudTrail默认仅记录管控类操作，数据事件（包含GetItem、BatchGetItem请求）需要手动勾选开启，按需选择要统计的表即可，避免全量开启产生不必要成本。
• 将CloudTrail日志投递到CloudWatch Logs或Kinesis数据流，对接计算层：
  • 轻量场景直接用Lambda触发日志消费即可，大流量场景用Kinesis Data Analytics做流处理
  • 解析日志时注意：BatchGetItem请求一次会携带多个主键，需要先拆分出每个独立主键再统计；如果只统计实际查询到存在的条目，需要匹配响应内容过滤掉查询不存在的主键
• 去重统计逻辑：
  • 短窗口（比如10分钟级）、要求100%精确：按时间窗口生成统计键，将主键存入支持自动过期的去重存储（比如Redis ZSET、内存级KV），窗口到期后集合长度就是该周期的唯一条目查询量，过期后自动清理数据释放空间
  • 长窗口（比如天级）、允许2%以内误差：用HyperLogLog（HLL）概率数据结构做去重，百万级主键仅占用几KB内存，成本极低，绝大多数业务场景都够用
• 统计结果可以按需推送到CloudWatch自定义指标、监控面板做展示告警。

方案2：业务层埋点统计（精度最高、成本最低） #

适合还在迭代、可以修改业务代码的场景，是优先推荐的方案：

• 统一封装所有DynamoDB的GetItem、BatchGetItem调用逻辑，在拿到查询返回结果后，异步将实际命中存在的条目主键上报到统计管道，不会阻塞主查询流程
• 后续去重、窗口统计逻辑和方案1一致，还可以按需扩展统计维度（比如按调用接口、调用方、条目类型拆分统计），灵活度远高于日志方案。

避坑提示 #

• 不要直接用DynamoDB原生的ReturnedItems指标计算：同一个条目被查询N次，该指标会累计N次，完全无法做去重统计
• 长周期精确去重需要存储全量主键，存储和计算成本会随查询量线性上涨，非强需求不建议使用
• 开启CloudTrail数据事件会产生额外的日志存储和查询费用，上线前先估算成本

简易Lambda处理日志参考代码 #

import gzip
import json
import redis

# 初始化去重存储连接
cache = redis.Redis(host="替换为你的缓存实例地址", port=6379, db=0)
WINDOW_SECONDS = 600  # 10分钟统计窗口，可按需调整为86400对应天级窗口
EXPIRE_SECONDS = WINDOW_SECONDS + 3600  # 统计key额外保留1小时供查询

def lambda_handler(event, context):
    # 解析CloudWatch Logs投递的压缩CloudTrail日志
    cw_data = gzip.decompress(event["awslogs"]["data"])
    log_events = json.loads(cw_data)["logEvents"]
    
    for evt in log_events:
        evt_body = json.loads(evt["message"])
        evt_name = evt_body["eventName"]
        if evt_name not in ("GetItem", "BatchGetItem"):
            continue
        
        # 生成当前窗口的统计key
        window_start = (evt["timestamp"] // (WINDOW_SECONDS * 1000)) * WINDOW_SECONDS * 1000
        table = evt_body["requestParameters"]["tableName"]
        stat_key = f"dynamo_unique_query:{table}:{window_start}"
        
        # 提取所有查询主键
        pks = []
        if evt_name == "GetItem":
            pks.append(json.dumps(evt_body["requestParameters"]["key"], sort_keys=True))
        else:
            for table_req in evt_body["requestParameters"]["requestItems"].values():
                for key_obj in table_req["keys"]:
                    pks.append(json.dumps(key_obj, sort_keys=True))
        
        # 写入HLL结构做去重
        if pks:
            cache.pfadd(stat_key, *pks)
            cache.expire(stat_key, EXPIRE_SECONDS)
    
    # 取统计值时直接调用cache.pfcount(stat_key)即可得到对应窗口的唯一条目数

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