Como fazer deduplicação incremental em ETL: passo a passo

Aprende a implementar deduplicação incremental em ETL para carregar apenas registos novos sem duplicar dados existentes — útil quando recebes ficheiros ou APIs periódicas e queres garantir integridade e eficiência. Este exemplo usa Python para mostrar o porquê e o como de forma prática.

Pré-requisitos

• Python 3.8+ instalado
• Bibliotecas: pandas, sqlalchemy, sqlite3 (ou equivalente)
• Conhecimentos básicos de SQL e pandas
• Editor de texto e terminal

Passo 1: Perceber o problema e escolher a estratégia

A deduplicação incremental evita recarregar tudo e refazer comparações dispendiosas. Estratégias comuns: chave única (surrogate key/business key), hash de linha, ou comparação por timestamp. Vamos usar uma business key e um hash de linha para detetar alterações. Isto permite identificar registos novos, alterados ou iguais.

Passo 2: Preparar um exemplo mínimo (ficheiro CSV)

Criar um pequeno ficheiro CSV que representa dados periódicos de vendas. Ter uma business key chamada order_id.

# vendas_1.csv
order_id,customer,amount,date
1,Ana,100,2026-01-01
2,Bruno,50,2026-01-02
3,Carlos,75,2026-01-03

# vendas_2.csv (entrega seguinte, com duplicados e alteração)
order_id,customer,amount,date
2,Bruno,50,2026-01-02
3,Carlos,80,2026-01-03
4,Daniela,60,2026-01-04

Passo 3: Escrever o código ETL básico com deduplicação incremental

Explicação: vamos ler o novo ficheiro, calcular um hash por linha, comparar com a tabela destino em SQLite e inserir apenas registos novos ou alterados. O exemplo é mínimo e funcional.

import pandas as pd
import hashlib
from sqlalchemy import create_engine, text

# Função para criar hash de uma linha (exclui order_id)
def row_hash(row):
    payload = '|'.join(str(row[c]) for c in sorted(row.index) if c != 'order_id')
    return hashlib.md5(payload.encode('utf-8')).hexdigest()

# Engine SQLite (trocar por SQL Server/Postgres com connection string apropriada)
engine = create_engine('sqlite:///etl_example.db')

# Ler novo ficheiro
df_new = pd.read_csv('vendas_2.csv')
# Calcular hash
df_new['row_hash'] = df_new.apply(row_hash, axis=1)

# Garantir tabela destino existe com colunas: order_id, customer, amount, date, row_hash
with engine.begin() as conn:
    conn.execute(text('''
        CREATE TABLE IF NOT EXISTS vendas (
            order_id INTEGER PRIMARY KEY,
            customer TEXT,
            amount REAL,
            date TEXT,
            row_hash TEXT
        )
    '''))

# Carregar hashes existentes
with engine.connect() as conn:
    existing = pd.read_sql('SELECT order_id, row_hash FROM vendas', conn)

# Determinar quais inserir ou actualizar
merged = df_new.merge(existing, on='order_id', how='left', suffixes=('', '_existing'))
# Novo se row_hash_existing for nulo; alterado se diferente
to_insert = merged[merged['row_hash_existing'].isna()]
to_update = merged[(~merged['row_hash_existing'].isna()) & (merged['row_hash_existing'] != merged['row_hash'])]

# Inserir novos
if not to_insert.empty:
    to_insert[['order_id','customer','amount','date','row_hash']].to_sql('vendas', engine, if_exists='append', index=False)

# Actualizar alterados (simples: UPDATE por order_id)
with engine.begin() as conn:
    for _, row in to_update.iterrows():
        conn.execute(text('''
            UPDATE vendas SET customer = :customer, amount = :amount, date = :date, row_hash = :row_hash
            WHERE order_id = :order_id
        '''), {
            'customer': row['customer'], 'amount': row['amount'], 'date': row['date'], 'row_hash': row['row_hash'], 'order_id': int(row['order_id'])
        })

print('Novos:', len(to_insert), 'Alterados:', len(to_update))

Passo 4: Tratar erros comuns e garantir idempotência

Erros comuns: colisões de chave, formatos de data inconsistentes, e ficheiros parcialmente carregados. Boas práticas: usar transacções (o create_engine().begin() já ajuda), validar schema com pandas dtypes, e gerar logs simples. A idempotência aqui resulta de usar a business key e comparação por hash.

# Exemplo simples de validação de schema antes de inserir
expected_cols = {'order_id','customer','amount','date'}
if not expected_cols.issubset(set(df_new.columns)):
    raise ValueError('Faltam colunas no ficheiro novo')

# Normalizar tipos
df_new['order_id'] = df_new['order_id'].astype(int)

Verificar o resultado

Para confirmar: consulta a tabela destino e valida os registos. Espera ver 4 registos após carregar vendas_1.csv e vendas_2.csv, com order_id 3 actualizado para amount 80.

from sqlalchemy import create_engine
import pandas as pd
engine = create_engine('sqlite:///etl_example.db')
print(pd.read_sql('SELECT * FROM vendas ORDER BY order_id', engine))

Conclusão

A deduplicação incremental com business key + hash de linha é uma técnica simples e eficiente para ETL diário/periódico. Próximos passos: adaptar para bases como SQL Server/Postgres, adicionar logging estruturado, e optimizar updates em lote. Dica: qual é a melhor business key para os teus dados?