Modèles, idempotence et remplissage rétroactif

Créer des pipelines de données avec Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Modèles Jinja dans Airflow

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

{{ ds }} produit la date logique au format YYYY-MM-DD
Chaque exécution a sa propre date, donc la même tâche génère un résultat différent par exécution
Fonctionne seulement dans les champs modélisables (comme bash_command, sql)

Le problème d'idempotence

Exécution 1 (31 mars) :

INSERT des ventes du 31 mars
Résultat : 3 lignes

Exécution 2 (relance du 31 mars) :

INSERT des ventes du 31 mars à nouveau
Résultat : 6 lignes (doublons !)

Doublons lors d'une relance

Le modèle supprimer-puis-insérer

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,
)

Vous pouvez aussi utiliser UPSERT ou MERGE

Supprimer puis insérer

Remplissage rétroactif (Backfilling)

Retraiter une plage de dates historiques
Airflow crée une exécution de DAG par date logique
Combiné à l'idempotence, le remplissage rétroactif est sécuritaire

Le remplissage dépend de l'horaire

Backfill dans l'interface

Déclenchez un DAG, choisissez Backfill, fixez des dates De et À
Choisissez Exécutions manquantes, Manquantes et en erreur ou Toutes
Réglez le parallélisme avec Max Active Runs et modifiez l'ordre d'exécution
Avec Run Backfill, le processus démarre et les exécutions sont créées

Interface de backfill d'Airflow

Passons à la pratique !

Créer des pipelines de données avec Airflow