Python Lambda访问DynamoDB遇容量不足报错，求代码排查

阿华AIGC实验室

2026-5-20

排查DynamoDB读容量报警及Lambda代码优化建议

嘿，作为AWS和Python新手遇到这种问题确实有点懵，我来帮你一步步拆解这个报警问题，顺便优化你的代码逻辑～

首先，先搞懂这个报警的含义：INSUFFICIENT_DATA Assets-ReadCapacityUnitsLimit-BasicAlarm 通常是CloudWatch还没收集到足够的数据点来确认是否触发阈值，但后面跟着的ConsumedReadCapacityUnits >= 240 for 5 datapoints within 5 minutes说明你的DynamoDB表读容量消耗已经接近或超过了预设上限，大概率是代码里的DynamoDB操作逻辑有问题，或者表的容量设置不合理。

下面是具体的排查和优化步骤：

1. 先检查DynamoDB表的核心配置

登录AWS控制台找到你的DynamoDB表，查看「容量」标签：
- 如果是预置模式（Provisioned），看看你设置的读容量单位（RCU）是不是低于240？如果是，那当请求量上来时很容易打满容量触发报警。
- 新手更推荐用按需模式（On-Demand），这种模式下DynamoDB会自动根据请求量扩容，不用手动调RCU，虽然成本略高，但能避免容量不足的麻烦。

2. 检查Lambda代码中的DynamoDB操作逻辑

这是最可能出问题的地方，新手常犯的错误是用全表扫描来判断用户是否存在，这会消耗巨量RCU：

错误示例（别这么写！）

# 用scan()全表扫描找用户，表数据多的时候会消耗大量RCU
response = table.scan(
    FilterExpression=Attr('slack_user_id').eq(user_id)
)

正确做法（精确查询）

你应该把Slack的user_id设为DynamoDB表的分区键，然后用get_item()做精确查询，这样消耗的RCU极少（通常只占1个RCU）：

# 用主键精确查询用户，效率高、耗量低
response = table.get_item(
    Key={
        'slack_user_id': user_id  # 这里的键要和你的表分区键名一致
    }
)

如果你的表还没把slack_user_id设为分区键，赶紧去修改表结构——这是优化DynamoDB性能的核心。

3. 验证读容量消耗的真实情况

去CloudWatch控制台找到这个报警对应的指标，查看ConsumedReadCapacityUnits的实际数值：

如果数值确实经常超过240，那要么是请求量真的很大，要么就是代码里有低效的读操作（比如刚才说的全表扫描）。
如果数值很低却收到报警，可能是CloudWatch刚开始收集数据，还没攒够5个有效数据点，这种情况过段时间会自动消失。

4. 给代码加日志辅助排查

在Lambda代码里添加日志，打印每次DynamoDB操作消耗的容量，这样能直观看到问题出在哪：

import boto3
import logging
from datetime import datetime

# 配置日志
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 初始化DynamoDB资源
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('你的表名')

def lambda_handler(event, context):
    # 解析Slack请求中的用户ID（根据你的实际事件结构调整）
    user_id = event['event']['user']  # 这里要对应Slack发送的事件字段
    
    # 查询用户是否已存在
    get_response = table.get_item(Key={'slack_user_id': user_id})
    if 'ConsumedCapacity' in get_response:
        logger.info(f"查询用户消耗RCU: {get_response['ConsumedCapacity']['CapacityUnits']}")
    
    if 'Item' not in get_response:
        # 添加用户到DynamoDB
        put_response = table.put_item(
            Item={
                'slack_user_id': user_id,
                'joined_at': datetime.now().isoformat()
            }
        )
        if 'ConsumedCapacity' in put_response:
            logger.info(f"添加用户消耗RCU: {put_response['ConsumedCapacity']['CapacityUnits']}")
        return {"text": "已成功把你加入数据库！"}
    else:
        return {"text": "你已经在数据库里啦～"}