こんにちは。最近NuPhy Air75 V2を衝動買いしたのですが、届いた直後にNuPhy Air60 V2が発売され、購入したい衝動を抑えるのに必死な@civitaspoです。バクラク事業部Platform Engineering部DevOpsグループとバクラク事業部Data &ML部Dataグループに所属しています。
この記事は、LayerXテックアドカレ2023の7日目の記事です。 昨日は@yuya-takeyamaが「Microsoft Graph API へのキーレス認証 (GitHub Actions編)」を書いてくれました。 次回はサクちゃんさんがエモい記事を書いてくれそうです。
2023年9月26日にDynamoDBがIncremental Exportをサポートしました。このIncremental Exportは「直近35日以内における特定期間の変更をChange Data Capture（CDC）としてS3にExportできる」という機能です。この機能を利用して、高頻度で継続的にIncremental Exportを実行することで、あたかもDynamoDB Streamを使ってS3にデータを出力しているような結果を作り出すことが出来ます。
この記事では、DynamoDBのIncremental Export機能を利用して、CDCのようにS3にデータを出力し続けるために調査した仕様について説明します。
なお、大事なことなので最初に書いておきますが、DynamoDBのIncremental Export機能にはいくつかの制約があり、一部のケースで正確なCDCを取得することができません。厳密にCDCを取得したい要件をお持ちの方は、DynamoDB StreamまたはKinesis Streamを用いてデータを取得することをお勧めします。
もくじ
コマンドからインターフェースについて理解する
仕様や制限に関する記述の前にDynamoDB Incremental Exportの使い方について書いておきます。AWSからコマンド提供されているので、以下のサンプルコマンドからインターフェースについて説明します。
aws dynamodb export-table-to-point-in-time \ --table-arn arn:aws:dynamodb:REGION:ACCOUNT:table/TABLENAME \ --s3-bucket BUCKET --s3-prefix PREFIX \ --incremental-export-specification ExportFromTime=1693569600,ExportToTime=1693656000,ExportViewType=NEW_AND_OLD_IMAGES \ --export-type INCREMENTAL_EXPORT
主なパラメータは以下のとおりです。
-
--table-arn：エクスポートしたいDynamoDBテーブルのARN。 -
--s3-bucket,--s3-prefix：エクスポートデータの保存先となるS3バケットとプレフィックス。 -
--incremental-export-specification：エクスポートする期間と出力のタイプ（NEW_IMAGEもしくはNEW_AND_OLD_IMAGES）を指定。
DynamoDBのテーブルから指定期間分のデータを、指定したS3の保存先に出力するのがよく分かるインターフェースですね。詳細なインターフェースは以下のリンクをご参照ください。他に、保存先のS3のKMS-SSE設定や、出力フォーマットの変更ができます。
仕様・制限について理解する
仕様や制限について細かく見ていきます。
過去35日分まで遡ってデータを出力することができる
出力データは過去35日分まで対象にすることが出来ます。この仕様はもともとDynamoDBの機能として存在していたPoint In Time Recovery（PITR）の仕様と同じです。
Exports don’t depend on each other. You can request an incremental export without having already done a full export, for example, and you can request more than one incremental export covering the same time period, so long as the start and end times are within the PITR window (which starts when you enabled PITR and goes back up to 35 days). Exports using the same time parameters will always include the same data in the output. To produce a contiguous view, you will generally want your exports to start with a full export, followed by a series of incremental exports that all share time boundaries. Introducing incremental export from Amazon DynamoDB to Amazon S3 | AWS Database Blog
一度のIncremental Exportで指定できる期間の最小幅は15分、最大幅は24時間、最大取得データサイズは100TB。Incremental Exportの最大同時実行数は300
DynamoDB Incremental Exportは15分より短い期間を指定してデータの出力ができません。そのため、CDCとしてデータをS3に保存するときもデリバリーの最短時間が15分になってしまいます。また、最大同時実行数が300なので長期間のデータを出力するときは、この制限を気にしながらオペレーションを実行する必要があります。
Incremental export: up to 300 concurrent jobs, or 100TB of table size, in an export period window between 15 minutes minimum and 24 hours maximum, can be exported concurrently. Service, account, and table quotas in Amazon DynamoDB - Amazon DynamoDB
指定期間の「始端時間以上」「終端時間より前」が出力対象となる
境界値の理解は大切なので抑えておきましょう。
The export period's start time is inclusive and the end time is exclusive. Requesting a table export in DynamoDB - Amazon DynamoDB
指定期間内でINSERTとDELETEが発生している場合、データとして出力されない。また、指定期間内でUPDATEが複数回発生している場合、中間状態のItemは出力されない
この仕様はDynamoDB Incremental Exportを利用する上で非常に大事です。この仕様があるので、DynamoDB Incremental Exportを実行するときは、可能な限り指定期間は最小幅で利用したほうが良いと考えます。また、15分より細かな粒度のCDCを必要とする場合はDynamoDB Incremental Exportは要件と合わないので、DynamoDB StreamまたはKinesis Streamを利用するべきです。
You can infer whether an item in your incremental export output was an insert, update, or delete by looking at the structure of the output. The incremental export structure and its corresponding operations are summarized in the table below for both export view types.
DynamoDB table export output format - Amazon DynamoDB
この仕様はいくつかのケースで興味深い挙動をします。Partition Keyがpk、Sort KeyをskとおいたDynamoDBテーブルを例にして説明します。エクスポートのタイプはNEW_AND_OLD_IMAGESを指定している前提でご覧ください。
ケース1：Sort Keyを変更
- pk=a, sk=a のItemを作成
- pk=a, sk=a のItemをpk=a, sk=bに変更
- 内部的にはpk=a, sk=aのItemを削除してpk=a, sk=bのItemを作成する挙動になる
このオペレーションを行って出力されるIncremental Exportの結果は以下になります。（実際はJSONLとして出力されていますが、理解しやすいように適度に整形しています。）
{ "Metadata": { "WriteTimestampMicros": { "N": "1698617280983556" } }, "Keys": {"pk": {"S": "a"}, "sk": {"S": "b"}}, "OldImage": {"pk": {"S": "a"}, "sk": {"S": "b"}}, "NewImage": {"pk": {"S": "a"}, "sk": {"S": "b"}} }
pk=a, sk=a の記録は期間内にINSERTとDELETEが発生しているため記録が存在しません。これはドキュメント通りの挙動です。一方、pk=a, sk=b の記録に関しては想像とは異なる挙動になります。pk=a, sk=b のItemはDynamoDBの内部挙動としてはINSERTであるはずですが、OldImageとNewImageの両方が存在しており、UPDATEのオペレーションを行ったように記録がされています。しかし、OldImageに記録されているItemは pk=a, sk=a ではなく、NewImageと同じ pk=a, sk=b です。
ケース2：Sort Keyを2回変更、最後の変更時にAttributeを追加
- pk=a, sk=a のItemを作成
- pk=a, sk=a のItemをpk=a, sk=bに変更
- pk=a, sk=b のItemをpk=a, sk=cに変更し、Attributeとしてv=aを追加
このオペレーションを行って出力されるIncremental Exportの結果は以下になります。
{ "Metadata": { "WriteTimestampMicros": { "N": "1698617334717531" } }, "Keys": {"pk": {"S": "a"}, "sk": {"S": "b"}}, "OldImage": {"pk": {"S": "a"}, "sk": {"S": "b"}} } { "Metadata": { "WriteTimestampMicros": { "N": "1698617334717579" } }, "Keys": {"pk": {"S": "a"}, "sk": {"S": "c"}}, "OldImage": {"pk": {"S": "a"}, "sk": {"S": "c"}}, "NewImage": {"pk": {"S": "a"}, "sk": {"S": "c"}, "v": {"S": "a"}} }
pk=a, sk=a の記録が存在しないのはケース1と同じです。pk=a, sk=b の記録はDynamoDBの内部挙動としてはINSERTとDELETEの発生であるため記録が存在しなくなるはずですが、DELETEオペレーションを行ったような記録が出力されています。また、pk=a, sk=c, v=aのItemに関しても同様にINSERTオペレーションではなくUPDATEオペレーションが発生した記録が出力されます。
ケース1とケース2からわかることはDynamoDBの内部挙動としてはDELETEとINSERTのオペレーションが発生しているとしても、元のItemが存在する場合はUPDATEオペレーションとして扱われるということです。ただし、その記録はOldImageがItem自身となり不完全な状態で記録されることに注意が必要です。
ケース3：Attributeを複数回更新
- pk=a, sk=a, v=a のItemを作成
- pk=a, sk=a, v=a のItemをpk=a, sk=a, v=bに変更
- pk=a, sk=a, v=b のItemをpk=a, sk=a, v=aに変更
このオペレーションを行って出力されるIncremental Exportの結果は以下になります。
{ "Metadata": { "WriteTimestampMicros": { "N": "1698616548761960" } }, "Keys": {"pk": {"S": "a"}, "sk": {"S": "a"}}, "OldImage": {"pk": {"S": "a"}, "sk": {"S": "a"}, "v": {"S": "a"}}, "NewImage": {"pk": {"S": "a"}, "sk": {"S": "a"}, "v": {"S": "a"}} }
Attribute v を複数回変更したケースですが、記録は1つしか存在しません。一度 v=b の状態になっているはずですが、記録上はその状態だったことが読み取れません。このことからDynamoDB Incremental Exportは指定期間における始端時刻の状態と終端時刻の状態から差を取っているようにみえます。
ここまでの内容からわかるように、DynamoDB Incremental Exportは特定のケースで変更が記録されないことがあるため、詳細な変更記録が必要なユースケースでは利用できないことを認識しておく必要があります。
出力フォーマットはJSONかAmazon Ion
特に細かく説明する必要はないと思うのでドキュメントだけ記載しておきます。
Amazon Ionについては利用事例をあまり聞きませんがAthenaでデータスキャンをすることもできるようです。
出力パスの構成は出力するデータを時系列にソートできる形式で並ばない
以下が出力パスの構成です。マニフェストファイル群とデータファイル群が別のパスに置かれる構成です。また、データファイル群はファイル名を時系列に並びません。どのファイルにいつのデータが含まれているかはマニフェストファイルを読む必要があります。もしAthenaでスキャンすることを考える場合はIncremental Export時に指定するS3パスプレフィックス内にパーティションに使うキーを入れておくように検討したほうが良いでしょう。
DynamoDB table export output format - Amazon DynamoDBDestinationBucket/DestinationPrefix . └── AWSDynamoDB ├── 01693685934212-ac809da5 // an incremental export ID │ ├── manifest-files.json // manifest points to files under 'data' outder folder │ ├── manifest-files.checksum │ ├── manifest-summary.json // stores metadata about request │ ├── manifest-summary.md5 │ └── _started // empty file for permission check ├── 01693686034521-ac809da5 │ ├── manifest-files.json │ ├── manifest-files.checksum │ ├── manifest-summary.json │ ├── manifest-summary.md5 │ └── _started ├── data // stores all the data files for incremental exports │ ├── sgad6417s6vss4p7owp0471bcq.json.gz │ ...
おわりに
この記事ではDynamoDB Incremental Exportの仕様について探ってみました。私はDynamoDB Incremental Exportの発表があったとき「S3にデータ保存するなら、これがベストプラクティスになるのかな」と思っていましたが、仕様をよく調べてみると、まだユースケースが限られることをよく理解できました。この情報がこれからDynamoDB Incremental Exportを検討する人たちの役に立つことを願っています。
実はこの記事はもともと「DynamoDB Incremental Exportを使って継続的にCDCを取得する」という名前でした。実際にStep Functionsのコードも書いていたのですが、分量の都合や記事の方向がブレるで割愛しました。第二弾の実装編として書こうと思っているので乞うご期待ください。
