Importar arquivo CSV ou JSON para o DynamoDB

No final, codifiquei uma função Python import_csv_to_dynamodb(table_name, csv_file_name, colunm_names, column_types)que importa um CSV para uma tabela do DynamoDB. Os nomes das colunas e as colunas devem ser especificados. Ele usa boto e se inspira muito nessa essência . Abaixo está a função, bem como uma demonstração ( main()) e o arquivo CSV usado. Testado no Windows 7 x64 com Python 2.7.5, mas deve funcionar em qualquer sistema operacional que tenha boto e 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 (deve estar localizado na mesma pasta que o script Python):

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

Alterou um pouco a resposta anterior para usar o módulo CSV para que seu arquivo CSV possa suportar strings com aspas.

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

Este pacote NPM converte um json arbitrário em uma solicitação PUT para o DynamoDB. https://www.npmjs.com/package/json-dynamo-putrequest

Definitivamente vale a pena tentar.

Sugiro que você use o AWS Database Migration Service (DMS).

Conforme descrito neste artigo: 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/ você pode usar o S3 como origem e o DynamoDB como destino para importar arquivos csv com muitas tuplas.

Implementei com sucesso um processo de importação completo do S3 para o DynamoDB e é a maneira mais simples e rápida de fazer isso.

Essencialmente, você deve:

• Tenha um bucket para colocar seus arquivos csv, com pelo menos dois níveis de pastas (o primeiro se refere a "esquema" e o segundo é "nome da tabela").
• Tenha uma tabela do DynamoDB com pelo menos a mesma chave de hash dos arquivos csv.
• Crie um elemento de origem no DMS apontando para S3 e mapeando a estrutura csv.
• Crie um elemento de destino no DMS apontando para a tabela do DynamoDB e o mapeamento da origem mapeada.
• Crie uma instância de replicação (preste atenção ao nível gratuito) no DMS.
• Crie uma tarefa de replicação no DMS que use elementos criados de origem e destino.
• Executar tarefa.

Modificando o rendimento da tabela do DynamoDB para 25 unidades de capacidade de leitura e 150 unidades de capacidade de gravação, consegui inserir mais de 124k tuplas em menos de 7 minutos, incluindo as tarefas de preparação.

A recomendação principal da AWS para esta tarefa é usar o serviço de pipeline de dados, mas eu já o usei e é mais caro e a inicialização do culster EMR subjacente é um processo muito lento, portanto, se você não quiser repetir essa tarefa de importação recorrentemente, use o DMS em vez de.