Cómo hacer deduplicación incremental en ETL: paso a paso

Aprende a implementar deduplicación incremental en ETL para cargar solo registros nuevos sin duplicar datos existentes —útil cuando recibes ficheros o APIs periódicas y quieres garantizar integridad y eficiencia. Este ejemplo usa Python para mostrar el porqué y el cómo de forma práctica.

Prerequisitos

• Python 3.8+ instalado
• Bibliotecas: pandas, sqlalchemy, sqlite3 (o equivalente)
• Conocimientos básicos de SQL y pandas
• Editor de texto y terminal

Paso 1: Entender el problema y elegir la estrategia

La deduplicación incremental evita recargar todo y rehacer comparaciones costosas. Estrategias comunes: clave única (surrogate key/business key), hash de fila, o comparación por timestamp. Vamos a usar una business key y un hash de fila para detectar cambios. Esto permite identificar registros nuevos, modificados o iguales.

Paso 2: Preparar un ejemplo mínimo (fichero CSV)

Crear un pequeño fichero CSV que representa datos periódicos de ventas. Tener una business key llamada 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 siguiente, con duplicados y modificación)
order_id,customer,amount,date
2,Bruno,50,2026-01-02
3,Carlos,80,2026-01-03
4,Daniela,60,2026-01-04

Paso 3: Escribir el código ETL básico con deduplicación incremental

Explicación: vamos a leer el fichero nuevo, calcular un hash por fila, comparar con la tabla destino en SQLite e insertar solo registros nuevos o modificados. El ejemplo es mínimo y 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))

Paso 4: Manejar errores comunes y garantizar idempotencia

Errores comunes: colisiones de clave, formatos de fecha inconsistentes y ficheros parcialmente cargados. Buenas prácticas: usar transacciones (el create_engine().begin() ya ayuda), validar esquema con pandas dtypes y generar logs simples. La idempotencia aquí resulta de usar la business key y la comparación 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('Faltan columnas en el fichero nuevo')

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

Verificar el resultado

Para confirmar: consulta la tabla destino y valida los registros. Espera ver 4 registros tras cargar vendas_1.csv y vendas_2.csv, con order_id 3 actualizado a 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))

Conclusión

La deduplicación incremental con business key + hash de fila es una técnica simple y eficiente para ETL diario/periódico. Próximos pasos: adaptar para bases como SQL Server/Postgres, añadir logging estructurado y optimizar updates por lotes. Consejo: ¿cuál es la mejor business key para tus datos?