本文介绍如何在 dynamodb 中避免低效分页查询，通过流式聚合（dynamodb streams + lambda）+ 聚合表模式，实现毫秒级获取指定 filename 的事件总数。
本文介绍如何在 dynamodb 中避免低效分页查询，通过流式聚合（dynamodb streams + lambda）+ 聚合表模式，实现毫秒级获取指定 filename 的事件总数。
在 DynamoDB 中，当需要高频、低延迟地统计某类数据（如“某文件关联的事件总数”）时，直接对 GSI 执行 Query 并遍历所有分页结果（LastEvaluatedKey 循环）不仅性能差（尤其在百万级数据下响应达数秒），还会显著增加读取容量消耗（RCU）和成本。你当前的 GSI 设计（filename 为分区键、eventId 为排序键）虽支持按文件名查询，但 DynamoDB 不原生支持 COUNT 聚合操作——Query 返回的是完整项，即使只取 Select: COUNT，底层仍需扫描并计数所有匹配项，无法跳过数据加载。
✅ 正确解法：将“实时计数”从查询时计算，转变为写入时维护。即采用“写时更新聚合表”的事件驱动架构：
核心架构：DynamoDB Streams + Lambda + 计数表
- 启用 DynamoDB Stream：在源表（含 fileName, BrandCode, eventId）上开启 NEW_IMAGE 类型流；
- Lambda 消费流事件：监听新增/删除项，提取 fileName，并原子更新专用计数表（如 FileNameCountTable）；
-
计数表设计：
{ "fileName": "ABC", // 分区键（PK） "count": 2 // 当前事件总数（Number 类型） } - 查询端极致优化：客户端只需一次 GetItem（强一致性可选），毫秒返回结果。
示例：Lambda 计数更新逻辑（Python）
import boto3
dynamodb = boto3.resource('dynamodb')
count_table = dynamodb.Table('FileNameCountTable')
def lambda_handler(event, context):
for record in event['Records']:
if record['eventName'] == 'INSERT':
filename = record['dynamodb']['NewImage']['fileName']['S']
# 原子自增：不存在则初始化为1
count_table.update_item(
Key={'fileName': filename},
UpdateExpression='ADD #cnt :inc',
ExpressionAttributeNames={'#cnt': 'count'},
ExpressionAttributeValues={':inc': 1},
ReturnValues='UPDATED_NEW'
)
elif record['eventName'] == 'REMOVE':
filename = record['dynamodb']['OldImage']['fileName']['S']
count_table.update_item(
Key={'fileName': filename},
UpdateExpression='ADD #cnt :dec',
ExpressionAttributeNames={'#cnt': 'count'},
ExpressionAttributeValues={':dec': -1},
ConditionExpression=boto3.dynamodb.conditions.Attr('count').gt(0) # 防负数
)关键注意事项
- ✅ 幂等性保障：Lambda 函数需具备重试容错能力，建议在 UpdateItem 中加入 ConditionExpression（如检查 count >= 0）或使用 UpdateExpression 的 ADD 操作（天然幂等）；
- ✅ 冷启动与吞吐：Lambda 并发需匹配写入峰值；计数表应启用按需模式或预置足够 WCU；
- ⚠️ 最终一致性：计数存在极短延迟（通常
- ? 避免反模式：切勿在应用层缓存计数后定期刷新（易失效）、也不要在 GSI 上建二级索引试图“加速 COUNT”——DynamoDB 无索引级聚合能力。
该方案将 O(N) 查询复杂度降为 O(1) 读取，同时保持线性可扩展性。对于日均千万写入的场景，实测 GetItem 平均延迟
