Import CSV or JSON file into DynamoDB

Question

I have 1000 CSV files. Each CSV file is between 1 and 500 MB and is formatted the same way (i.e. same column order). I have a header file for column headers, which match my DynamoDB table's column names. I need to import those files into a DynamoDB table. What's the best way / tool to do so?

I can concatenate those CSV files into a single giant file (I'd rather avoid to though), or convert them into JSON if needed. I am aware of the existence of BatchWriteItem so I guess a good solution would involve batch writing.

---

Example:

• The DynamoDB table has two columns: first_name, last_name
• The header file only contains: first_name,last_name
• One CSV file looks like

:

John,Doe
Bob,Smith
Alice,Lee
Foo,Bar

Answer 1

In the end I coded a Python function import_csv_to_dynamodb(table_name, csv_file_name, colunm_names, column_types) that imports a CSV into a DynamoDB table. Column names and column must be specified. It uses boto, and takes a lot of inspiration from this gist. Below is the function as well as a demo (main()) and the CSV file used. Tested on Windows 7 x64 with Python 2.7.5, but it should work on any OS that has boto and Python.

import boto

MY_ACCESS_KEY_ID = 'copy your access key ID here'
MY_SECRET_ACCESS_KEY = 'copy your secrete access key here'


def do_batch_write(items, table_name, dynamodb_table, dynamodb_conn):
    '''
    From https://gist.github.com/griggheo/2698152#file-gistfile1-py-L31
    '''
    batch_list = dynamodb_conn.new_batch_write_list()
    batch_list.add_batch(dynamodb_table, puts=items)
    while True:
        response = dynamodb_conn.batch_write_item(batch_list)
        unprocessed = response.get('UnprocessedItems', None)
        if not unprocessed:
            break
        batch_list = dynamodb_conn.new_batch_write_list()
        unprocessed_list = unprocessed[table_name]
        items = []
        for u in unprocessed_list:
            item_attr = u['PutRequest']['Item']
            item = dynamodb_table.new_item(
                    attrs=item_attr
            )
            items.append(item)
        batch_list.add_batch(dynamodb_table, puts=items)


def import_csv_to_dynamodb(table_name, csv_file_name, colunm_names, column_types):
    '''
    Import a CSV file to a DynamoDB table
    '''        
    dynamodb_conn = boto.connect_dynamodb(aws_access_key_id=MY_ACCESS_KEY_ID, aws_secret_access_key=MY_SECRET_ACCESS_KEY)
    dynamodb_table = dynamodb_conn.get_table(table_name)     
    BATCH_COUNT = 2 # 25 is the maximum batch size for Amazon DynamoDB

    items = []

    count = 0
    csv_file = open(csv_file_name, 'r')
    for cur_line in csv_file:
        count += 1
        cur_line = cur_line.strip().split(',')

        row = {}
        for colunm_number, colunm_name in enumerate(colunm_names):
            row[colunm_name] = column_types[colunm_number](cur_line[colunm_number])

        item = dynamodb_table.new_item(
                    attrs=row
            )           
        items.append(item)

        if count % BATCH_COUNT == 0:
            print 'batch write start ... ', 
            do_batch_write(items, table_name, dynamodb_table, dynamodb_conn)
            items = []
            print 'batch done! (row number: ' + str(count) + ')'

    # flush remaining items, if any
    if len(items) > 0: 
        do_batch_write(items, table_name, dynamodb_table, dynamodb_conn)


    csv_file.close() 


def main():
    '''
    Demonstration of the use of import_csv_to_dynamodb()
    We assume the existence of a table named `test_persons`, with
    - Last_name as primary hash key (type: string)
    - First_name as primary range key (type: string)
    '''
    colunm_names = 'Last_name First_name'.split()
    table_name = 'test_persons'
    csv_file_name = 'test.csv'
    column_types = [str, str]
    import_csv_to_dynamodb(table_name, csv_file_name, colunm_names, column_types)


if __name__ == "__main__":
    main()
    #cProfile.run('main()') # if you want to do some profiling

test.csv's content (must be located in the same folder as the Python script):

John,Doe
Bob,Smith
Alice,Lee
Foo,Bar
a,b
c,d
e,f
g,h
i,j
j,l

Answer 2

Changed the previous answer slightly to use the CSV module so you CSV file can support strings with quotes.

import boto
from csv import reader

MY_ACCESS_KEY_ID = 'copy your access key ID here'
MY_SECRET_ACCESS_KEY = 'copy your secrete access key here'


def do_batch_write(items, table_name, dynamodb_table, dynamodb_conn):
    '''
    From https://gist.github.com/griggheo/2698152#file-gistfile1-py-L31
    '''
    batch_list = dynamodb_conn.new_batch_write_list()
    batch_list.add_batch(dynamodb_table, puts=items)
    while True:
        response = dynamodb_conn.batch_write_item(batch_list)
        unprocessed = response.get('UnprocessedItems', None)
        if not unprocessed:
            break
        batch_list = dynamodb_conn.new_batch_write_list()
        unprocessed_list = unprocessed[table_name]
        items = []
        for u in unprocessed_list:
            item_attr = u['PutRequest']['Item']
            item = dynamodb_table.new_item(
                    attrs=item_attr
            )
            items.append(item)
        batch_list.add_batch(dynamodb_table, puts=items)


def import_csv_to_dynamodb(table_name, csv_file_name, colunm_names,     column_types):
    '''
    Import a CSV file to a DynamoDB table
    '''        
    dynamodb_conn =     boto.connect_dynamodb(aws_access_key_id=MY_ACCESS_KEY_ID, aws_secret_access_key=MY_SECRET_ACCESS_KEY)
    dynamodb_table = dynamodb_conn.get_table(table_name)     
    BATCH_COUNT = 2 # 25 is the maximum batch size for Amazon DynamoDB

    items = []

    count = 0
    csv_file = open(csv_file_name, 'r')
    for cur_line in reader(csv_file):
        count += 1

        row = {}
        for colunm_number, colunm_name in enumerate(colunm_names):
            row[colunm_name] = column_types[colunm_number]    (cur_line[colunm_number])

        item = dynamodb_table.new_item(
                    attrs=row
            )           
        items.append(item)

        if count % BATCH_COUNT == 0:
            print 'batch write start ... ', 
            do_batch_write(items, table_name, dynamodb_table, dynamodb_conn)
            items = []
            print 'batch done! (row number: ' + str(count) + ')'

    # flush remaining items, if any
    if len(items) > 0: 
        do_batch_write(items, table_name, dynamodb_table, dynamodb_conn)


    csv_file.close() 


def main():
    '''
    Demonstration of the use of import_csv_to_dynamodb()
    We assume the existence of a table named `test_persons`, with
    - Last_name as primary hash key (type: string)
    - First_name as primary range key (type: string)
    '''
    colunm_names = 'facebookID age ethnicity gender hometown name party sfw url'.split()
    table_name = 'OneAmericaDB'
    csv_file_name = 'test_data.csv'
    column_types = [str, str, str, str, str, str, str, str, str]
    import_csv_to_dynamodb(table_name, csv_file_name, colunm_names, column_types)


if __name__ == "__main__":
    main()
    #cProfile.run('main()') # if you want to do some profiling

Answer 3

This NPM package converts an arbitray json into a PUT request for DynamoDB. https://www.npmjs.com/package/json-dynamo-putrequest

Definitely worth a try.

Answer 4

I suggest you to use AWS Database Migration Service (DMS).

As described in this article: https://aws.amazon.com/es/blogs/database/migrate-delimited-files-from-amazon-s3-to-an-amazon-dynamodb-nosql-table-using-aws-database-migration-service-and-aws-cloudformation/ you can use S3 as a origin and DynamoDB as a target to import csv files with a lot of tuples.

I've sucessfully implemented a full import process from S3 to DynamoDB and is the simpliest and fastest way to do it.

Essentially, you have to:

• Have a bucket to put your csv files in, with at least two folders levels (first one reffers to "schema" and the second one is "table name").
• Have a DynamoDB table with at least the same hash key than in csv files.
• Create an origin element in DMS pointing to S3 and mapping the csv structure.
• Create a target element in DMS pointing to DynamoDB table and mapping from mapped origin.
• Create a replication instance (pay attention to free tier) in DMS.
• Create a replication task in DMS that use origin and target created elements.
• Execute task.

Modifying DynamoDB table troughput to 25 read capacity units and 150 write capacity units, I've been able to insert more than 124k tuples in less than 7 minutes including thee preparation tasks.

AWS primary recommendation for this task is to use Data pipeline service, but I've used it and it's more expensive and the underlying EMR culster initialization is a very slow process, so if you don't want to repeat this import task recurrently use DMS instead.