AWS Glue を使用して DynamoDB テーブルエクスポートをフィルタリング、変換、ロードする

本記事は 2026 年 05 月 12 日に公開された “Filter, transform, and load your DynamoDB table exports using AWS Glue” を翻訳したものです。翻訳は Solutions Architect の嶋田 朱里が担当しました。

この記事では、Amazon DynamoDB のフルまたはインクリメンタルテーブルエクスポートを 2 つ目の DynamoDB テーブルにロード（インポート）する方法を紹介します。何をロードするか、どの書き込みレートでロードするかを正確にコントロールでき、進捗を観察する機能も備えています。この技術は、最大限の制御が必要な大規模なデータマイグレーションや同期を推進するのに役立ちます。ここで説明するメカニズムは、オープンソースの Bulk Executor for DynamoDB のコマンドとして提供されています。これは、DynamoDB テーブルに対してバルクコマンドを実行するためのユーティリティセットです。このコマンドを使用することで、DynamoDB のエクスポート（済みデータ）を細かく制御しながらロードでき、アイテムを変換したり、データをフィルタリングしたり、コマンドのオプションの transform パラメータを使用して宛先テーブルに何を配置するかをカスタマイズしたりできます。

背景

DynamoDB はフルおよびインクリメンタル両方のテーブルエクスポートをサポートしています。両タイプのエクスポートはサービス駆動型で、読み込みキャパシティを消費せず、AWS マネジメントコンソール、コマンドライン、または SDK を通じて開始できます。エクスポートを実行するには、テーブルでポイントインタイムリカバリ (Point in Time Recovery, PITR) を有効にする必要があります。

フルエクスポートには、指定された時点でのテーブル内のすべてのアイテムが含まれます。エクスポートプロセスは、データを一連のオブジェクトとして Amazon Simple Storage Service (Amazon S3) に送信します。ファイルは DDB-JSON フォーマットまたは Ion フォーマットで、1 行に 1 アイテムずつ記録されます。インクリメンタルエクスポートのフォーマットは少し異なります。指定された期間中に変更されたアイテムのみが含まれ、各アイテムのプライマリキー、指定された期間内の最終変更時刻、（オプションで）期間開始時のアイテムの変更前イメージ、期間終了時のアイテムの変更後イメージが含まれます。アイテムの変更前と変更後の両方のイメージを確認できるため、エクスポートされた期間中にそのアイテムがどのように変化したかを把握できます。

ネイティブ機能として、DynamoDB は、S3 にフルエクスポートされたデータを新しいテーブル（ロードの過程で作成されます）に取り込む機能をサポートしています。AWS は、ロード中に処理された圧縮されていないデータの GB 単位でこの機能を価格設定しています。この記事で説明する技術は、この機能を補完し、ロードレート制御、カスタムアイテム操作、フィルタリングされたアイテムロード、既存テーブルの進捗追跡の機能を提供します。

なお、DynamoDB には現在、インクリメンタルエクスポートに基づいてデータをロードするネイティブ機能はありません。ここで説明する技術を使用すると、フルエクスポートと同じ柔軟性でこれらのファイルをロードできます。

使用方法

Bulk Executor は、”load-export” コマンドを使用してフルおよびインクリメンタルエクスポートされたデータのロードをサポートしています。インストール手順では、ツールのセットアップ方法を説明しています。セットアップ後の使用方法は次のとおりです。

./bulk load-export --table <target> --s3-path <s3://bucket/path/to/data> [--transform <transform_module>]

<target> を宛先テーブルの名前に置き換えます。このテーブルは既に存在している必要があります。空でもデータがあっても構いません。ロードはプライマリキーが一致する既存のアイテムを上書きします。

s3-path パラメータには、エクスポートされたデータを含む Amazon S3 のロケーションへのフルパスを指定します（例: s3://<bucket-name>/prod/AWSDynamoDB/01716790307109-5f9d6aaa）。エクスポートされたデータは、フルまたはインクリメンタルエクスポートのいずれかです。データフォーマットは自動的に検出され処理されます。

オプションの transform パラメータは、各アイテムをロードする前に加工を加えるためのユーザー定義ロジックを含む Python ファイルの名前を受け取ります。ロードが各アイテムを処理する際、フルロードかインクリメンタルロードかに応じて、Python ファイルの transform_full_record(FullExportRecord) または transform_incremental_record(IncrementalExportRecord) 関数を呼び出します。この関数内で、以下を返すことができます。

1. レコード: 調整なしでアイテムをロード（または削除）
2. 空のリスト: このアイテムのロード（または削除）を抑制
3. 変更されたレコード: 調整付きでアイテムをロード
4. 複数のアイテムのリスト: 1 つのソースレコードから複数のアイテムを書き込むためにファンアウト

