DynamoDBの変更データキャプチャを試してみた

# 背景・目的
DynamoDB（以降、DDBといいます。）の変更データキャプチャ（CDC）について、触る機会があったので知識の整理と、簡単に試してみます。

# まとめ
下記に特徴を整理します。

|特徴| 説明
|:--|:--|
|概要|・DDBでは、テーブルに保存された項目の変更を、変更の発生時にキャプチャできる<BR>・CDCレコードのニアリアルタイムをサポートしている|
|ストリーミングオプション|・DDB用のKDS<BR>・DynamoDB Streams|

# 概要
[Amazon DynamoDB の変更データキャプチャ](https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/streamsmain.html)を元に整理します。

> 多くのアプリケーションでは、DynamoDB テーブルに保存された項目の変更を、変更の発生時にキャプチャすることで利点を利用できます。以下に示しているのは、いくつかのユースケースの例です。

> - 人気のモバイルアプリケーションは、1 秒あたり数千件の更新速度で、DynamoDB テーブルのデータを変更します。別のアプリケーションがこれらの更新に関するデータをキャプチャして保存し、 near-real-time モバイルアプリの使用状況メトリックを提供します。
> - 金融アプリケーションは、DynamoDB テーブル内の株式市場データを変更します。parallel 実行されるさまざまなアプリケーションが、これらの変化をリアルタイムで追跡し、株価の動きに基づいてポートフォリオを計算し value-at-risk、自動的に再調整します。
> - 輸送車両や産業機器のセンサーは、DynamoDB テーブルにデータを送信します。さまざまなアプリケーションがパフォーマンスをモニタリングし、問題が検出されたときにメッセージングアラートを送信し、機械学習アルゴリズムを適用して潜在的な欠陥を予測し、データを圧縮して Amazon Simple Storage Service (Amazon S3) にアーカイブします。
> - アプリケーションは、友人の 1 人が新しい画像をアップロードするとすぐに、グループ内のすべての友人のモバイルデバイスに通知を自動送信します。
> - 新しいお客様がデータを DynamoDB テーブルに追加します。このイベントにより、新しいお客様にようこそメールを送信する別のアプリケーションが起動されます。

> DynamoDB は、項目レベルの変更データキャプチャレコードのストリーミングをほぼリアルタイムでサポートします。これらのストリーミングを使用し、内容に基づいてアクションを実行するアプリケーションを構築できます。


- DDBでは、テーブルに保存された項目の変更を、変更の発生時にキャプチャできる
- CDCレコードのニアリアルタイムをサポートしている

## 変更データキャプチャのストリーミングオプション
DDBには、2つのストリーミングもでるがあります。
下記に特徴をまとめます。

|プロパティ|DDB 用 KDS | DDB Streams|
|:--|:--|:--|
|データ保持期間|1年まで|24H|
|KCLサポート|KCL 1.x、2.X| KCL 1.X|
|コンシューマー数|シャードごとに最大5つの同時コンシューマー<BR>または、ファンアウトが強化されたシャードごとに最大20の同時コンシューマ|シャードごとに最大2つの同時コンシューマ|
|スループットクォーター|無制限|DDBテーブルとリージョンごとのスループットクォーターの対象|
|レコードの配信モデル|ファンアウト拡張でPush|HTTP経由でPull|
|レコードの順序|各ストリーミングレコードのタイムスタンプ属性を使用して、DDB テーブルで変更が発生した実際の順序を特定できる。|DDBテーブルで変更された各項目について、ストリーミングレコードは項目に対する実際の変更と同じ順序で出現する|
|重複レコード|表示される場合がある|表示されることはない|
|処理オプション|Lambda、Flink、KDF、Glueストリーミング|Lambda、DDB Stream Kinesis adapter|
|耐久性|AZにより中断することなく、自動でF/Oできる|同左|

# 実践
今回、このような構成で試してみます。
CDCのストリーミングオプションは、2つありますが、今回はKDSを使用します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/e0af8001-47ee-8fb1-a05d-9dd76341345e.png)


## S3バケットの作成
1. AWSにサインインします
1. S3バケットを作成します

## Kinesis Data Streamsの作成
1. Kinesisのトップページに移動します
1. 「Kinesis Data Stream」を選択し、データストリームを作成をクリックします
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/59cdfb8a-ffba-3ea5-e9b9-d56d85830423.png" width=300 />

1. データストリーム名を指定し、「データストリームの作成」をクリックします
    <img src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/74e42432-75d1-dff1-2504-dca4bee5c354.png" width=500 />


## DynamoDBの作成
### テーブルの作成
以前、[DynamoDBを試してみた](https://qiita.com/zumax/items/ef8a5c855161f9e4ba92)でDDBを作成しましたが、同じようにテーブルを作成します。

1. DDBのトップページに遷移します
1. 「テーブルの作成」をクリックします
1. 下記を入力し、「テーブルの作成」をクリックします
    - テーブル名
    - PKとソートキー
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/32db3257-afe4-dd76-a7f1-b7d89f800212.png)

