DynamoDBのデータをS3にCSV形式でエクスポートする機能を作成する

DyanamoDBのテーブルに格納しているデータをS3にエクスポートするためのLambdaを作成しました。
DynamoDBの標準機能でも「S3へのエクスポート」機能がありましたが、ファイル形式が限られておりCSV形式では出力できませんでした。
データ連携やバックアップの意味でも使いやすいCSV形式で出力したかったので、作りこみを行いました。

このページのLambdaと他のAWSサービスを使用することで以下のようなことが実現できます。

• 指定したDynamoDBのテーブルデータをS3に定期出力が可能です。

※本ページではリストアップするLambdaのご紹介のみです。
　各サービスとの連携はAWSのサービスドキュメントを確認しながら実装してください。

EventBridgeにて定期実行のタイマー機能が実装可能です。
実行開始されたLambdaはDyanmoDBの内容をスキャンし、S3にCSV形式で出力ができます。
EventBridgeの作り方は、「EventBridgeを作成する」を参考に作成してください。

import boto3
import json
import os
import csv
import io

dynamodb_client = boto3.client('dynamodb')
s3_client = boto3.client('s3')
def lambda_handler(event, context):
    TableName = os.environ['TABLENAME']
    BucketName = os.environ['BUCKETNAME']
    FileName = os.environ['FILENAME']
    
    TableData = dynamodb_client.scan(TableName=TableName)
    TableItems = TableData['Items']
    while 'LastEvaluatedKey' in TableData:
        TableData = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
        TableItems.extend(TableData['Items'])
    
    if TableItems:
        keys = TableItems[0].keys()
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=keys)
        writer.writeheader()
        for TableItem in TableItems:
            writer.writerow(TableItem)
        
        s3_client.put_object(
            Bucket=BucketName,
            Key=FileName,
            Body=output.getvalue()
        )

    return 0
                

LambdaにアタッチするIAMロールに「dynamodb:Scan」、「s3:PutObject」の権限を付与してください。

紹介した機能のフローは以下の通りです。

1. 対象のDynamoDBのテーブルをスキャン
2. 「LastEvaluatedKey」を確認し、一度のスキャンで内容を確認しきれなかったデーブル情報を追加スキャン
3. スキャンしたテーブル情報をCSV形式でS3に出力

DynamoDBのスキャン機能はテーブルの容量によっては一度にスキャンしきれません。
その場合、応答結果に「LastEvaluatedKey」が含まれます。
「LastEvaluatedKey」が含まれる場合は再度スキャンを繰り返すような処理を作成しています。

対象のDynamoDBや出力先のS3バケット、CSVファイル名は環境変数にて指定しますので、作成するLambdaには以下の環境変数を設定してください。

• TABLENAME → 対象のDynamoDBテーブル名
• BUCKETNAME → 出力先のS3バケット名
• FILENAME → 出力先CSVファイル名

上記の環境変数を設定し、Lambdaを実行するだけでテーブル内の全ての情報がCSVファイルでS3にエクスポートができるようになります。