DynamoDB と Amazon EventBridge の統合

DynamoDB デベロッパーガイドの「ストリームの有効化」セクションの手順に従って、ソーステーブルで DynamoDB Streams を有効にします。ソーステーブルで DynamoDB Streams がすでに有効になっている場合は、コンシューマーの数が 2 つ以下であることを確認します。コンシューマーには、Lambda 関数、DynamoDB グローバルテーブル、Amazon DynamoDB と Amazon OpenSearch Service のゼロ ETL 統合、または DynamoDB Streams Kinesis Adapter などのストリームから直接読み取るアプリケーションなどが含まれます。

「EventBridge ユーザーガイド」の「Creating an Amazon EventBridge event bus」セクションの手順に従って EventBridge イベントバスを作成します。

「EventBridge ユーザーガイド」の「Creating an Amazon EventBridge pipe」セクションの手順に従って EventBridge パイプを作成します。

ソース DynamoDB テーブルにサンプル項目を書き込み、イベントをトリガーします。これにより、EventBridge はサンプル項目からスキーマを推測できます。このスキーマを使用して、イベントをルーティングするためのルールを作成できます。例えば、属性のオーバーロードを伴う設計パターンを実装する場合、ソートキーの値に応じて異なるルールをトリガーできます。DynamoDB への項目の書き込み方法の詳細については、「DynamoDB デベロッパーガイド」の「項目と属性の操作」を参照してください。

「Lambda 開発者ガイド」の「Python による Lambda 関数の構築」セクションの手順に従ってターゲットとして使用する Python Lambda 関数の例を作成します。関数を作成する際は、以下のサンプルコードを使用して統合のデモを行えます。コードが呼び出されると、イベントで受信した NewImage と OldImage を表示し、これは CloudWatch Logs で確認することができます。

import json

def lambda_handler(event, context):
    dynamodb = event.get('detail', {}).get('dynamodb', {})
    new_image = dynamodb.get('NewImage')
    old_image = dynamodb.get('OldImage')
    
    if new_image:
        print("NewImage:", json.dumps(new_image, indent=2))
    if old_image:
        print("OldImage:", json.dumps(old_image, indent=2))
    
    return {'statusCode': 200, 'body': json.dumps(event)}

「EventBridge ユーザーガイド」の「Creating Amazon EventBridge rules that react to events」セクションの手順に従って、新しい Lambda 関数にイベントをルーティングする EventBridge ルールを作成します。

これで、「EventBridge ユーザーガイド」の「Starting or stopping schema discovery on event buses」セクションの手順に従って、イベントバスでのスキーマの検出を停止することができるようになります。

ソース DynamoDB テーブルに 2 つ目のサンプル項目を書き込み、イベントをトリガーします。イベントが各ステップで正常に処理されたことを確認します。