Data Pipeline Orchestration

Integraciones en Python entre un ERP y sistemas externos: snapshots de datos maestros, detección de cambios, envíos controlados y trazabilidad de cada ejecución.

Caso sanitizado a partir de una plataforma de integración en Python que reemplaza flujos de un middleware anterior. Los sistemas, rutas y tablas se describen de forma genérica.

Resultado: TODOEj.: ejecuciones por día o % de envíos con reintento exitoso.

Python
SQL
ETL
SAP HANA
Microsoft Fabric

Arquitectura

Etapas de una integración, del origen al registro de la ejecución

OrigenDatos maestros del ERPSAP HANA · SQL
SnapshotCopia del estado actual en una base intermediaPython
CambiosHash por fila y marca de pendientePython
ValidaciónReglas de elegibilidad por registroPython
EnvíoAPI del sistema destino, con reintentos y concurrencia acotadaHTTP
RegistroEstado y resultado de cada ejecuciónSQL · Fabric Warehouse

Dependencias

El snapshot debe terminar antes de calcular los cambios.
El envío solo toma registros marcados como pendientes y elegibles.
Los conectores de base de datos se eligen por entorno (QA o producción) mediante configuración, no en el código.

Controles y trazabilidad

Detección de cambios por hash de la fila de origen: si el contenido no cambia, no se reenvía.
Validación de elegibilidad antes del envío, replicando las reglas de negocio del flujo anterior.
Reintentos acotados y concurrencia limitada para no saturar el sistema destino.
Envíos selectivos de registros puntuales para reprocesar sin relanzar todo el lote.

Decisiones técnicas

Comparar la nueva integración con el flujo anterior (análisis de paridad) antes de reemplazarlo.
Guardar el estado de sincronización en la base de datos, no en la memoria del proceso.
Separar extracción, decisión y envío para poder probar cada etapa por separado.

Problemas técnicos y soluciones

Problema
Un campo del payload se interpretaba de otra forma en el sistema destino y marcaba productos con un atributo incorrecto.
Solución
Se confirmó la regla con el proveedor del sistema destino y se ajustó el mapeo para enviar el valor esperado.
Causa raíz confirmada
Problema
La marca de «pendiente de sincronizar» dependía del hash de una tabla intermedia y no de la fecha real de cambio en el ERP.
Solución
Se documentó el comportamiento y se planteó filtrar por las fechas del origen mediante un join, en lugar de copiarlas en cada snapshot.
Diagnóstico documentado; refactor propuesto

Código de ejemplo

Detección de cambios y envío con reintentos (código ilustrativo)python

def detect_changes(rows, previous_hashes):
    """Mark a row as pending only when its source content changed."""
    pending = []
    for row in rows:
        digest = hashlib.sha256(canonical_json(row).encode()).hexdigest()
        if previous_hashes.get(row["id"]) != digest:
            pending.append({**row, "row_hash": digest, "needs_sync": True})
    return pending


def push_pending(records, client, *, max_workers=4, max_retries=3):
    """Send eligible records with bounded concurrency and retries."""
    eligible = [r for r in records if is_eligible(r)]
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        results = list(pool.map(lambda r: send_with_retry(client, r, max_retries), eligible))
    log_run(results)  # one structured entry per execution
    return results

Límites del caso

El código es ilustrativo y está sanitizado: no es el código fuente original.
BigQuery no forma parte de este caso; lo uso en propuestas de arquitectura, no en esta integración.
No se publican volúmenes, frecuencias ni nombres de sistemas.