DynamoDB Streamの有効化

設定状況の確認

設定状況はマネコンの一般的な情報 > 追加情報から確認できます。

有効化

「エクスポートおよびストリーム」タブから設定します。

streamにpushする情報を選択できます。
トリガーされるlambdaがeventとして処理できる情報になります。
今回は「新旧イメージ」を選択します。

ストリームが有効化されます。
また、ストリームそのもののARNも定義されます。
トリガーに設定するlambdaのサービスロールには、dynamodb本体ではなく、このstreamへのアクセスポリシーが必要です。

トリガーの追加

lambdaサービスロールの準備

トリガーで実行されるlambdaのサービスロールを設定します。

上記ガイドを参考にします。

policy.json

{
    "Statement": [
        {
            "Action": [
                "logs:PutLogEvents",
                "logs:CreateLogStream"
            ],
            "Effect": "Allow",
            "Resource": "arn:aws:logs:ap-northeast-1:123456789012:log-group:/aws/lambda/lambda:*"
        },
        {
            "Action": [
                "dynamodb:ListStreams",
                "dynamodb:GetShardIterator",
                "dynamodb:GetRecords",
                "dynamodb:DescribeStream"
            ],
            "Effect": "Allow",
            "Resource": [
                "arn:aws:dynamodb:ap-northeast-1:123456789012:table/example/stream/2024-09-07T11:19:49.079/*"
            ]
        }
    ],
    "Version": "2012-10-17"
}

`Cannot access stream` と注意されてしまう場合

※バグなのか、2024/09/07時点ではドキュメントを参考にした上記ポリシーではトリガー作成時に以下エラーがでてしまいます

原因ですが、IAMポリシーのリソースブロックに ${Stream_Arn} が存在しない場合、上記エラーがでてしまいます。

ドキュメントにある通り、許可を与える対象は ${Stream_Arn} ではなく、 ${Stream_ARN}/* なのですが、エラー扱いにされてしまいます。

policy.json

            "Resource": [
                "arn:aws:dynamodb:ap-northeast-1:123456789012:table/example/stream/2024-09-07T11:19:49.079",
                "arn:aws:dynamodb:ap-northeast-1:123456789012:table/example/stream/2024-09-07T11:19:49.079/*"
            ]

今回は、上のようにポリシーを設定し、トリガー作成完了後、不要な ${Stream_Arn} のほうを削除しました。

lambdaの作成

トリガーで実行されるlambda

受け取ったevent情報内のレコードをそのままログ出力するlambdaを作成しました。

index.js

exports.handler = async (event, context) => {
    console.log("========================================");
    console.log(event);
    event.Records.forEach((record, index) => {
        console.log(index);
        console.log('%o', record);
    });
    console.log("========================================");

    return {
        "statusCode": "200",
        "body": JSON.stringify({ "test": "value" })
    };
};

DynamoDBに書き込むlambda

5件のレコードをINSERTするlambdaを作成しました。

index.mjs

import { BatchWriteItemCommand, DynamoDBClient } from "@aws-sdk/client-dynamodb";

const client = new DynamoDBClient({});

export const handler = async (event, context) => {
    const command = new BatchWriteItemCommand({
        RequestItems: {
            example: [
                {
                    PutRequest: {
                        Item: {
                            id: { N: "1" },
                            name: { S: "White Chocolate Chip" },
                        },
                    },
                },
                {
                    PutRequest: {
                        Item: {
                            id: { N: "2" },
                            name: { S: "White Chocolate Chip" },
                        },
                    },
                },
                {
                    PutRequest: {
                        Item: {
                            id: { N: "3" },
                            name: { S: "White Chocolate Chip" },
                        },
                    },
                },
                { PutRequest: {
                        Item: {
                            id: { N: "4" },
                            name: { S: "White Chocolate Chip" },
                        },
                    },
                },
                {
                    PutRequest: {
                        Item: {
                            id: { N: "5" },
                            name: { S: "White Chocolate Chip" },
                        },
                    },
                },
            ],
        },
    });

    const response = await client.send(command);
    console.log(response);
    return response;
};

トリガーの設定

DynamoDBのマネコンまたはlambdaのマネコンから設定できます。

DynamoDBからは「トリガー」という機能で表現されています。
lambda側からは「イベントソースマッピング」という機能として表現されます。SNSとかと同じです。

今回はlambdaから設定してみます。

トリガー追加を選択します。

トリガーの設定を埋めていきます。
動作確認のために、バッチサイズを2にしておきます。
追加ボタンをクリックします。

イベントソースマッピングが設定されました。

動作確認

レコードをINSERTするlambdaを実行します。
DynamoDBに5件レコードが追加され、トリガーに設定されたlambdaが5件分のレコードをログ出力してくれるはずです。

cloudwatchを確認します。
ログストリームが2つできてます。

片方のログを確認してみます。
トリガーされた関数が、1度の実行で、2件分のレコードを受け取っているのが確認できます。
バッチサイズ設定が有効に働いていることが確認できました。

他方のログを確認してみます。
残る3つのレコードが2回に分けてログ出力されています。

バッチサイズを上回るレコードのCRUDが発生したときのトリガーの挙動が確認できました。
バッチサイズ上限を守りながら、全てのレコードを並列でトリガー先へ通知するようです（lambdaのログストリームが2つできたことから、並列実行されたことが類推できます）。

良い感じの機能ですね！！

TTL

TTL機能も試してみます。
セッションなどの一時データを保存する場合、あると便利な機能ですね。

設定状況の確認

設定状況はマネコンの一般的な情報 > 追加情報から確認できます。

レコードの準備

動作確認のため、ttl属性に適当なUNIX時間形式のNumberを設定したレコードを2件追加します。
一件目は1時間前を指しています。2件目は遠い未来ですね。

有効化

「追加の設定」タブから設定します。

中段にありました。

「オンにする」をクリックすると、設定画面が表示されます。
TTL属性名を入力し、プレビューを実施します。

1件削除候補が表示されました。2件目のレコードは遥か未来なので、動作は大丈夫そうです。

「TTLをオンにする」をクリックします。

設定が有効化されていることを確認しました。

少し待つと、先ほどTTL機能の削除候補になっていたレコードが削除され、グラフに表示されていました。

やったー！

おわりに

DynamoDBの主力機能のうち2つを実際にさわってみました。
たのしかったです。