Step functions でDynamoDBのテーブルを増分エクスポートする

はじめに

Step functionsを利用してDynamoDBのテーブルを増分エクスポートするための方法を残します。

方法

• 以下のワークフローを作成。
• このリンクを参考にするとエクスポートの処理をStep functionsから呼べるが、増分エクスポートの場合はパラメータの設定にエラーが発生し実現ませんでした（解決方法があればどなたか教えてください）。エラーは下記に記載しています。

CallExport (Lambda): 増分エクスポートの処理を呼び出す

Lambdaで下記のコードを書きました。

import json
import boto3
from datetime import datetime 

def lambda_handler(event, context):
    try:
        fromTime = datetime(2024,5,15,10,0,0)
        toTime = datetime(2024,5,15,11,0,0)
        
        client = boto3.client('dynamodb')
        tablearn = 'tablearn' 
        #'tablearn'は、DynamoDBのテーブル→概要→一般的な情報からARNを取得
        client.export_table_to_point_in_time(
            TableArn = tablearn,
            IncrementalExportSpecification={
                'ExportFromTime': fromTime,
                'ExportToTime': toTime,
                'ExportViewType': 'NEW_AND_OLD_IMAGES'
            },
            ExportType = 'INCREMENTAL_EXPORT',
            S3Bucket = 'bucket', #エクスポート先のbucket
            S3Prefix = 'prefix', #エクスポート先のprefix (あれば)
            S3SseAlgorithm='AES256',
            ExportFormat='DYNAMODB_JSON'
        )
    
        return {"Status": "Success"}
    except Exception as e:
        return {"Status": "Failed"}

pythonのプログラムの最後にreturn {"Status": "Success"}やreturn {"Status": "Failed"]を入れていますが以降使う場面はありません。。

Wait: 待機

待機時間を入力します。ここでは300秒としました。

checkExport: エクスポート状況を取得

DynamoDBのListExportsを利用してエクスポート状況を取得します。APIパラメータは、対象のテーブルを入力する必要があります。

{
  "MaxResults": 1,
  "TableArn": "table"
}

checkExportStatus: エクスポート状況の確認

条件分岐するフローになります。条件分岐は下記のように設定しました。

• 「完了（COMPLETED）」の場合は、「ExportCompleted」に遷移
• 「実行中（IN_PROGRESS)」の場合はWaitに遷移
• それ以外（FAILED）の場合は、「ExportFailed」に遷移
  

ExportCompleted または ExportFailed

最後に、「ExportCompleted（Success）」と「ExportFailed（Fail）」のフローを追加しておきます。

失敗例

Step functionsのDynamoDB:ExportTableToPointInTimeで増加エクスポートしてみた

当初、以下のワークフローを作成していました。

増加エクスポートを実行する際に、パラメータ「ExportFromTime」と「ExportToTime」が必要でAWSのExportTableToPointInTimeを参考にするとUnix epochのタイムスタンプが必要であったためLambdaで出力するようにしていました。
しかし、実行すると以下の様なエラーが出力されました。

An error occurred while executing the state 'CallExport' (entered at the event id #7). 
The Parameters '{"S3Bucket":"bucket","S3Prefix":"prefix","TableArn":"tablearn",
"ExportType":"INCREMENTAL_EXPORT","IncrementalExportSpecification":
{"ExportViewType":"NEW_AND_OLD_IMAGES","ExportToTime":1715749248,"ExportFromTime":1715745048},
"S3SseAlgorithm":"AES256","ExportFormat":"DYNAMODB_JSON"}' could not be used to start 
the Task: [The value for the field 'ExportToTime' must be a TIMESTAMP STRING]

「ExportFromTime」と「ExportToTime」の型を「TIMESTAMP STRING」にしないといけないのですが、
解決する術を見つけられず、上記の方法となっております。
解決策がある方は教えていただけると幸いです。