如何用Boto3分批扫描DynamoDB并获取无重复的第101-200条记录

场景：

• 从包含20万条记录的DynamoDB表中分批检索数据
• 第一次请求获取100条记录
• 第二次请求获取接下来的100条记录（需排除第一次已获取的）

现有实现代码：

scan_kwargs=None
if scan_kwargs is None:
    scan_kwargs = {}
complete = False

while not complete:
    try:
        
        response = table.scan(Limit=10000, **scan_kwargs,
            FilterExpression=Key('timestamp').between(dateFrom, dateTo) 
                ) 
    except botocore.exceptions.ClientError as error:
        raise Exception('Error')
    next_key = response.get('LastEvaluatedKey')
    scan_kwargs['ExclusiveStartKey'] = next_key
    complete = True if next_key is None else False

    if response['Items']:
        for record in response['Items']:
            print(record)
            totalRecords = totalRecords + 1
            if totalRecords > 100:
                break
    if totalRecords > 100:
        break

当前代码仅能获取前100条记录，无法实现获取第101至200条记录的需求，求可行的实现示例。

现有代码的核心问题是：获取完前100条后直接终止扫描流程，且未保留扫描的终止位置（LastEvaluatedKey），导致无法从上次中断的地方继续获取下一批数据。

要实现分批获取，需做到：

1. 保留每次扫描的LastEvaluatedKey，作为下一次扫描的起始位置（通过ExclusiveStartKey参数传入）
2. 封装可复用的分页获取逻辑，支持指定批次大小、起始位置和过滤条件
3. 注意：DynamoDB的Limit参数是过滤前的记录数，若使用FilterExpression，一次扫描返回的过滤后记录可能不足指定批次大小，需循环扫描直到凑够数量或遍历完所有数据

实现示例 #

import boto3
from botocore.exceptions import ClientError
from boto3.dynamodb.conditions import Key

def get_dynamodb_batch(table, batch_size, start_key=None, date_from=None, date_to=None):
    """
    从DynamoDB表中获取指定批次的记录
    :param table: DynamoDB表对象
    :param batch_size: 要获取的记录数量
    :param start_key: 上次扫描的LastEvaluatedKey，首次调用传None
    :param date_from: 时间范围起始值
    :param date_to: 时间范围结束值
    :return: (获取到的记录列表, 下一次扫描的LastEvaluatedKey)
    """
    scan_kwargs = {
        'Limit': 1000,  # 每次扫描请求的原始记录数，可根据过滤条件调整
        'ExclusiveStartKey': start_key
    }
    
    if date_from and date_to:
        scan_kwargs['FilterExpression'] = Key('timestamp').between(date_from, date_to)
    
    collected_items = []
    next_key = start_key
    
    while len(collected_items) < batch_size:
        try:
            response = table.scan(**scan_kwargs)
        except ClientError as e:
            raise Exception(f"DynamoDB扫描错误: {e.response['Error']['Message']}")
        
        # 添加过滤后的记录到结果
        collected_items.extend(response['Items'])
        next_key = response.get('LastEvaluatedKey')
        
        # 如果没有更多记录，终止循环
        if not next_key:
            break
        
        # 更新下一次扫描的起始键
        scan_kwargs['ExclusiveStartKey'] = next_key
    
    # 只返回指定批次大小的记录，多余的截断（若不需要严格返回batch_size可去掉此步）
    return collected_items[:batch_size], next_key

# 初始化DynamoDB表
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('你的表名')
dateFrom = '2024-01-01'
dateTo = '2024-01-31'

# 第一次获取前100条
first_batch, last_key = get_dynamodb_batch(table, 100, date_from=dateFrom, date_to=dateTo)
print(f"获取到第1-100条记录，共{len(first_batch)}条")
for record in first_batch:
    print(record)

# 第二次获取接下来的100条（从上次的LastEvaluatedKey开始）
second_batch, next_last_key = get_dynamodb_batch(table, 100, start_key=last_key, date_from=dateFrom, date_to=dateTo)
print(f"\n获取到第101-200条记录，共{len(second_batch)}条")
for record in second_batch:
    print(record)

关键说明 #

• get_dynamodb_batch函数封装了分页逻辑，会自动循环扫描直到凑够指定批次的记录或遍历完所有数据
• 每次调用后返回的last_key必须保存，作为下一次调用的start_key参数，确保从上次中断的位置继续扫描
• 调整Limit参数：如果过滤条件较严格，可适当调大Limit减少请求次数；反之则调小，节省带宽
• 若不需要严格返回batch_size条记录（比如最后一批不足），可去掉collected_items[:batch_size]的截断逻辑

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