DynamoDB Streamsを使ってS3にデータを自動でアーカイブしてみる

コンサルティング部の吉井です。

早速ですが皆さんはDynamoDB使われていますか？
最近はAIエージェントを組み込んだアプリも増え、軽量なデータを保持をしておく等の用途で使用される場面も増えてるんじゃないかと思います。

一方で軽いデータだし大丈夫でしょのノリで大量のデータを蓄積していくと、いずれコストの壁にぶち当たってしまうという悲しい現実もあるんじゃないかと思います。
そうならないためにもデータのライフサイクルとアーカイブ戦略を決めておくことは非常に大切です。

今回はDynamoDB Streamsを使用して、自動的にS3にデータをアーカイブする手法を試してみます。

DynamoDBのデータコスト

最初にDynamoDBのデータ関連コストを確認しておきましょう。

特筆すべき点はDynamoDBのストレージ保存コストが高めなことでしょうか。
S3と比べて役10倍の保存の金額です。

データ数が少ないPoCレベルのアプリケーションであれば、そこまでコストインパクトはないものの、ここは本番運用していくとなると気になってきそうですね。

参考：

https://aws.amazon.com/jp/dynamodb/pricing/

https://aws.amazon.com/jp/s3/pricing/

https://aws.amazon.com/jp/rds/mysql/pricing/

ライフサイクルを決めてデータをアーカイブ化しよう

データの保存コストを小さくするにあたって、データのライフサイクルは非常に大切です。
DynamoDB等を検討する場合、履歴データやログデータ等の時系列データを保存しておくことが多いのではないでしょうか。

ここで考えるべきなのはそのデータがどの程度の頻度でどのくらいの量アクセスされるのかという点です。
例えば、データ登録後1年間はそれなりに頻度でアクセスがあるが、それ以降は全くアクセスされないデータの場合、1年をデータの有効期限とし、有効期限が切れるとアーカイブ化するような仕組みを導入すると良いでしょう。

シュミレーション

具体的なユースケースを考えてみましょう。日時あたり1GBのデータをDynamoDBに保存するという前提でアーカイブの有無によってどの程度コストに差が出るかをシュミレーションしてみます。

全部DynamoDBに置き続ける

当然ですが期間が経つにつれてデータ数が増えコストが積み上がっていってしまいます。

1年経ったデータはS3にアーカイブする

続いて、DynamoDBには1年分のデータのみを保持するようにし、それ以降はS3にアーカイブするパターンを見てみましょう。

年数が経ってもDynamoDB自体に保存されているデータ量は一定のため、合計費用が大きく下がっています。

これであれば安心して運用できそうですね。

どうやってS３にアーカイブするの？

コスト感のイメージはついたので、実際にどうやってDynamoDBからS3にアーカイブしていくかを見ていきましょう。

流れとしては以下のような流れでS3へのアーカイブを実現できます。

1. レコード登録時に TTL 属性（例: deleted_at）を設定する
2. TTL 到達後、DynamoDB が非同期でレコードを削除する
3. 削除イベントが DynamoDB Streams に記録される
4. Lambda が TTL による削除イベントを処理する
5. Lambda が Amazon Data Firehose に送信する
6. Amazon Data Firehose が S3削除レコードを配信する

参考：
https://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/automatically-archive-items-to-amazon-s3-using-dynamodb-ttl.html

やってみた

まずは検証用のDynamoDBのテーブルを作っておきます。
テーブル名やPK等は任意の設定でOKです。

テーブルを作成したら

テーブル詳細画面 > 設定タブ > Time to Live(TTL) という順で遷移していき、TTLを有効化にしましょう。

TTL属性名はdeleted_atとしておきました。(任意でOK)

次にDynamoDB Streamsを有効化しておきます。

テーブル詳細画面 > エクスポートおよびストリームタブ > DynamoDB ストリームの詳細 から有効化できます。

表示タイプは最低限旧イメージが入っていればOKです。
今回は新旧イメージにしておきます。

続いて、アーカイブ先のS3バケットも作っておきます。

次に削除レコードをS3に登録するための、Data Firehoseを作リます。

ソースはLambdaから直接登録するので、Direct PUT　送信先はS3とします。

送信先の設定では先ほど作成したS3バケットを設定しておきましょう
タイムゾーンやプレフィックスの追加設定もできますので、お好みで設定してください。

最後にDynamoDB Streamsから呼ばれるLambdaを作ります。
ランタイムはPython3.14で作成しました。

続いて、下記２つのポリシーを作成し、Lambdaの実行ロールに追加しておきます。

LambdaからData Firehoseへ送信するためのポリシー

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "PutRecordsToDynamodbTtlArchiveFirehose",
      "Effect": "Allow",
      "Action": [
        "firehose:PutRecordBatch"
      ],
      "Resource": "<作成したData FirehoseのARN>"
    }
  ]
}

LambdaからDynamoDB Streamsを参照するためのポリシー

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadDynamodbStream",
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeStream",
        "dynamodb:GetRecords",
        "dynamodb:GetShardIterator",
        "dynamodb:ListStreams"
      ],
      "Resource": "＜作成したDynamoDB StreamsのARN＞"
    }
  ]
}

作成後Lambdaの実行ロールに上記で作成した2つのポリシーを紐づけておきます。

次にLambdaのコードを設定しましょう。
必要に応じて、FIREHOSE_NAME変数は作成したData Firehoseの名前に置き換えてください。

関数のデプロイも忘れずに！

import json
import os

import boto3

firehose = boto3.client("firehose")

FIREHOSE_NAME = "dynamodb-ttl-archive"
def lambda_handler(event, context):
    old_images = []

    for record in event["Records"]:
        # TTL による削除イベントだけを対象にする
        is_ttl_delete = (
            record["eventName"] == "REMOVE"
            and record.get("userIdentity", {}).get("principalId")
            == "dynamodb.amazonaws.com"
        )

        if is_ttl_delete:
            old_images.append(record["dynamodb"]["OldImage"])

    # TTL 削除が含まれていない場合は何もしない
    if not old_images:
        return

    firehose.put_record_batch(
        DeliveryStreamName=FIREHOSE_NAME,
        Records=[
            {
                "Data": json.dumps(old_images).encode("utf-8")
            }
        ],
    )

最後にDynamoDB StreamsのトリガーにLambda関数を設定すればOKです。

テーブル詳細画面 > エクスポートおよびストリームタブ > トリガー から設定できます。

ここまでできれば準備完了です。
deleted_at(TTL属性名)を含んだレコードをDynamoDBに登録し、動作確認をしましょう。

今回は期限切れのデータという体で、過去の日付をdeleted_at(TTL属性名)に設定しました。

{
  "id": {
    "S": "8c54bd9f-3575-4a65-93b3-56918bfa2b16"
  },
  "message": {
    "S": "TTL archive test"
  },
  "deleted_at": {
    "N": "1785456000"
  }
}

登録後、以下の状態になっていれば完了です！

• 登録したレコードがDynamoDBから削除されている事
• S3バケットに登録したレコードがアーカイブされている事

まとめ

今回はDynamoDB Streamsを使ってS3にデータをアーカイブしてみました。

自動化することで、運用面やコスト面でもメリットがありそうです！

参考になりましたら幸いです。