この関数は完全にカスタマイズ可能なので、これから行うアクションのあらゆる側面を操作できます。transform モジュールコードのブートストラップに関する詳細情報については、リポジトリ内の readme ファイルを参照してください。

AWS Glue ジョブは、任意の大きさのデータセットを処理するために並列実行でロードを実行し、標準の Bulk Executor のレート制限パラメータを介してレート制限が利用可能です。

ユースケース

フルおよびインクリメンタルエクスポートをロードする機能は、さまざまなユースケースをサポートします。次のセクションでは、2 つの一般的なユースケースを概説します。

独立したテーブル同期

一般的なユースケースは、グローバルテーブルとそのアクティブ-アクティブ伝播を使用せずに、一連のインクリメンタルエクスポートをロードして、2 つのテーブルを切り離された一方向の方法で同期することです。これは、ステージングまたは開発テーブルを本番と一致させたり、サードパーティのテーブルコピーを自分のものと同期させたり、隔離された環境内のテーブルコピーを維持したりするのに有用です。通常、毎回 2 つ目のテーブルを完全に作成するよりも、インクリメンタルエクスポートを適用する方がコスト効率と時間効率が高くなります。

以前の記事では、継続的なデータ保持を推進する方法について説明しています。

データの変換

もう 1 つの一般的なユースケースは、変換機能を使用して、あるテーブルから別のテーブルにデータを移動する際にデータをフィルタリングまたは変更することです。

例えば、フィルタリング関数を使用して、フルエクスポートを実行し、属性駆動型のフィルタリングで変換を行うことで、あるテーブルから別のテーブルにデータの一部をコピーできます。ロード中に各アイテムの属性を評価し、そのアイテムが 2 つ目のテーブルに伝播するかどうかを判断できます。SaaS (software as a service) 企業で、複数のテナントを持つテーブルがあり、1 つのテナントを切り出したい場合などに使用できます。これを実現するには、そのテナントのデータのみをロードします。または、最近のデータのみを保持する新しいテーブルのコピーが必要な場合は、最近のタイムスタンプを持つアイテムのみを許可する変換でロードを実行できます。

別の用途は、修飾子としての使用です。SSN や DOB 属性などの個人を特定できる情報 (Personally Identifiable Information, PII) をダウンストリームテーブルにロードする際にロードを抑制できます。また、タイムスタンプ属性を ISO 8601 文字列から数値エポック形式に調整することもできます。ファンアウト関数を使用すると、単一のアイテムから値のリストを個々の値の個別アイテムに分割できます。キースキーマを調整することもできます。

transform 関数を使用して、新しいキースキーマに移行することもできます。例えば、エクスポートされたデータで last_name と first_name がそれぞれパーティションキー (PK) とソートキー (SK) として含まれている場合などです。宛先テーブルには、PK として user_id という新しいフィールドがあります。transform 関数を使用すると、ロード中に各アイテムに GUID を user_id として正常に注入し、他の既存の属性を維持できるため、新しいキースキーマに効果的に移行できます。

技術的な内部構造

load-export コマンドは、Bulk Executor フレームワークを使用して実行されます。Bulk Executor は差し込み式のコマンドをサポートしているため、最初に load-export コマンドをカスタムコマンドとして作成し、その後公式リポジトリに貢献しました。

検証

ロードプロセスが開始される前に、コードはマニフェストに基づいてエクスポート全体を検証し、すべてのファイルが存在し、その署名がマニフェストと一致することを確認します。これにより、エクスポートデータに偶発的に生じた破損を防ぎます。

読み込み

ロードプロセスは AWS Glue を使用して、Amazon S3 オブジェクトを Spark Resilient Distributed Dataset (RDD) にロードし、フルまたはインクリメンタルエクスポートかどうかを検出します。

変換

transform 機能は、2 つの主要な関数を公開します。1 つはフルロード中にレコードを変換するためのもので、もう 1 つはインクリメンタルロードのためのものです。現在処理中のソース Amazon S3 レコードがこの関数にパラメータとして渡されます。このレコードはすべての内部データを辞書型で公開するため、ロードプロセスを完全に制御できます。次の例は、status 属性が active 値を持つアイテムのみを含める方法を示しています。

def transform_full_record(record: FullExportRecord) -> list[FullExportRecord]:
    """
    Example: Only load items where 'status' attribute is 'active'.

    Args:
        record: record.item is the deserialized Item dict,
                record.table_key_schema has key info.

    Returns:
        list[FullExportRecord]: Single-element list to keep, empty list to skip
    """
    if record.item.get("status") == "active":
        return [record]
    return []

