por Mark Tse, desenvolvedor de software, desenvolvimento de plataforma D2L
publicado originalmente em 24 de março de 2018
Adicionar uma coluna, desde que ela seja anexada ao final de um conjunto de dados e, portanto, não interrompa a ordem das colunas existentes, é considerada de baixo impacto. Qualquer sistema que ingere as versões anteriores do conjunto de dados que não continha essas colunas não deve gerar erros ou não funcionar devido à adição dessas colunas anexadas.
Neste artigo, faremos alterações no Exemplo de código em Exemplo de cliente headless dos conjuntos de dados do Brightspace e Conjuntos de dados do Brightspace - Exemplo de cliente de conjuntos de dados diferenciais para oferecer suporte a alterações aditivas no esquema de conjuntos de dados do Brightspace.
Visão geral das mudanças
Para suportar extrações com colunas no CSV que a tabela do banco de dados não possui, teremos que truncar essas colunas extras antes de atualizar a tabela do banco de dados.
Ignorando novas colunas
Lembre-se de que, para cada extração de dados, carregamos toda a extração em uma tabela temporária antes de fazer um upsert (inserir ou atualizar):
def update_db(db_conn_params, table, csv_data): ''' Em uma única transação, atualize a tabela: - Carregando os dados CSV em uma tabela temporária - Execute uma consulta update ou insert para atualizar a tabela principal com os dados na tabela temporária - Exclua a tabela temporária ''' com psycopg2.connect(**db_conn_params) as conn: with conn.cursor() as cur: tmp_table_id = sql. Identifier('tmp_' + table) # 1) Crie a tabela temporária cur.execute( sql. SQL(''' CRIAR TABELA TEMPORÁRIA {tmp_table} AS SELECT * FROM {tabela} LIMIT 0; ''') .format( tmp_table=tmp_table_id, table=sql. Identifier(table) ) ) # 2) Atualize em massa o CSV na tabela temporária cur.copy_expert( sql. SQL(''' COPIA {tmp_table} DO STDIN COM (FORMAT CSV, HEADER); ''') .format(tmp_table=tmp_table_id), csv_data ) # 3) Execute o upsert upsert_query_file = os.path.join( os.path.dirname(os.path.abspath(__file__)), 'schema', 'upserts', 'upsert_{table}.sql'.format(table=table) ) com open(upsert_query_file) como upsert_query: cur.execute(upsert_query.read()) # 4) Remova a tabela temporária cur.execute(sql. SQL('DROP TABLE {tmp_table}').format(tmp_table=tmp_table_id)) conn.commit()
Etapa 2) usa o COPIAR no PostgreSQL, que requer que o arquivo CSV de entrada tenha o número exato de colunas como na tabela. Se o banco de dados tiver um esquema mais antigo definido e uma extração de dados com um esquema mais recente for usada (que tem mais colunas do que a tabela), o PostgreSQL lançará uma exceção.
Para evitar que isso ocorra, processaremos o arquivo CSV para remover todas as colunas irrelevantes. Primeiro, precisaremos determinar quantas colunas a tabela no banco de dados tem:
def get_number_of_columns(db_conn_params, table): com psycopg2.connect(**db_conn_params) as conn: with conn.cursor() as cur: cur.execute( sql. SQL(''' SELECT * FROM {tabela} LIMITE 0; ''') .format(table=sql. Identifier(table)) ) # cur.description é uma lista de tuplas nomeadas # Cada tupla nomeada descreve uma coluna na tabela # O comprimento dessa lista é, portanto, o número de colunas na # tabela. return len(cur.description)
Em seguida, adicionaremos um método que processa a extração de dados dado o número de colunas na tabela:
def process_csv_stream(csv_input_stream, num_columns_in_table): ''' Ignore o excesso de colunas no CSV devido a alterações aditivas / pequenas alterações do BDS, ignorando todas as colunas no CSV além do número de colunas na tabela ''' csv_rows = [] # Leia cada linha no CSV e trunque todas as colunas ainda não definidas para # a tabela em que estamos carregando os dados csv_reader = csv.reader(csv_input_stream, citação=csv. QUOTE_MINIMAL) for line in csv_reader: csv_rows.append(line[:num_columns_in_table]) # Reescreva as linhas processadas em um novo objeto semelhante a um arquivo CSV (ou seja, um # fluxo) csv_data = io. StringIO() csv_writer = csv.writer(csv_data, citing=csv. QUOTE_MINIMAL) csv_writer.writerows(csv_rows) # Rebobine o fluxo para o início antes de retorná-lo para que o # leitor downstream leia desde o início csv_data.seek(io. SEEK_SET) devolução csv_data
Preocupações com a memória
Observe que em process_csv_stream() lemos todo o CSV na memória. Isso se torna inviável com grandes extrações de dados. Para evitar problemas de falta de memória, processaremos o CSV em lotes em vez de todo o arquivo ao mesmo tempo. Para fazer isso, adicionaremos um novo método:
def batch_update_db(db_conn_params, table, csv_file, batch_size=10000): # Remove a primeira linha, que contém os cabeçalhos csv_file.readline() num_columns = get_number_of_columns(db_conn_params, table) csv_input_stream = io. StringIO() def update_db_with_batch(input_stream): ''' Método auxiliar que forma um fechamento para que não tenhamos que passar muitos dos valores usados neste método como argumentos ''' # Retroceda o fluxo para o início antes de passá-lo para input_stream.seek(io. SEEK_SET) com process_csv_stream(input_stream, num_columns) como csv_data: # Reutilize este método já existente update_db(db_conn_params, table, csv_data) input_stream.close() i = 0 for line in csv_file: csv_input_stream.write(line.decode('utf-8')) i += 1 # Para cada 10000 linhas, atualizamos a tabela antes de continuar. # O tamanho do lote é configurável e a amostra de código pode ser estendida para # passar o tamanho do lote como um argumento (fora do escopo deste artigo) se i == batch_size: update_db_with_batch(csv_input_stream) # Atribua um novo objeto semelhante a um arquivo para que o Python saiba que o antigo # pode ser limpo csv_input_stream = io. StringIO() i = 0 # Atualiza o banco de dados com as linhas restantes update_db_with_batch(csv_input_stream)
Em seguida, mudamos nosso código para usar .batch_update_db() Em vez de update_db() diretamente:
def unzip_and_update_db(response_content, db_conn_params, tabela): com io. BytesIO(response_content) como response_stream: com zipfile. ZipFile(response_stream) as zipped_data_set: files = zipped_data_set.namelist() assert len(files) == 1 csv_name = files[0] with zipped_data_set.open(csv_name) as csv_file: # update_db(db_conn_params, table, csv_data) batch_update_db(db_conn_params, table, csv_file)
Por fim, como agora estamos removendo o cabeçalho no código, precisamos dizer ao PostgreSQL para não remover a primeira linha do CSV removendo o CABEÇALHO na opção update_db() método:
cur.copy_expert( sql. SQL(''' COPIA {tmp_table} DO STDIN COM (FORMATO CSV); ''') .format(tmp_table=tmp_table_id), csv_data)
Opções alternativas
Essas alterações são necessárias, pois o exemplo de código usa o PostgreSQL como banco de dados de backup, que não tem suporte SQL integrado para ignorar colunas em um arquivo CSV. Opções alternativas estão disponíveis para os seguintes mecanismos de banco de dados que usam a funcionalidade incorporada ao próprio mecanismo:
É provável que haja uma melhoria de desempenho usando a funcionalidade incorporada ao mecanismo de banco de dados. No entanto, as alterações descritas neste artigo são independentes do mecanismo de banco de dados e podem ser adaptadas para qualquer mecanismo de banco de dados. Cabe ao leitor decidir qual opção seguir.
Resumo
Neste artigo, mostramos como podemos pré-processar uma extração de dados no código para gerenciar alterações aditivas no esquema BDS. Por favor, deixe-nos saber de qualquer feedback, perguntas ou comentários!