DAGs ontwikkelen

Deze pagina is voor iedereen die pipelinecode schrijft voor de Airflow-instantie van dit platform: wat een DAG is, hoe je code daadwerkelijk op de instantie terechtkomt, hoe je er één ontwikkelt en test, en de werkwijzen die voorkomen dat DAGs elkaar of zichzelf breken.

Wat is een DAG

Een DAG (Directed Acyclic Graph) is de planningseenheid van Airflow: een Python-bestand dat een set taken declareert en de volgorde waarin ze moeten draaien — “draai B nadat A geslaagd is,” nooit een cyclus terug naar iets dat al gedraaid heeft. Airflow leest dat bestand volgens een schema, en voert voor elke verschuldigde run de taken uit in afhankelijkheidsvolgorde, met retries of alerting per taak bij een fout. Het DAG-bestand zelf doet het werk niet — het koppelt operators aan elkaar, elk in staat om één soort taak uit te voeren (een notebook draaien, een API aanroepen, wachten op een bestand). Zie Custom Airflow providers voor de operators die dit platform bovenop de ingebouwde operators van Airflow toevoegt.

Hoe je DAG-code bij deze Airflow-instantie terechtkomt

Er is geen handmatige deploystap en geen bestanden kopiëren. De Airflow-instanties van dit platform zijn geconfigureerd met een Git DAG bundle (apache-airflow-providers-git-azure) die je DAG-repository rechtstreeks ophaalt — voor Azure DevOps-repo’s zelfs helemaal zonder statische credential (PAT/SSH-key), via workload identity. De bundle volgt één branch (of tag) per Airflow-instantie, dus:

  • Pushen naar die gevolgde branch (doorgaans main) is wat een DAG “deployt”. Er is verder niets om uit te voeren.

  • Een niet-productie Airflow-instantie kan een andere branch volgen door tracking_ref te overschrijven op zijn eigen connection (zie de sectie “Overriding the tracked branch” in apache-airflow-providers-git-azure). Gebruik dit om een DAG te testen voordat deze een instantie bereikt waar mensen op vertrouwen.

  • Nieuwe taken, gewijzigde schema’s, alles in het DAG-bestand: het wordt allemaal automatisch opgepikt de volgende keer dat de bundle synchroniseert. Er komt geen Airflow-herstart aan te pas.

Een DAG ontwikkelen

Begin bij een notebook, als de job datawerk is. De apache-airflow-providers-notebook (SparkNotebookOperator) van het platform draait een Jupyter-notebook als een geplande, cluster-mode Spark-job — hetzelfde notebook dat je al interactief kunt draaien in JupyterHub, tegen dezelfde image en dezelfde Delta Lake catalog. Dit is het bedoelde pad voor “maak van analyse een productiejob”: schrijf en valideer het notebook eerst interactief, en verpak het daarna in een DAG met één taak die het volgens een schema draait. Je zou zelden zelf Spark-submissielogica hoeven te schrijven.

from datetime import datetime
from airflow import DAG
from airflow.providers.notebook.operators.spark_notebook import (
    SparkNotebookOperator,
)

with DAG(
    dag_id="daily_open_data_refresh",
    schedule="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["open-data"],
) as dag:
    SparkNotebookOperator(
        task_id="run_notebook",
        repo_url="https://dev.azure.com/<org>/<project>/_git/example-dags",
        repo_branch="main",
        notebook_path="pipelines/open_data_notebook.ipynb",
    )

Voor werk zonder notebook gebruik je de andere platformproviders (Custom Airflow providers) of de eigen upstream providers van Airflow waar die van toepassing zijn — bijvoorbeeld git_default-achtige connections voor repotoegang, of de mail/Fabric/Power BI/SSIS-providers die dit platform meelevert voor die specifieke integraties. Grijp pas naar een kale PythonOperator/@task als niets anders past; alles wat Spark-werk uitvoert hoort via SparkNotebookOperator te lopen in plaats van een zelfgebouwde submissie, zodat het automatisch de image, catalog en credential-bekabeling van het platform overneemt in plaats van die per DAG opnieuw af te leiden.

Testen voordat het een instantie bereikt waar mensen op vertrouwen. Wijs de connection van een niet-productie Airflow-instantie naar je feature branch (zie hierboven), laat hem synchroniseren, en bevestig dat de DAG daar parseert en draait — het paneel DAG import errors (zichtbaar voor Airflow Admin, zie Toegang beheren: Airflow-rollen) is de snelste manier om een syntax- of importfout te vinden zonder op een geplande run te wachten. Merge pas naar de gevolgde productiebranch zodra hij schoon draait.

Best practices

Tasks get retried, and runs get cleared and re-triggered by hand, so design them to be idempotent: running the same task twice with the same inputs should produce the same result, not a duplicate row or a corrupted state. Overwriting or upserting a partition works; appending to it doesn’t.

Zet geen toegangssleutel, wachtwoord of token in een DAG-bestand, zelfs niet als “voorbeeld”. Lees het in plaats daarvan uit een Airflow connection. De eigen operators van het platform zijn op dezelfde manier gebouwd: SparkNotebookOperator biedt bewust geen enkele credential-achtige parameter (zie apache-airflow-providers-notebook voor wat vastligt op deploymentniveau in plaats van DAG-instelbaar), dus is er geen manier om per ongeluk af te wijken van hoe het platform daadwerkelijk beveiligd is.

Zet catchup=False tenzij je daadwerkelijk backfill nodig hebt. Zonder dit triggert een DAG met een start_date in het verleden, zodra deze wordt gedeployt, een run voor elk gemist interval sindsdien, allemaal tegelijk. Zelden wat je wilt voor een operationele pipeline.

Geef elke DAG retries en een redelijke timeout, op taakniveau (retries, retry_delay) of via default_args voor de hele DAG. Een trage upstream-API of een tijdelijk resourcetekort zou niet moeten vereisen dat een mens het opmerkt en de run handmatig opnieuw triggert.

Benoem en tag DAGs zodat eigenaarschap in één oogopslag duidelijk is. dag_id en tags zijn het enige dat een ander team ziet wanneer je DAG falend verschijnt in de gedeelde Grid-view, dus gebruik iets team- of domeingebonden in plaats van een generieke naam zoals pipeline of test.

Houd de blast radius van één DAG tot één DAG beperkt. Twee DAGs die onafhankelijk van elkaar naar dezelfde tabel schrijven op eigen schema’s is een race condition in wording. Als ze echt een afhankelijkheid delen, modelleer die dan expliciet (een sensor, of een dataset-aware schedule) in plaats van het aan toeval van timing over te laten.

En controleer de Grid/Graph-view en task-logs voordat je aanneemt dat een fix werkte. Een DAG die parseert en een DAG die correcte output produceert zijn twee verschillende garanties; Airflow controleert automatisch alleen de eerste voor je.