DynamoDB+boto3分页异常：每次请求仅返回4条重复数据

问题描述 #

使用boto3操作DynamoDB表时，分页查询特定所有者关联的视频数据遇到异常：预期每次请求获取10条数据，直至取完该所有者的所有视频，但当前实现每次仅返回4条数据，且每次返回的都是同一组视频。

代码概述 #

def get_videos_for_owner_id(owner_id, video_id=None, batch_size=10):    

    if video_id is None:
        # call repeatedly for pagination
        page_args = {'Limit': batch_size}
        all_items = []
        num_calls = 0
        while True:
            res = my_table.query(
                KeyConditionExpression='owner_id = :xx',
                Select='SPECIFIC_ATTRIBUTES',                
                ProjectionExpression='video_id,events,last_update,preview,StateMachineArn,video_metadata_key,video_status,files,video_settings',
                ExpressionAttributeValues={':xx':  owner_id},
                **page_args
            )
            num_calls += 1
            
            all_items.extend(res.get('Items'))
            
            if res.get('LastEvaluatedKey', None) is None or len(all_items) >= batch_size:
                break
            else:
                page_args = {'ExclusiveStartKey': res['LastEvaluatedKey'], 'Limit': batch_size}
    else:
        res = my_table.get_item(Key={
              'owner_id': owner_id,
              'video_id': video_id
            })
        all_items = res.get('Item', None)
        if all_items is None:
            return []
        all_items = [all_items]

    if len(all_items) > 0:
        res = [process_flattened_record(r) for r in all_items]
    return res

问题分析 #

1. 循环终止逻辑错误：当前代码用len(all_items) >= batch_size作为终止条件，会导致还未取完所有数据就提前停止。比如若单次查询返回4条，总条数未达10时本应继续分页，但如果该条件误判（或数据实际不足），会直接终止循环。
2. 分页参数传递验证缺失：若LastEvaluatedKey未正确生成或传递，会导致每次查询都从初始位置开始，返回重复数据。
3. 主键结构可能存在问题：若表仅以owner_id作为唯一分区键（无排序键），DynamoDB内部排序逻辑可能导致分页异常，但这种情况较少见。

排查步骤 #

• 验证数据总量：通过AWS控制台或scan操作（注意性能）确认该owner_id下实际的视频数据条数，若确实只有4条，则属于正常情况。
• 打印分页关键参数：在循环内打印res.get('LastEvaluatedKey')，确认每次查询后是否返回有效的起始键；同时打印page_args，验证ExclusiveStartKey是否正确传递。
• 检查数据处理函数：确认process_flattened_record是否存在修改数据的逻辑，导致返回结果看似重复。

解决方案 #

根据需求场景，提供两种修正方案：

方案1：一次性返回该所有者的所有视频（内部分页查询） #

去掉总条数限制的终止条件，仅当无更多数据时停止循环：

def get_videos_for_owner_id(owner_id, video_id=None, batch_size=10):    
    if video_id is None:
        page_args = {'Limit': batch_size}
        all_items = []
        while True:
            res = my_table.query(
                KeyConditionExpression='owner_id = :xx',
                Select='SPECIFIC_ATTRIBUTES',                
                ProjectionExpression='video_id,events,last_update,preview,StateMachineArn,video_metadata_key,video_status,files,video_settings',
                ExpressionAttributeValues={':xx':  owner_id},
                **page_args
            )
            
            all_items.extend(res.get('Items', []))
            
            last_key = res.get('LastEvaluatedKey')
            if not last_key:
                break  # 无更多数据，终止循环
            page_args = {'ExclusiveStartKey': last_key, 'Limit': batch_size}
    else:
        res = my_table.get_item(Key={
              'owner_id': owner_id,
              'video_id': video_id
            })
        all_items = res.get('Item', None)
        if all_items is None:
            return []
        all_items = [all_items]
    
    return [process_flattened_record(r) for r in all_items] if all_items else []

方案2：支持分页调用，每次返回一批10条 #

修改函数参数，允许传入上次的起始键，实现分批获取数据：

def get_videos_for_owner_id(owner_id, video_id=None, batch_size=10, exclusive_start_key=None):    
    if video_id is None:
        page_args = {'Limit': batch_size}
        if exclusive_start_key:
            page_args['ExclusiveStartKey'] = exclusive_start_key
            
        res = my_table.query(
            KeyConditionExpression='owner_id = :xx',
            Select='SPECIFIC_ATTRIBUTES',                
            ProjectionExpression='video_id,events,last_update,preview,StateMachineArn,video_metadata_key,video_status,files,video_settings',
            ExpressionAttributeValues={':xx':  owner_id},
            **page_args
        )
        
        items = res.get('Items', [])
        last_key = res.get('LastEvaluatedKey')
        
        processed_items = [process_flattened_record(r) for r in items] if items else []
        return processed_items, last_key
    else:
        res = my_table.get_item(Key={
              'owner_id': owner_id,
              'video_id': video_id
            })
        item = res.get('Item')
        if not item:
            return [], None
        return [process_flattened_record(item)], None

调用示例：

# 首次调用
current_items, last_key = get_videos_for_owner_id('target_owner_id')
# 循环获取所有数据
while last_key:
    next_items, last_key = get_videos_for_owner_id('target_owner_id', exclusive_start_key=last_key)
    current_items.extend(next_items)

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