AWSのDynamoDBのテーブルをCSVファイルでエクスポートするPythonスクリプトを作成しました
目的
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'処理終了')
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'処理終了')
Author And Source
この問題について(AWSのDynamoDBのテーブルをCSVファイルでエクスポートするPythonスクリプトを作成しました), 我々は、より多くの情報をここで見つけました https://qiita.com/syanseto/items/4fa002dbda4710154bea著者帰属:元の著者の情報は、元のURLに含まれています。著作権は原作者に属する。
Content is automatically searched and collected through network algorithms . If there is a violation . Please contact us . We will adjust (correct author information ,or delete content ) as soon as possible .