はじめに

この記事はミロゴス Advent Calendar 2023 2日目の投稿です。

弊社ではデータをDynamoDBにストアすることが多く、そのデータを活用するにあたって、CSVファイルをS3に置くためのLambdaを実装、定期的に実行しています。今回、DDBのS3エクスポート機能を使うことで実装をなくし、さらに定期的な実行とそのステータス管理のためにStep Functionsで処理をしていきます。

ステートマシンの完成図は下記のような形です。

DDBからS3へのエクスポート

LambdaでDDBのデータを出力するにはテーブルに対して、ScanやQueryを行う必要があります。これらにはRead Capacity Unitが消費されるため、コストやパフォーマンス、可用性に影響が出てしまいます。それに対して、DDBのS3エクスポート機能はRead Capacityを消費せずに利用できます。

このS3エクスポート機能を利用するには下記の条件、制限があります。

DDBのテーブルにPoint In Time Recoveryの設定が有効化されていること。
出力する形式はサポートされているもののみに制限される。
- CSVはサポートされていません（2023/11/27現在）。

さらに今回はStep Functionsで実行するため、ステートマシンがS3エクスポートを実行できるように下記の権限がStep FunctionsのIAM Roleに必要です。

s3:AbortMultipartUpload
s3:PutObject
s3:PutObjectAcl

サポートされている出力形式

S3エクスポート機能では出力の形式としてCSVはなく、DynamoDB JSONかAmazon Ionになります。今回は出力したデータをTreasureDataなど複数のシステムが利用することを考慮し、Amazon固有のIonは辞め、JSON Linesで定義される.json.gzで出力されるDynamoDB JSONを選択しました。なお、.jsonl.gzではなく、.json.gzになります。

また、出力内容に関しては全量データと増分データがサポートされています。増分データのサポートは2023/09に新たにサポートされました。

この投稿では全量データをDynamoDB JSONの形式で出力していきます。

なお、増分データの出力の場合、出力のFromとToを指定する必要があります。
Step Functionsだけではそれらのデータを保持できないので、別途DDBなどのデータストアを利用する作りを考える必要があります。DDBを使った方法は下記が参考になるかと思います。

DynamoDB Incremental Export with Step Functions

エクスポートの実行

コンソールからの操作も可能ですが、今回はAWS CLIを使います。

aws dynamodb export-table-to-point-in-time \
--table-arn [エクスポートするDDBテーブルのARN] \
--s3-bucket [エクスポート先のバケット名] \
--export-format DYNAMODB_JSON --export-type FULL_EXPORT \
--s3-prefix [バケットオブジェクトのプレフィックス]

{
    "ExportDescription": {
        "ExportArn": "arn:aws:dynamodb:ap-northeast-1:xxxxxxxxxxxx:table/test/export/xxxxxxxxxxxxxx-xxxxxxxx",
        "ExportStatus": "IN_PROGRESS",
        "StartTime": "2023-11-27T15:07:58.449000+09:00",
        "TableArn": "arn:aws:dynamodb:ap-northeast-1:xxxxxxxxxxxx:table/test",
        "TableId": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
        "ExportTime": "2023-11-27T15:07:58.449000+09:00",
        "ClientToken": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
        "S3Bucket": "hoge.bucket",
        "S3Prefix": "aaa",
        "S3SseAlgorithm": "AES256",
        "ExportFormat": "DYNAMODB_JSON"
    }
}

export-formatオプション以降は任意です。設定しない場合はデフォルトでフォーマットはDYNAMODB_JSON、エクスポートタイプはFULL_EXPORTになります。
また、s3-prefixを指定することでエクスポート先のバケットに対して、オブジェクトのプレフィックスを切ることができます。
ただし、下記のエラーが出るため指定するプレフィックスは3文字以上の必要があります。

An error occurred (ValidationException) when calling the ExportTableToPointInTime operation: Invalid Request: 1 validation error detected: Value 'aa' at 's3Prefix' failed to satisfy constraint: Member must have length greater than or equal to 3

エクスポート処理の確認

エクスポート処理はDescribeExport APIで処理を確認できます。

aws dynamodb describe-export --export-arn [ExportArn] | jq .ExportDescription.ExportStatus

"COMPLETED"

ステータスはIN_PROGRESS（処理中）と COMPLETED（完了）、 FAILED（失敗）があります。

定期的にエクスポートする

Step FunctionsはSDK統合によって、多くのAWSサービスをサポートしています。DDBからS3へのエクスポート機能もこの形でサポートされています。

サービス統合には３パターンありますが、DDBのエクスポート処理はAWS SDK統合になるためリクエストレスポンスか、コールバックまで待機のどちらかになります。今回はリクエストレスポンスを用います。

作成したステートマシンに対して、定期的な起動にはEventBridge Schedulerをcron設定して実行させます。

ステートマシンを作成する

