[Python] Como Otimizar Inserção de Milhões de Registros de CSV para PostgreSQL usando COPY...

Estou desenvolvendo uma aplicação em Python que precisa processar grandes volumes de dados de um arquivo CSV, aplicar algumas transformações e depois persistir esses dados em um banco de dados PostgreSQL. O desafio é otimizar a inserção para que seja eficiente, pois estou lidando com milhões de registros.

Meu objetivo é encontrar a melhor estratégia para inserir dados massivamente de um arquivo CSV para o PostgreSQL usando Python, priorizando a performance e a robustez. Preciso de uma solução que evite a lentidão de inserções INSERT linha por linha e que possa lidar com a possibilidade de dados duplicados ou erros durante o processo.

Atualmente, estou lendo o CSV com o módulo csv do Python e fazendo inserções INSERT individuais para cada linha. Isso está extremamente lento, levando horas para processar arquivos que deveriam ser importados em minutos.

Comportamento Atual:

import csv
import psycopg2

# Conexão com o banco de dados (exemplo)
conn = psycopg2.connect("dbname=meubanco user=meuuser password=minhasenha")
cur = conn.cursor()

with open('dados_grandes.csv', 'r') as f:
reader = csv.reader(f)
next(reader) # Pula o cabeçalho
for row in reader:
try:
cur.execute("INSERT INTO minha_tabela (col1, col2, col3) VALUES (%s, %s, %s)", tuple(row))
except Exception as e:
print(f"Erro ao inserir linha: {row} - {e}")
conn.rollback() # Desfaz a transação em caso de erro
# Potencialmente, logar a linha com erro para reprocessamento
conn.commit() # Comita a cada linha, o que é lento
cur.close()
conn.close()


Resultados Atuais: A inserção de 1 milhão de linhas leva aproximadamente 2-3 horas.

Resultados Esperados: Gostaria de reduzir esse tempo para alguns minutos (idealmente, menos de 5-10 minutos para 1 milhão de linhas), aproveitando recursos do PostgreSQL para inserção em massa. Além disso, a solução deve ter algum mecanismo para lidar com duplicatas (por exemplo, ignorar, atualizar ou reportar) e erros de dados (por exemplo, logar as linhas com erro e continuar o processamento).

ão há mensagens de erro específicas no momento, apenas o problema de performance. No entanto, em um cenário real, poderíamos ter erros de duplicação de chave primária (psycopg2.errors.UniqueViolation) ou erros de tipo de dado (psycopg2.errors.InvalidTextRepresentation) se os dados do CSV não corresponderem ao esquema do banco de dados. A solução deve prever esses cenários.

O que já foi tentado:

1. Inserção linha a linha com execute(): Esta é a abordagem atual, e é a que está causando o problema de performance.
2. executemany() com psycopg2: Pensei em usar cur.executemany(), mas ele ainda constrói múltiplas INSERT statements. Li que para volumes realmente grandes, isso ainda pode ser ineficiente se não for acompanhado de um COPY.
3. Transações maiores: Tentei agrupar as inserções em transações maiores (comitar a cada 1000 linhas, por exemplo), o que melhorou um pouco, mas ainda não é ideal para milhões de registros.
4. Pesquisas no Stack Overflow: Encontrei algumas discussões sobre COPY FROM do PostgreSQL, mas não tenho certeza da melhor forma de integrá-lo com Python e como ele lida com a validação de dados e tratamento de erros/duplicatas diretamente no nível do banco de dados.

Estrutura da Tabela PostgreSQL:

CREATE TABLE minha_tabela (
id SERIAL PRIMARY KEY,
nome VARCHAR(255) NOT NULL,
email VARCHAR(255) UNIQUE, -- Para testar duplicatas
idade INTEGER
);


O que vocês recomendam para uma inserção massiva mais eficiente e robusta, considerando os pontos de performance, tratamento de duplicatas (ex: ON CONFLICT DO NOTHING/UPDATE) e tratamento de erros de dados? Existe alguma biblioteca Python que facilite o uso do COPY FROM do PostgreSQL com essas validações?

Continue reading...