DynamoDBさわってみた（DynamoDB Stream, TTL）

Posted at 2024-09-07

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;
};