ZrobDev/pipelines-ETL

Pipelines ETL

Ingestión por lotes en PostgreSQL (almacén predeterminado) con Python (uv, ruff, pytest). Transformaciones locales: polars 2.x o DuckDB 2.0.

Procedimiento

1. Declarar el grano: qué significa una fila, identificada por qué clave natural. Sin clave, no hay carga idempotente; definir una (o un hash de contenido) primero.
2. Elegir el patrón de carga (abajo). Elegir la estrategia incremental. Elegir el orquestador al final.
3. Escribir la ejecución para que volver a ejecutar cualquier ventana dé el mismo estado final. Probarlo: ejecutarlo dos veces en una prueba y afirmar que los conteos de filas y un checksum no cambian.
4. Agregar las comprobaciones de observabilidad (abajo) antes de la primera ejecución en producción.
5. Revisar contra la lista de verificación al final.

Patrón de carga idempotente

• Las filas tienen una clave natural estable y pueden cambiar: INSERT ... ON CONFLICT (key) DO UPDATE (Postgres) o MERGE. Agregar WHERE target.hash IS DISTINCT FROM excluded.hash para que las filas sin cambios no generen actividad.
• Los datos llegan como una partición completa (día, id de lote) y se reescriben como una unidad: eliminar-insertar esa partición en UNA transacción. Un DELETE confirmado por separado del INSERT deja un hueco en caso de fallo.
• La tabla completa se reconstruye: cargar en una tabla temporal, validar, luego intercambiar (renombrar en una transacción). Los lectores nunca ven una tabla medio cargada.
• Hechos de solo agregar sin clave: agregar un batch_id determinista / hash del archivo fuente y hacer que la carga rechace o reemplace un id de lote ya cargado.
• Nunca hacer INSERT a ciegas y luego deduplicar después. Los duplicados llegan a un panel antes de que se ejecute la deduplicación.
• Cargar mediante COPY a una tabla temporal/etapa, luego un upsert basado en conjunto. Los upserts fila por fila desde Python son 10-100 veces más lentos.

Cargas incrementales

• Marca de agua en una columna que la fuente garantiza monótona e indexada (updated_at, id de secuencia). Almacenarla en una tabla de estado, escrita en la misma transacción que la carga, nunca en un archivo aparte.
• Leer watermark - lookback (ej. 1 a 24 h), no > watermark. Las confirmaciones de la fuente no están ordenadas por timestamp; el lookback más el upsert absorbe filas tardías. Dimensionar el lookback según la latencia medida, no una suposición.
• Columna de marca de agua no confiable (no se actualiza en modificaciones, huso horario mixto, eliminaciones suaves faltantes): hacer una reconciliación completa periódica (diferencia de conjuntos de claves o count + checksum por partición) en lugar de confiar en ella.
• Las eliminaciones duras en la fuente son invisibles para las marcas de agua. Decidir explícitamente: reconciliar por diferencia de claves, o aceptar y documentar.
• CDC (replicación lógica de Postgres, lectores de registros estilo Debezium): usar solo cuando necesites eliminaciones, latencia de menos de un minuto, o la fuente no se pueda consultar por timestamp. Agrega slots, riesgo de retención (un slot inactivo llena el disco de la fuente) y manejo de ordenamiento/snapshots. De lo contrario, mantener las marcas de agua.

Backfills

• Cada trabajo recibe --start y --end (semiabierto [start, end), UTC) y deriva todo de ellos. Sin now() dentro de la transformación.
• Un backfill es la ejecución normal del trabajo sobre ventanas antiguas. Si necesita código diferente, el trabajo normal no es idempotente; corregir eso.
• Fragmentar por partición, ejecutar fragmentos de forma independiente, registrar el estado por fragmento, de modo que un fallo en el fragmento 40 se reanude en el 40.
• Hacer backfill en el mismo destino mediante el mismo upsert/reemplazo de partición. Verificar que no hay doble conteo: count(*) y count(distinct key) por ventana antes y después.
• Limitar la velocidad contra la API/BD fuente; un backfill es un pico de carga autoinfligido.

Reintentos, timeouts,