【Python】AWS Dynamoに大量データをimportするツール

Pythonのboto3を使ってAWS Dynamoのテーブルに大量にデータを登録するツールを作成したのでメモ書きします。
私の環境では一時間で約200万件のデータを登録できました。

登録データの準備

今回は下記のようなデータをtest_table.csvとして用意します。
キーは指定せず、キーの順番に値を記載します。（キーはNo, Name, Age, Hobbyの順）

1,taro,12,abcd
2,jiro,22,efgh
3,sato,32,abcd
4,goro,42,efgh

既存のdynamodbからデータを取得する場合は下記コマンドで取得できます。

aws dynamodb scan --table-name [任意のテーブル名]> test_table.csv

python3とboto3の準備

python3とboto3がインストールされいていない場合はインストールします。
下記はmacでの手順となります。

# python3のインストール
brew install python3

# boto3もインストールしておく
pip3 install boto3

コード実装

下記の様に実装を行います。
ファイル名はimport_items.pyとしています。
先ほど作成したtest_table.csvもimport_items.pyと同じディレクトリに配置します。

# import_items.py

import subprocess
import pathlib
import csv
import os
import time
import boto3

dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')

# 固定の変数指定
TAMEL_NAME = "test"
CURRENT_PATH = os.path.dirname(__file__)
# 読み込ませるcsvファイルを設定
IMPUT_CSV = pathlib.Path(CURRENT_PATH + "/", f'test_table.csv') 


# 処理開始
def excute():
    table = dynamodb.Table(TAMEL_NAME)
    items_source = []
    # 出力ファイルの件数確認
    with open(IMPUT_CSV, 'r') as imput_csv:
        read_imput_csv = csv.reader(imput_csv)
        row_count = sum(1 for row in read_imput_csv)

    with open(IMPUT_CSV, 'r') as imput_csv:
        print(f"row_count:{row_count}")
        read_imput_csv = csv.reader(imput_csv)
        for i, csv_line in enumerate(read_imput_csv):
            # キーにテーブルに登録する属性名を設定する。左側はcsvファイルから読み込んだ値が設定される
            items_source.append(
            f"'No':'{str(csv_line[0])}',"
            f"'Name':'{csv_line[1]}'," +\
            f"'Age':'{csv_line[2]}'," +\
            f"'Hobby':{csv_line[3]}"
            )
        # レコードを一件ずつ読み込む
        # `overwrite_by_pkeys` に指定したカラムについては、重複していた場合、レコードを上書きする
        # これを指定しない場合、値が重複していた時エラーとなる
        with table.batch_writer(overwrite_by_pkeys=['No']) as batch:
            exec_count = 0
            for item in items_source:
                batch.put_item(Item=eval('{'+ item +'}'))
                exec_count = exec_count + 1
                print(f"[{exec_count}]: {item}")

if __name__ == '__main__':
    # ツール実行
    excute()

ツールの実行

import_items.pyのあるディレクトリに移動し、下記コマンドを実行します。

python3 import_items.py