テーブルの変更を、リアルタイムできっかけにする

手順に迷ったときや、用語の意味を確かめたいときに開きましょう。

※ リンク切れの場合は、ページタイトルで検索してください。

テーブルへの変更が DynamoDB Streams によって検知され、Lambda 関数を起動して CloudWatch Logs へ出力するまでの流れです。3 つの要素が一列につながるパイプラインです。

以下の要件を満たし、テーブルの変更が Lambda 関数を経由して CloudWatch Logs に記録される一連の流れを確認してください。

def lambda_handler(event, context):
    for record in event["Records"]:
        event_name = record["eventName"]
        keys = record["dynamodb"].get("Keys", {})
        print(f"変更の種類: {event_name}")
        print(f"対象のキー: {keys}")

        if "OldImage" in record["dynamodb"]:
            print(f"変更前: {record['dynamodb']['OldImage']}")
        if "NewImage" in record["dynamodb"]:
            print(f"変更後: {record['dynamodb']['NewImage']}")
    return {"processedCount": len(event["Records"])}

「テーブルを作る → Streams を有効化する → Lambda を作る → トリガーをつなぐ → 変更して確かめる」の順に進めます。

1. マネジメントコンソールにサインインし、リージョンを合わせる

   ブラウザで AWS マネジメントコンソールにサインインし、画面右上のリージョンが「アジアパシフィック（東京）ap-northeast-1」になっていることを確認します。

2. DynamoDB コンソールでテーブルを作成する

   DynamoDB コンソールの「テーブルの作成」を開き、テーブル名とパーティションキー名を自由に決めて作成します（例：テーブル名 StreamDemoTable、パーティションキー ItemId）。

   設定はデフォルトのままでよい

   テーブルクラスや読み書きの設定は、今回はデフォルトのままで問題ありません。

3. テーブルの「エクスポートとストリーム」タブから、DynamoDB Streams を有効化する

   作成したテーブルを開き、「エクスポートとストリーム」タブを選びます。「DynamoDB streams の詳細」の「有効にする」を押し、ビュータイプとして「新規および旧イメージ（新旧の両方のイメージ）」を選択して有効化します。

   ビュータイプの選び方

   ビュータイプには「新しいイメージのみ」「古いイメージのみ」「新旧両方のイメージ」「キーのみ」などの選択肢があります。今回は変更前後の両方を見たいため、「新旧両方のイメージ」を選びます。

4. Lambda 関数を作成する

   Lambda コンソールで「関数の作成」→「一から作成」を選びます。関数名を自由に決め、ランタイムはPythonの任意のバージョンを選んで作成します。コードエディタの中身を、上記の要件セクションに掲載したコードに置き換え、「Deploy」ボタンを押して保存します。

5. Lambda 関数の「トリガーを追加」から、DynamoDB Streams を接続する

   作成した Lambda 関数の画面で「トリガーを追加」を選び、トリガーの種類として「DynamoDB」を選びます。DynamoDB テーブルの欄で手順 2 のテーブルを指定し、トリガーを作成します。

   権限はコンソールが自動で設定してくれる場合が多い

   このトリガーを追加すると、Lambda 関数の実行ロールに DynamoDB Streams を読み取るための権限が、コンソール側で自動的に追加される場合が多いです。

6. DynamoDB の「項目を探索」から、新しい項目を 1 件追加する

   テーブルの「項目を探索」タブを開き、「項目を作成」から、パーティションキーの値と、任意の属性（例：Status = new）を持つ項目を 1 件追加します。

7. 追加した項目の属性を更新し、続けて削除する

   同じ項目を開き、先ほど追加した属性の値を変更して保存します（更新）。その後、同じ項目を選んで削除します。

   3 種類の変更を意図的に発生させる

   追加・更新・削除という 3 種類の変更を順番に発生させることで、それぞれに対応するログの違いを見比べられるようにします。

8. CloudWatch Logs でログを確認する

   CloudWatch コンソールの「ロググループ」から、Lambda 関数に対応するロググループ（/aws/lambda/関数名）を開きます。最新のログストリームを開き、追加・更新・削除それぞれに対応するログが出力されていることを確認します。変更の種類（イベント名）、変更前/変更後の内容に注目しましょう。

初学者がよく引っかかる箇所を先回りでまとめました。答えそのものは載せていませんが、「どこを見直せばよいか」の手がかりとして使ってください。

要件の再確認ではなく、画面のどこを見れば達成を確認できるかをまとめました。各コンソールを開いて、次を順に確かめましょう。

• テーブルの「エクスポートとストリーム」タブで、DynamoDB Streams が有効になっている。
• Lambda 関数の「トリガー」欄に、DynamoDB が表示されている。
• CloudWatch Logs に、3 種類（追加・更新・削除）のログが見つかる。
• 各ログで、変更の種類（INSERT / MODIFY / REMOVE）が読み取れる。
• 更新のログで、変更前・変更後の値が両方確認できる。

手を動かすことに加えて、次の問いに自分の言葉で答えられるようにしておくと、理解がより深まります。

1. 定期的にテーブルを確認する方法と比べて、Streams を使う方法はどんな場面で特に効果を発揮するでしょうか。
   ヒント

   変更の発生頻度が低く、かつ変更からの反応の速さが重要な場面を考えてみましょう。定期確認では、確認の間隔を短くするほど無駄な処理が増え、間隔を長くするほど反応が遅れるというトレードオフがあります。
2. 同時に複数の項目を変更したら、Lambda は何回起動しそうでしょうか（1 件ずつか、まとめてか、という観点で考えてみましょう）。
   ヒント

   今回のコードは event["Records"] をループで処理しています。これは、1 回の Lambda の起動で複数件の変更レコードがまとめて渡される場合があることを意味します。「変更 1 件 = 起動 1 回」とは限らない、という前提で考えてみましょう。
3. 今回はログに出力するだけでしたが、変更内容をもとに別のテーブルへ反映したり、通知を送ったりする処理に発展させるとしたら、何に注意が必要そうでしょうか。
   ヒント

   同じ変更レコードが何らかの理由で複数回処理される可能性や、処理に失敗したときの再試行の挙動を考えてみましょう。「1 回だけ確実に処理される」とは限らない前提で、繰り返し実行されても結果が変わらないような処理の作り方が手がかりになります。

作ったものを片づけるところまでが一連の流れです。次の順で進めると迷いにくいです。

1. Lambda 関数を削除する：Lambda コンソールの関数一覧から、作成した関数を選び、削除します。
2. トリガーが解除されていることを確認する：Lambda 関数を削除すると、DynamoDB Streams のトリガーも合わせて解除されます。
3. DynamoDB テーブルを削除する：DynamoDB コンソールのテーブル一覧から、作成したテーブル（例：StreamDemoTable）を選び、削除します。
4. CloudWatch Logs のロググループを削除する：/aws/lambda/関数名 のロググループを CloudWatch Logs コンソールから削除します。