如何在boto3中实现无Query、NextToken的DynamoDB全量数据分页

以下是几种符合要求的技术实现方案，各有适用场景，可根据你的数据规模和业务需求选择：

方案1：服务端全量拉取后内存分页（适合小数据集） #

核心是一次性拉取全量数据，在服务端完成分页逻辑，前端仅需传递页码参数。

实现步骤 #

1. 前端请求时携带page参数（如page=1表示第一页）。
2. 在Lambda中首次调用时执行全量扫描获取所有数据（这里为了拉全量，会用到LastEvaluatedKey迭代，但分页逻辑不依赖NextToken传递给前端）：

   import boto3
   from math import ceil
   
   dynamodb = boto3.resource('dynamodb')
   table = dynamodb.Table('your-table-name')
   
   # 全局变量缓存全量数据（利用Lambda实例复用特性，冷启动时会重新拉取）
   cached_items = []
   
   def fetch_all_data():
       global cached_items
       if not cached_items:
           all_items = []
           response = table.scan()
           all_items.extend(response['Items'])
           while 'LastEvaluatedKey' in response:
               response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
               all_items.extend(response['Items'])
           cached_items = all_items
       return cached_items
   
   def lambda_handler(event, context):
       page = int(event.get('page', 1))
       page_size = 10
       all_items = fetch_all_data()
       
       start_idx = (page - 1) * page_size
       end_idx = start_idx + page_size
       page_items = all_items[start_idx:end_idx]
       
       return {
           'statusCode': 200,
           'body': {
               'items': page_items,
               'total_pages': ceil(len(all_items) / page_size),
               'current_page': page
           }
       }
   

3. 将全量数据存入Lambda全局变量（或外部Redis缓存，避免冷启动重复拉取），根据页码计算数据段返回。

优缺点 #

• ✅ 实现简单，前端交互逻辑清晰
• ❌ 数据量大时会占用过多Lambda内存（上限10GB），且冷启动耗时较长
• ❌ 数据更新后缓存数据会不一致，需设置缓存过期机制

方案2：基于有序主键的手动范围分页（需主键有序） #

如果你的表主键（分区键/排序键）是连续有序的（如自增ID、时间戳），可以通过主键范围筛选实现分页，无需依赖NextToken。

实现步骤 #

1. 预先统计全表数据总量（可通过scan(Select='COUNT')获取，或定期同步到缓存）：

   def get_total_count():
       response = table.scan(Select='COUNT')
       return response['Count']
   

2. 前端传递page参数，服务端计算对应主键范围，用FilterExpression筛选数据：

   def get_page_data(page):
       page_size = 10
       total_count = get_total_count()
       start_id = (page - 1) * page_size + 1
       end_id = page * page_size
       
       response = table.scan(
           FilterExpression=boto3.dynamodb.conditions.Attr('id').between(start_id, end_id),
           Limit=page_size
       )
       return {
           'items': response['Items'],
           'total_pages': ceil(total_count / page_size),
           'current_page': page
       }
   

优缺点 #

• ✅ 无需拉取全量数据，内存占用低
• ✅ 数据实时性好，每次请求都是最新数据
• ❌ 依赖主键的有序连续性，若主键存在断档（如删除数据），会导致单页数据不足10条
• ❌ scan的过滤是后过滤逻辑，大数据集下扫描效率极低

方案3：预生成分页快照存储（适合大数据集） #

通过定时任务预先生成分页数据快照，存储到S3等服务，前端请求时直接返回对应快照内容，彻底避免每次扫描DynamoDB。

实现步骤 #

1. 编写定时Lambda任务（由CloudWatch Events触发），定期拉取全量数据并按页拆分：

   import boto3
   import json
   from math import ceil
   
   s3 = boto3.client('s3')
   bucket_name = 'your-s3-bucket'
   
   def generate_page_snapshots():
       all_items = fetch_all_data()  # 复用方案1的全量拉取函数
       page_size = 10
       total_pages = ceil(len(all_items) / page_size)
       
       for page in range(1, total_pages + 1):
           start_idx = (page - 1) * page_size
           end_idx = start_idx + page_size
           page_data = all_items[start_idx:end_idx]
           s3.put_object(
               Bucket=bucket_name,
               Key=f'dynamo-pages/page_{page}.json',
               Body=json.dumps(page_data),
               ContentType='application/json'
           )
       # 存储总页数信息
       s3.put_object(
           Bucket=bucket_name,
           Key='dynamo-pages/total_info.json',
           Body=json.dumps({'total_pages': total_pages}),
           ContentType='application/json'
       )
   

2. 前端请求时，Lambda根据page参数从S3读取对应文件返回：

   def lambda_handler(event, context):
       page = int(event.get('page', 1))
       try:
           response = s3.get_object(Bucket=bucket_name, Key=f'dynamo-pages/page_{page}.json')
           page_data = json.loads(response['Body'].read())
           total_info = json.loads(s3.get_object(Bucket=bucket_name, Key='dynamo-pages/total_info.json')['Body'].read())
           return {
               'statusCode': 200,
               'body': {
                   'items': page_data,
                   'total_pages': total_info['total_pages'],
                   'current_page': page
               }
           }
       except s3.exceptions.NoSuchKey:
           return {'statusCode': 404, 'body': 'Page not found'}
   

优缺点 #

• ✅ 大数据集下分页请求效率极高，无需扫描DynamoDB
• ✅ 不占用Lambda内存资源
• ❌ 数据存在延迟，无法实时获取最新数据（需根据业务设置合理的快照更新频率）
• ❌ 需要额外的S3存储资源和定时任务维护

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