DynamoDBの変更履歴をS3に保存する仕組みをAWS CDKを使って実装してみた

DynamoDBの変更履歴をS3に保存する仕組みをAWS CDKを使って実装してみた

#AWS
#Amazon DynamoDB
#DynamoDB Streams
#TypeScript
#AWS CDK
鈴木純

鈴木純

facebook logohatena logotwitter logo
Clock Icon2023.10.05

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

DynamoDBを使っていると、何かあった時のために変更履歴を後から確認できるようにしておきたいですよね。

以下のブログのような仕組みを作ることで、Athenaから調査できる状態にしておけることを知りました。

今回はこの仕組みをAWS CDKを使って実装してみます。アーキテクチャとして新しいことはないので、CDK(Typescript)実装の参考としてください。

事前準備

cdk環境を用意するため、以下のコマンドをターミナルで実行してください。

cdk init app --language typescript

Stack

スタックのリソースは以下の通りです。

  • DynamoDBテーブル
  • S3バケット
  • Kinesis Firehose
    • IAMロール
  • Lambda
    • IAMロール
    • トリガー

import * as cdk from 'aws-cdk-lib';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as firehose from 'aws-cdk-lib/aws-kinesisfirehose';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import { Construct } from 'constructs';
import { DynamoEventSource } from 'aws-cdk-lib/aws-lambda-event-sources';

export class CdkDynamoStreamsStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // ===================================================================
    // DynamoDBのテーブル
    const sampleTable = new dynamodb.Table(this, 'SampleTable', {
      tableName: 'sample-table',
      partitionKey: {
        name: 'Id',
        type: dynamodb.AttributeType.STRING,
      },
      billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
      pointInTimeRecovery: true,
      stream: dynamodb.StreamViewType.NEW_AND_OLD_IMAGES,
    });

    // ===================================================================
    // DynamoDBの変更履歴を保存するS3バケット
    const dynamodbStreamsBucket = new s3.Bucket(this, 'DynamodbStreamsBucket', {
      bucketName: 'dynamodb-streams-bucket-'+this.account,
      versioned: true,
    });

    // ===================================================================
    // S3バケットへ出力するためのFirehose
    const dynamodbStreamsDeliveryStream = new firehose.CfnDeliveryStream(this, 'DeliveryStream', {
      deliveryStreamType: 'DirectPut',
      deliveryStreamName: 'dynamodb-streams-delivery-stream',
      extendedS3DestinationConfiguration: {
        bucketArn: dynamodbStreamsBucket.bucketArn,
        bufferingHints: {
          intervalInSeconds: 60,
          sizeInMBs: 50,
        },
        prefix: 'sample-table/',
        compressionFormat: 'UNCOMPRESSED',
        roleArn: new iam.Role(this, 'FirehoseRole', {
          assumedBy: new iam.ServicePrincipal('firehose.amazonaws.com'),
          managedPolicies: [iam.ManagedPolicy.fromAwsManagedPolicyName('AmazonS3FullAccess')],
        }).roleArn,
      },
    });

    // ===================================================================
    // Lambda関連リソース

    const dynamodbStreamsFunctionRole = new iam.Role(this, 'DynamodbStreamsFunctionRole', {
      roleName: 'dynamodb-streams-function-role',
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
      managedPolicies: [
        new iam.ManagedPolicy(this, 'dynamodb-streams-function-policy', {
          managedPolicyName: 'dynamodb-streams-function-policy',
          statements: [
            new iam.PolicyStatement({
              effect: iam.Effect.ALLOW,
              actions: ['logs:CreateLogGroup', 'logs:CreateLogStream', 'logs:PutLogEvents'],
              resources: ['arn:aws:logs:*:*:*'],
            }),
            new iam.PolicyStatement({
              effect: iam.Effect.ALLOW,
              actions: ['firehose:PutRecord'],
              resources: ['*'],
            }),
          ],
        })
      ],
    });

    const dynamodbStreamsFunction = new lambda.Function(this, 'DynamodbStreamsFunction', {
      code: new lambda.AssetCode('src'),
      handler: 'streams_to_kinesis.handler',
      runtime: lambda.Runtime.PYTHON_3_10,
      role: dynamodbStreamsFunctionRole,
      functionName: 'dynamodb-streams-function',
      environment: {
        DELIVERY_STREAM_NAME: String(dynamodbStreamsDeliveryStream.deliveryStreamName),
      },
      timeout: cdk.Duration.seconds(180),
    });
    // DynamoDBテーブルのストリームをトリガーとしてLambda関数を設定
    dynamodbStreamsFunction.addEventSource(
      new DynamoEventSource(sampleTable, {
        startingPosition: lambda.StartingPosition.TRIM_HORIZON,
      })
    );
  }
}

Lambda

LambdaのコードはPython3.10を使っています。 ルートディレクトリにsrcフォルダーを作成して、その中にstreams_to_kinesis.pyを作成します。

こちらのコードは前述したこちらのブログで記載しているものをそのまま使わせて頂きました。