### ストリームの作成
今回の本命である、変更データキャプチャを設定します。

1. 上記で作成したDDBテーブルを選択します
1. 「エクスポートおよびストリーム」タブをクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/fc680e9d-72f9-1707-b816-db68261eb59f.png)

1. KDSの「オンにする」をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/e3c1cd0d-efd1-d21b-45df-e2319a164efd.png)

1. すでに作成したKDSを指定し、「ストリームをオンにする」をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/7aa82160-7b9d-5e63-c271-6ab8ae05f86f.png)

## Data Firehoseの作成
1. Kinesisのトップページに移動します
1. 「Amazon Data Firehose 」を選択し、Firehoseストリームを作成をクリックします
1. ソースと送信先を選択で、下記を選択します
    - ソース：Kinesis Data Streamｓ
    - 送信先：S3
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/03f762c6-e6e5-862b-f346-a4718d74de7c.png)
1. ソースの設定で、作成したKDSを指定します
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/84a780e8-3849-6ad5-d02e-ddf1836b6b03.png)
1. Firehoseストリーム名を指定します
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/283591e4-247c-3572-6e10-02ddb6484dc6.png)

1. 送信先を設定します
    - S3バケット:作成したS3バケットのパス
    - タイムゾーン：Asia/Tokyo
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/3faa58d2-694d-424b-6a3d-117065ce6f29.png)

1. バッファヒント等で、サイズは1MiB、バッファ間隔を150秒にしました。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/4d2afe6f-07d1-cc0e-5c25-8e8999ab903b.png)

1. 最後に「Firehoseストリームを作成」をクリックします

## データの書き込み〜S3の確認
### データの登録
1. 作成したDDBテーブルを選択し、「項目を探索」をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/919511eb-2877-f226-8e70-f4d20ff2157e.png)
1. 項目を作成をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/27047677-a84b-2fa1-9932-a53dfecce9b4.png)

1. 値を入力し、「項目を作成」をクリックします
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/6be8db1e-3dff-91b3-8257-5c664ee0e0a3.png)
1. 再度繰り返し、下記のようになりました
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/59c74c63-bb52-08b4-ac8d-a9930f398a2e.png)

### KDSの確認
1. 分かりづらいですが、レコードは受信したようです
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/dcead742-6e71-415e-d2aa-51471687ed0a.png)

### Firehouseの確認
1. レコードは受信され、S3への配信も成功したようです
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/6e1a9ad2-bdc2-c318-0bdf-72e9c8b8a22d.png)

### S3の確認
1. ファイルが到着していました
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/b113ca72-5838-b4e6-16c0-3a35bef21c58.png)

1. S3 Selectで表示してみます。下記のようなデータが表示できました。たしかにとうろく
    ```
    {
      "awsRegion": "ap-northeast-1",
      "eventID": "6d42f935-f293-40fb-bf3e-7ae9065dd5c6",
      "eventName": "INSERT",
      "userIdentity": null,
      "recordFormat": "application/json",
      "tableName": "streamtest",
      "dynamodb": {
        "ApproximateCreationDateTime": 1712969103175023,
        "Keys": {
          "id": {
            "S": "1"
          },
          "timestamp": {
            "S": "20240413094400"
          }
        },
        "NewImage": {
          "id": {
            "S": "1"
          },
          "timestamp": {
            "S": "20240413094400"
          }
        },
        "SizeBytes": 52,
        "ApproximateCreationDateTimePrecision": "MICROSECOND"
      },
      "eventSource": "aws:dynamodb"
    }
    {
      "awsRegion": "ap-northeast-1",
      "eventID": "e6d41715-de17-4599-bf22-d6216751651c",
      "eventName": "INSERT",
      "userIdentity": null,
      "recordFormat": "application/json",
      "tableName": "streamtest",
      "dynamodb": {
        "ApproximateCreationDateTime": 1712969128680631,
        "Keys": {
          "id": {
            "S": "2"
          },
          "timestamp": {
            "S": "202404130944500"
          }
        },
        "NewImage": {
          "id": {
            "S": "2"
          },
          "timestamp": {
            "S": "202404130944500"
          }
        },
        "SizeBytes": 54,
        "ApproximateCreationDateTimePrecision": "MICROSECOND"
      },
      "eventSource": "aws:dynamodb"
    }
    ```


# 考察
今回、DDBのCDCを使って、S3までデータを連携してみました。ニアリアルタイムで、S3に連携できるのはよいですね。
次回以降は、FirehoseのTransformについて試してみようと思います

# 参考
https://docs.aws.amazon.com/ja_jp/amazondynamodb/latest/developerguide/streamsmain.html

https://qiita.com/zumax/items/ef8a5c855161f9e4ba92