Plantillas, idempotencia y backfilling

Creación de canalizaciones de datos con Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Plantillas Jinja en Airflow

@task.bash
def export_data():
    return "cp /data/sales.csv /export/sales_{{ ds }}.csv"

{{ ds }} muestra la fecha lógica en formato YYYY-MM-DD
Cada ejecución tiene su propia fecha, así que la misma tarea produce salidas distintas por ejecución
Solo funciona en campos con plantilla (como bash_command, sql)

El problema de la idempotencia

Ejecución 1 (31 de marzo):

INSERT de ventas del 31 de marzo
Resultado: 3 filas

Ejecución 2 (re-ejecución del 31 de marzo):

INSERT de ventas del 31 de marzo otra vez
Resultado: 6 filas (¡duplicados!)

Filas duplicadas al re-ejecutar

El patrón eliminar‑e‑insertar

staged_rows = SQLExecuteQueryOperator(
    task_id="get_staged_rows",
    conn_id="duckdb_default",
    sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
)

load = SQLInsertRowsOperator(
    task_id="load_sales",
    conn_id="duckdb_default",
    table_name="sales",
    columns=["date", "product", "amount"],
    preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
    rows=staged_rows.output,
)

También puedes usar enfoques UPSERT o MERGE

Eliminar y luego insertar

Backfilling

Reprocesa un rango de fechas históricas
Airflow crea un run del DAG por fecha lógica
Combinado con idempotencia, el backfilling es seguro

El backfill depende de la planificación

Backfill en la interfaz

Dispara un DAG, selecciona Backfill, y define fechas From y To
Puedes elegir Missing Runs, Missing and Errored Runs o All Runs
Ajusta el paralelismo con Max Active Runs y cambia el orden de ejecución
Con Run Backfill empieza el proceso y se crean los runs

Interfaz de backfill de Airflow

¡Vamos a practicar!

Creación de canalizaciones de datos con Airflow