目的

AWSのDynamoDBのテーブルをCSVファイルでエクスポートをするPythonスクリプトを作成しました。

※下記の記事を参考にさせていただきました。
DynamoDBのテーブルをscanするPythonコード

前提条件

Python3がインストール済みであること
AWS CLI バージョン2がインストール済みであること
aws configure コマンドで認証情報、リージョン、出力形式を設定済みであること
AWSのDynamoDBにテーブルが存在すること

※今回は、DynamoDBに以下のテーブルが存在している前提です。
※テーブル名：GPSLog（GPSログデータ）

実行

出力CSVファイルを指定してください。
Pythonスクリプトを実行してください。

DynamoDBtoCSV.py

import os
import csv
import json
import datetime
import boto3
from boto3.session import Session

# 出力CSVファイル指定
output_csvfile = "出力CSVファイルを指定してください"

# 出力CSVファイルの存在チェック(ヘッダー有無の判断)
if(os.path.exists(output_csvfile)):
    header_flg = 0
else:
    header_flg = 1

# boto3でDynamoDBに接続
dynamodb = boto3.resource('dynamodb')

# テーブルをフルスキャン
def get_records(table, **kwargs):
    while True:
        response = table.scan(**kwargs)
        for item in response['Items']:
            yield item
        if 'LastEvaluatedKey' not in response:
            break
        kwargs.update(ExclusiveStartKey=response['LastEvaluatedKey'])

records = get_records(dynamodb.Table('GPSLog'))

# 出力CSVファイルに書き込む
with open(output_csvfile, 'a', encoding='shift_jis') as output_csvfile_fp:
    fieldnames = ['datetime', 'lat', 'lon', 'status', 'mode', 'speed', 'epx', 'epy']
    csvfile_writer = csv.DictWriter(output_csvfile_fp, fieldnames=fieldnames, lineterminator='\n')
    
    # 出力CSVファイル新規作成時、ヘッダー出力
    if header_flg == 1:
        csvfile_writer.writeheader()
        header_flg == 0

    for record in records:
        # print(record)
        # 出力CSVファイルに書き込む
        csvfile_writer.writerow({
            'datetime': record['datetime'] + '+09:00',
            'lat': record['lat'],
            'lon': record['lon'],
            'status': record['status'],
            'mode': record['mode'],
            'speed': record['speed'],
            'epx': record['epx'],
            'epy': record['epy']
        })

print(u'処理終了')