【DynamoDB】ScanでLastEvaluatedKeyを使用して大量データを取得するサンプル

DynamoDBには全件を取得するScanという機能があります。
しかし、これには1回で取得できる件数に制限があります。
その場合は、何回かに分けてScanリクエストをDynamoDBに投げてやる必要があります。
今回はそのような状況の時に使える分割リクエストのサンプルを紹介していきます。

前提知識：LastEvaluatedKey

Scanリクエストは、全件を取得できなかったときに、LastEvaluatedKeyという値を返します。
これは、今回のScanリクエストでどこまでのデータを取得したかを表す値を示します。
言い換えると、これを使用することで次回のScanリクエストで前回の続きからデータを取得することができます。

具体的なプログラム

では、この仕組みをプログラムに落とし込むとどのようなソースコードになるのでしょうか。
今回は、pythonを例に見ていきます。

DynamoScanサンプル

import botocore
import boto3
from boto3.dynamodb.conditions import Key, Attr
dynamodb = boto3.resource('dynamodb')
dynamoClient = boto3.client('dynamodb')
dynamoPaginator = dynamoClient.get_paginator('scan')

# メイン処理
def main()
    # Scanリクエスト実行
    dynamoDBName = 'yourDynamoDBName'
    scanRes = scan_dynamo(dynamoDBName, None)
   
    # 結果セットが0件の場合、終了する
    if(scanRes['Count'] <= 0):
        print('DynamoDBのScan結果が０件のため処理を終了します。TableName={0}'.format(dynamoDBName))
        return noticeCount

    # LastEvaluatedKeyの取得
    lastEvaluatedKey = None
    if('LastEvaluatedKey' in scanRes):
        lastEvaluatedKey = scanRes['LastEvaluatedKey']
    
    # 結果セット分だけループ
    for r in scanRes['Items']:
        # 行いたい処理
    
    # LastEvaluatedKeyが存在すれば再リクエストして続きを取得する
    if lastEvaluatedKey is not None:
        scanRes2 = scan_dynamo(dynamoDBName, lastEvaluatedKey)
        
    # 結果セット分だけループ
    for r in scanRes2['Items']:
        # 行いたい処理


# DynamoDBを全件Scan
def scan_dynamo(dynamoDBName, lastEvaluatedKey=None):
    
    # パラメータ設定
    paginationConfig={
        'MaxItems': 100,
        'PageSize': 100,   # ページネーション設定(今回はページネーションしないので取得件数とそろえている)
    }
    
    # Scan実行
    try:
        # lastEvaluatedKeyあり
        if(lastEvaluatedKey):
            print('lastEvaluatedKeyあり。lastEvaluatedKey={0}'.format(lastEvaluatedKey))
            paginator = dynamoPaginator.paginate(
                TableName=dynamoDBName,
                PaginationConfig=paginationConfig,
                ExclusiveStartKey=lastEvaluatedKey
            )
        
        # lastEvaluatedKeyなし
        else:
            print('lastEvaluatedKeyなし')
            paginator = dynamoPaginator.paginate(
                TableName=dynamoDBName,
                PaginationConfig=paginationConfig
            )
    except Exception as e:
        print('DynamoDBのScanに失敗しました。TableName={0}, Error={1}'.format(dynamoDBName, e))
    
    # ページを取得(今回は1ページだけなので先頭のものを取得するのみ)
    for r in paginator:
        return r

初回リクエストを投げて、LastEvaluatedKeyがあれば、再リクエストして続きを取得しています。
2回で全件取得できるとは限らないので、実際は2回目以降がループになるかと思います。