def transform_incremental_record(record: IncrementalExportRecord) -> list[IncrementalExportRecord]:
    """
    Example: Only load items where 'status' attribute is 'active'.

    Behavior:
        - If new_image exists and status is 'active': load the item (PUT)
        - If new_image exists but status is not 'active': skip the item
        - If new_image is None (a delete): return the record, i.e. respect the delete

    Args:
        record: record.keys, record.new_image, record.old_image,
                record.table_key_schema, record.write_timestamp_micros

    Returns:
        list[IncrementalExportRecord]: Single-element list to keep, empty list to skip
    """
    if record.new_image:
        if record.new_image.get("status") == "active":
            return [record]
        else:
            return []
    return [record]

transform コードは完全にあなたの管理下にあります。通常、効率のために単純なアルゴリズムチェックを実行させますが、レイテンシーを許容できる場合は、決定を下しながら CRM や別のデータベースにクエリを実行するなど、外部ソースをクエリすることもできます。

並列化とレート制限

次に、コマンドはデータを AWS Glue ワーカーに分散して、DynamoDB に並列で書き込みます。各 AWS Glue スロットは独立したライターとして動作します。この大量並列化がテーブルのスループット容量を超えるのを防ぐために、Bulk Executor の分散レート制限機能を使用します。これにより、設定可能な読み込みまたは書き込みキャパシティを使用できます。

検証

ロードが完了した後、bulk diff ツール機能を使用して2 つのテーブルを比較することで、データの整合性を検証できます。両方のテーブルが一致する場合、次のような出力が表示されます。

./bulk diff --table source --table2 destination
…
No differences found

コストに関する考慮事項

ロード中にコストを左右する主な要因は、次の 2 つです。

1. DynamoDB テーブルの書き込み消費: 書き込まれるアイテムの数とサイズに基づきます。
2. AWS Glue データ処理ユニット (Data Processing Unit, DPU): AWS Glue ジョブ中のワーカーの数、ワーカーのタイプ、実行時間に基づきます。

次の実行は、1 億レコードを保持するフルエクスポートに対するフルロードのテスト実行を示しています。デフォルトのワーカータイプ (G.1X) と デフォルトの AWS Glue ワーカー数 (220 までオートスケーリング) を使用しました。ロード速度を上げるために、毎秒 240,000 書き込みユニットで事前ウォームアップされたオンデマンドテーブルを作成しました。bulk ツールに 240,000 すべてを使用するよう指示しました。総実行時間は 12 分 27 秒でした。

>> ./bulk load-export \
	--XMaxWriteRate 240000 \
	--table ... \
	--s3-path s3://... \
	--XTimeout 10080 \
	--XWaitForDPU
…
Destination Table: …
S3 export: 100,000,000 items across 512 files (FULL_EXPORT, DYNAMODB_JSON)
DynamoDB write costs depend on how many items are being written and the size of the items.
Here we estimate the command will write 100,000,000 items
 with average size 256 bytes;
 each write incurs an average of 1 write units
Write units required (approx): 100,000,000
This does not include costs for secondary indexes!
Approx DynamoDB cost for on-demand writes consuming 100,000,000 WRUs (using us-east-1 prices): $69.50
Writing items to DynamoDB...
===============================================================
JOB COMPLETED SUCCESSFULLY
  - Total items in export: 100,000,000
  - Total items written: 100,000,000
  - Execution time: 707.3 seconds
…
Waiting 40 seconds for DPU metrics to gather...
Job completed successfully. Job duration: 0:12:27 (24.06 DPU hours)

実行では、ロードのサイズに基づいて DynamoDB のコストを早期に見積もり、ここでは小さなアイテムの約 1 億回の書き込みに対して $69.50 となっています。出力には最後に AWS Glue のコストが表示されます。us-east-1 で 1 DPU 時間あたり $0.44 の価格で、約 $10 です。

制限事項

load-export コマンドは DDB-JSON フォーマットを使用したエクスポートのみをサポートし、ロードは同じプライマリキーを持つ既存のアイテムを上書きします。異なる動作が必要な場合は、コードを変更できます。

まとめ

この記事では、Bulk Executor を使用して DynamoDB のフルおよびインクリメンタルエクスポートをバルクロードする方法を概説しました。オプションの変換機能を使用して、ロードするデータとその構造を思いどおりに選択でき、組み込みのレート制限が書き込みキャパシティの飽和を防ぎます。Bulk Executor のリポジトリをクローンして、今日最初のロードを実行することから始めましょう。

著者について