如何从分页DynamoDB表获取指定记录，分Lambda处理不同分页？

核心逻辑是利用DynamoDB扫描返回的LastEvaluatedKey作为分页标记，将第一页处理与后续页面处理拆分到两个Lambda中，通过状态存储传递分页标记实现衔接。

1. 处理第一页的Lambda代码 #

该Lambda负责获取并处理第一页数据，判断是否存在后续页面，若存在则存储分页标记并触发第二个Lambda。

import boto3
import json

# 用于存储分页标记的状态表（需提前创建）
STATE_TABLE_NAME = "pagination_state"
TARGET_TABLE_NAME = "你的目标表名"

def lambda_handler(event, context):
    dynamo_client = boto3.client('dynamodb')
    paginator = dynamo_client.get_paginator("scan")

    # 初始化分页器，可指定PageSize强制第一页最多50条
    page_iterator = paginator.paginate(
        TableName=TARGET_TABLE_NAME,
        PageSize=50  # 可选，按需设置每页最大条目数
    )

    # 获取第一页数据
    first_page = next(page_iterator)
    first_page_items = first_page["Items"]

    # 这里替换为你的第一页数据处理逻辑
    for item in first_page_items:
        print("处理第一页条目:", item)

    # 检查是否有后续页面
    last_evaluated_key = first_page.get("LastEvaluatedKey")
    if last_evaluated_key:
        # 存储分页标记到状态表
        state_client = boto3.client('dynamodb')
        state_client.put_item(
            TableName=STATE_TABLE_NAME,
            Item={
                "task_id": {"S": context.aws_request_id},
                "last_key": {"M": last_evaluated_key},
                "target_table": {"S": TARGET_TABLE_NAME}
            }
        )

        # 异步触发处理后续页面的Lambda
        lambda_client = boto3.client('lambda')
        lambda_client.invoke(
            FunctionName="你的后续处理Lambda名称",
            InvocationType="Event",
            Payload=json.dumps({"task_id": context.aws_request_id})
        )

    return {
        "status": "第一页处理完成",
        "processed_count": len(first_page_items),
        "has_more_pages": bool(last_evaluated_key)
    }

2. 处理第二页及以后的Lambda代码 #

该Lambda读取存储的分页标记，从第二页开始扫描并处理所有后续页面，完成后清理状态标记。

import boto3
import json

STATE_TABLE_NAME = "pagination_state"

def lambda_handler(event, context):
    task_id = event["task_id"]
    state_client = boto3.client('dynamodb')
    dynamo_client = boto3.client('dynamodb')

    # 获取分页标记
    state_response = state_client.get_item(
        TableName=STATE_TABLE_NAME,
        Key={"task_id": {"S": task_id}}
    )
    if "Item" not in state_response:
        return {"status": "无有效分页任务"}

    last_key = state_response["Item"]["last_key"]["M"]
    target_table = state_response["Item"]["target_table"]["S"]

    # 从分页标记位置开始扫描后续页面
    paginator = dynamo_client.get_paginator("scan")
    for page in paginator.paginate(
        TableName=target_table,
        ExclusiveStartKey=last_key
    ):
        page_items = page["Items"]
        # 这里替换为你的后续页面数据处理逻辑
        for item in page_items:
            print("处理后续页面条目:", item)

        # 更新分页标记或清理状态
        current_last_key = page.get("LastEvaluatedKey")
        if current_last_key:
            state_client.put_item(
                TableName=STATE_TABLE_NAME,
                Item={
                    "task_id": {"S": task_id},
                    "last_key": {"M": current_last_key},
                    "target_table": {"S": target_table}
                }
            )
        else:
            # 无后续页面，删除状态记录
            state_client.delete_item(
                TableName=STATE_TABLE_NAME,
                Key={"task_id": {"S": task_id}}
            )
            break

    return {"status": "后续页面处理完成"}

关键注意事项 #

• 分页标记存储：示例用DynamoDB存储状态，也可根据需求换成S3、SQS或参数存储，只要能在两个Lambda间传递LastEvaluatedKey即可。
• PageSize设置：若需严格限制第一页为50条，必须在paginate中指定PageSize=50，否则DynamoDB默认按1MB数据量返回对应条目数。
• 触发方式：除了Lambda异步调用，还可使用SQS队列传递任务，第一个Lambda发送含分页标记的消息到队列，第二个Lambda监听队列处理。
• 断点续传：若后续处理可能中断，建议每处理一页就更新状态表的LastEvaluatedKey，确保故障后可从断点继续。

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