前項で実施していたエクスポート処理を実行し、60秒間の待ち状態の後、処理のステータス確認をし、ステータスがIN_PROGRESSなら、待ち状態に戻るようにループを実施します。また、エクスポート処理が失敗した場合、Failedステートに遷移させます。

下記がそのASLの内容です。

{
  "Comment": "A description of my state machine",
  "StartAt": "ExportTableToPointInTime",
  "States": {
    "ExportTableToPointInTime": {
      "Type": "Task",
      "Parameters": {
        "S3Bucket": "[エクスポートするDDBテーブルのARN]",
        "TableArn": "[エクスポート先のバケット名]",
        "ExportFormat": "DYNAMODB_JSON",
        "ExportType": "FULL_EXPORT"
      },
      "Resource": "arn:aws:states:::aws-sdk:dynamodb:exportTableToPointInTime",
      "Next": "Wait Export Task"
    },
    "Wait Export Task": {
      "Type": "Wait",
      "Seconds": 60,
      "Next": "DescribeExport"
    },
    "DescribeExport": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:dynamodb:describeExport",
      "Parameters": {
        "ExportArn.$": "$.ExportDescription.ExportArn"
      },
      "Next": "Choice Export Status"
    },
    "Choice Export Status": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.ExportDescription.ExportStatus",
          "StringMatches": "COMPLETED",
          "Next": "Success"
        },
        {
          "Variable": "$.ExportDescription.ExportStatus",
          "StringMatches": "IN_PROGRESS",
          "Next": "Wait Export Task"
        }
      ],
      "Default": "Fail"
    },
    "Success": {
      "Type": "Succeed"
    },
    "Fail": {
      "Type": "Fail"
    }
  }
}

ASLは自分で書かず、Step Functions Studioで作成したものをJSONの形式でエクスポートしています。

エクスポート処理が完了すると、指定したprefixを付与してエクスポート先バケットに出力データが配置されます。

aws s3 ls --recursive s3://[エクスポート先のバケット名]

2023-11-27 15:08:17          0 aaa/AWSDynamoDB/01701065278449-9b3c467c/_started
2023-11-27 15:13:23       6955 aaa/AWSDynamoDB/01701065278449-9b3c467c/data/h7sjj2scwuyvnfcsfr3sxyfmg4.json.gz
2023-11-27 15:13:48        201 aaa/AWSDynamoDB/01701065278449-9b3c467c/manifest-files.json
2023-11-27 15:13:48         24 aaa/AWSDynamoDB/01701065278449-9b3c467c/manifest-files.md5
2023-11-27 15:13:48        711 aaa/AWSDynamoDB/01701065278449-9b3c467c/manifest-summary.json
2023-11-27 15:13:48         24 aaa/AWSDynamoDB/01701065278449-9b3c467c/manifest-summary.md5

AWSDynamoDBの後ろの01701065278449-9b3c467cはエクスポートIDになります。この値はエクスポート処理時に自動生成されるため、エクスポート結果は実行時のIDから紐づけて取得する必要があります。

定期的に動かした結果を実行ごとに取得したい場合、ExportTableToPointInTime APIのprefixを利用して日付でオブジェクトを分ける手があります。

なお、データ量によっては.json.gzファイルは複数作成されることがあります。

2023-11-20 17:25:45     926913 hoge/AWSDynamoDB/01700468343538-dec955b1/data/cntkk3twyy6xtlm73mwc5i5qzu.json.gz
2023-11-20 17:26:01     697238 hoge/AWSDynamoDB/01700468343538-dec955b1/data/dcb2leftwi63vgoo7skmnvyoby.json.gz
2023-11-20 17:25:54        189 hoge/AWSDynamoDB/01700468343538-dec955b1/data/djz4msstlu5nnbu5kmmxwn4ehi.json.gz
2023-11-20 17:25:53      70313 hoge/AWSDynamoDB/01700468343538-dec955b1/data/nf7a3ifiem2fhbjneqb3o6vpiy.json.gz
2023-11-20 17:25:28         20 hoge/AWSDynamoDB/01700468343538-dec955b1/data/x73h5vegve45lag535vfa5uvpe.json.gz

ステートマシンを定期的に動かす

ステートマシンを定期的に実行するにあたっては、EventBridge Schedulerを利用します。

EventBridge SchedulerとRuleの違いや設定内容は昨年の投稿に書いていますので、こちらをご参考ください。

まとめ

サーバレス構成でのストア先として選ばれることの多いDDBからS3へデータを定期的にエクスポートする構成を作成しました。

DDBに蓄積されたデータを起点に別の作業、連携を行うことはあるかと思います。その際、この構成を使うことで実装なしでデータを出力できるようになりますので、参考にして頂けたらと思います。

次の私の投稿では、再利用性を高めるために弊社でいつも使っているCDKで今回の構成をコード管理させるようにしていく予定です。