AWS DynamoDB 테이블을 CSV 파일로 내보내는 파이썬 스크립트를 만들었습니다.

목적



AWS의 DynamoDB 테이블을 CSV 파일로 내보내는 Python 스크립트를 만들었습니다.

※아래의 기사를 참고로 했습니다.
DynamoDB 테이블을 스캔하는 파이썬 코드

전제 조건


  • 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'処理終了')
    

    좋은 웹페이지 즐겨찾기