# 参考：　https://dev.classmethod.jp/articles/dynamodb-streams-kinesis-firehose-athena/
import json
import os
from datetime import datetime

import boto3
from boto3.dynamodb.types import TypeDeserializer

firehose = boto3.client("firehose")
dynamodb_deserializer = TypeDeserializer()


def handler(event, context):
    for item in event["Records"]:
        timestamp = int(item["dynamodb"]["ApproximateCreationDateTime"])
        to_firehose_item = {
            "action": item["eventName"],
            "keys": deserialize(item["dynamodb"]["Keys"]),
            "new_image": None,
            "old_image": None,
            "timestamp": timestamp,
            "timestamp_utc": datetime.utcfromtimestamp(timestamp),
        }

        if "NewImage" in item["dynamodb"]:
            to_firehose_item["new_image"] = deserialize(item["dynamodb"]["NewImage"])
        if "OldImage" in item["dynamodb"]:
            to_firehose_item["old_image"] = deserialize(item["dynamodb"]["OldImage"])

        firehose.put_record(
            DeliveryStreamName=os.environ["DELIVERY_STREAM_NAME"],
            Record={"Data": json.dumps(to_firehose_item, default=json_default) + "\n"},
        )


def deserialize(item):
    """
    {'fooId': {'S': 'bar1234'}} を {'fooId': 'bar1234'} に変換する
    """
    d = {}
    for key in item:
        d[key] = dynamodb_deserializer.deserialize(item[key])
    return d


def json_default(obj):
    try:
        return str(obj)
    except Exception as e:
        print(e)
        raise TypeError("Failed convert to str.")

デプロイ・テスト

(npx) cdk deployを実行して、スタックをデプロイします。

デプロイが成功したらDynamoDBのテーブル（sample-table）が作成されているので、テスト用の項目を追加していきます。

少し待つと、S3バケットにログが出力されていることが確認できました。

中身を開くとJSON Linesの形式で保存されています。actionや対象のkey、変更前後が記録されていていい感じです。サンプルとして載せておきます。

追加

{"action": "INSERT", "keys": {"Id": "1"}, "new_image": {"Id": "1", "Age": "20", "Name": "Suzuki"}, "old_image": null, "timestamp": 1696398445, "timestamp_utc": "2023-10-04 05:47:25"}
{"action": "INSERT", "keys": {"Id": "2"}, "new_image": {"Id": "2", "Age": "20", "Name": "Tanaka"}, "old_image": null, "timestamp": 1696398463, "timestamp_utc": "2023-10-04 05:47:43"}
{"action": "INSERT", "keys": {"Id": "3"}, "new_image": {"Id": "3", "Age": "30", "Name": "Sato"}, "old_image": null, "timestamp": 1696398479, "timestamp_utc": "2023-10-04 05:47:59"}

変更

{"action": "MODIFY", "keys": {"Id": "1"}, "new_image": {"Id": "1", "Age": "40", "Name": "Suzuki"}, "old_image": {"Id": "1", "Age": "20", "Name": "Suzuki"}, "timestamp": 1696472278, "timestamp_utc": "2023-10-05 02:17:58"}

削除

{"action": "REMOVE", "keys": {"Id": "3"}, "new_image": null, "old_image": {"Id": "3", "Age": "30", "Name": "Sato"}, "timestamp": 1696472283, "timestamp_utc": "2023-10-05 02:18:03"}

このデータがあれば何かあった時Athenaで確認して、変更履歴を追うことができそうですね。

まとめ

DynamoDBテーブルへの変更履歴を保存する仕組みをAWS CDKを使って実装してみました。CDKを使う方の参考になれば幸いです。

参考

Share this article

facebook logohatena logotwitter logo

関連記事

Amazon Redshift ソートキーのテーブル最適化を有効化するビューを作成してみた

Amazon Redshift ソートキーのテーブル最適化を有効化するビューを作成してみた

User avatar
石川覚
2024.11.14
グローバルセカンダリインデックスで消費される書き込みキャパシティユニットがベーステーブルで消費される書き込みキャパシティユニットより大きいのはなぜですか

グローバルセカンダリインデックスで消費される書き込みキャパシティユニットがベーステーブルで消費される書き込みキャパシティユニットより大きいのはなぜですか

User avatar
村上紗敏
2024.11.14
AWS ChatbotをSlackと連携させて、SlackからAWS CLIを使ってみた

AWS ChatbotをSlackと連携させて、SlackからAWS CLIを使ってみた

User avatar
sora
2024.11.14
#HashiTalks Japan 2024 で「TerraformでEKSを操る実践的Tipsと効率化テクニック」という登壇をしました

#HashiTalks Japan 2024 で「TerraformでEKSを操る実践的Tipsと効率化テクニック」という登壇をしました

User avatar
かずえ
2024.11.14
Classmethod's white logo
Facebook's logo
X's logo
YouTube's logo

© Classmethod, Inc. All rights reserved.