Airflow Best Practice - Dokumentation für DAG Autoren

• Airflow: 3.x
• Infrastruktur: Helm Deployment auf Kubernetes, Celery Executor + Kubernetes Executor
• Zielgruppe: Data Engineers, Data Analysten und alle, die DAGs schreiben

---

1. Einleitung

Dieses Dokument beschreibt verbindliche Regeln und Empfehlungen für das Schreiben von Apache Airflow DAGs (Directed Acyclic Graphs) in der Datenfabrik. Ziel ist es, dass mehrere Teams produktiv und sicher auf derselben Plattform arbeiten können, ohne sich gegenseitig zu blockieren.

Die drei Leitprinzipien, die sich durch das gesamte Dokument ziehen:

• Atomarität – Jeder Task erledigt genau eine Aufgabe (Extract oder Transform oder Load).
• Idempotenz – Gleiche Eingabe liefert immer das gleiche Ergebnis, auch bei erneutem Ausführen.
• Modularität – Wiederverwendbare Logik wird in separate Module ausgelagert, der DAG selbst bleibt möglichst schlank.

Quellen zum Vertiefen:

• Airflow DAGs (offiziell)
• Airflow Best Practices (offiziell)
• Astronomer: Best Practices for ETL/ELT with Airflow 3

---

2. Die Basics

Dieser Abschnitt vermittelt dir, die Grundprinzipien für die Erstellung eines DAG in der Datenfabrik: das grundlegende ETL-Pattern, die Ablagestruktur im Repository und wie Daten zwischen Tasks ausgetauscht werden.

2.1 ETL-Pattern

Unsere DAGs folgen dem ETL-Pattern (Extract → Transform → Load) — entsprechend dem Leitprinzip Atomarität erledigt jeder Task genau einen dieser Schritte:

1. Extract: Daten aus der Quelle lesen (API, S3/MinIO, Datenbank, Datei(-upload)).
2. Transform: Daten im Airflow-Worker verarbeiten (Pandas, Python). Aktuell leichte bis mittelschwere Transformationen.
3. Load: Ergebnis ins Zielsystem schreiben, möglichst vorher validieren (→ Data Quality Checks).

Zielsystem ist dabei im Normalfall:

• Postgres – die Use-Case-spezifische Fachdatenbank im selben Kubernetes-Cluster. Insbesondere relevant für weitere Self-Service Tools.
• S3 (MinIO) – wenn das Ergebnis als Datei bereitgestellt wird (Parquet, JSON, CSV).
• API - externe APIs

2.2 Verzeichnisstruktur & Datenfabrik Libraries

Jeder Use Case in der Datenfabrik erhält ein eigenes Git-Repository, das die dags des Use Cases enthält. Die Struktur in deinem Use Case Repository sieht beispielhaft so aus:

dags/
├── .airflowignore                          Ausnahmen für den DAG-Parser
├── 01_pull_data.py                         DAG-Dateien
├── 02_transform.py
├── 03_csv_to_parquet.py
└── modules/                                Use-Case-spezifische Module
    ├── __init__.py
    ├── transformation_helpers.py
    └── mapdata_generator.py

Regeln

• Jede DAG-Datei enthält genau einen DAG.
• Businesslogik (je Task), die mehr als 10–15 Zeilen umfasst, wird in modules/ ausgelagert.
• Domänenspezifische Module liegen ebenso unter modules/.
• Jeder modules/-Ordner braucht eine __init__.py.

Datenfabrik Libraries

Das Python-Package datenfabrik-libraries ist in dem von der Datenfabrik verwendeten Airflow Image bereits installiert und stellt plattformspezifische Hooks und Helfer bereit (z. B. DF_S3Hook, FileAccessS3). Die vollständige Dokumentation mit Installation und Verwendungsbeispielen findet sich in Custom Airflow Library der Datenfabrik.

2.3 Datenaustausch zwischen Tasks (XCom vs. S3)

S3 (aktuell MinIO) dient als Zwischenspeicher zwischen Tasks. Statt große Datenmengen über XCom zwischen Tasks zu übergeben empfehlen wir diese explizit in S3 abzulegen:

[Quelle] → Extract → S3 → Transform → S3 (Ergebnis) oder Postgres

Aufbewahrungsdauer der Daten

Um zu steuern, wie lang diese Daten aufbewahrt werden, müssen entsprechende S3 Tags raw/intermediate/output genutzt werden. Es ist davon auszugehen dass Eingangsdateien (raw) und Ergebnisdaten (output) aus Transparenz-/Nachvollziehbarkeitsgründen langfristig aufbewahrt werden. Dem gegenüber sind Zwischenergebnisse (intermediate) nur vorübergehend relevant.

Daher werden nur DAGs akzeptiert, welche unseren Datenfabrik S3 Hook ansprechen. (s. Doku )

Regeln für den Datenaustausch zwischen Tasks

Was übergeben?	Methode	Beispiel
Metadaten, Dateipfade, kleine Konfigurationen (möglichst \< 1 KB)	XCom (Standard)	Dateiname, Zeilenzahl, Status Im Ausnahmefall sind auch XCom \> 1 KB hier werden die Daten dann in S3 abgelegt.
Tabellarische Daten, DataFrames, Dateien	S3 (explizit)	CSV/Parquet in Bucket ablegen, Pfad per XCom übergeben

---

3. Verbindliche Code-Regeln

Dieser Abschnitt definiert die Regeln, die für jeden DAG in der Datenfabrik gelten. Kapitel mit ⚠ Pflicht sind nicht verhandelbar — DAGs, die dagegen verstoßen, werden nicht akzeptiert. Kapitel mit 💡 Empfehlung sind Best Practices, deren Nichteinhaltung begründet werden muss.

3.1 Ruff — 💡 Empfehlung

Das Linting-Tool ruff unterstützt die Codequalität zu verbessern. Bspw. erkennt es veraltete Muster und potenzielle Fehler.

pip install "ruff>=0.14.14"

Ausführung von ruff um potenzielle Probleme frühzeitig zu erkennen:

ruff check dags/ --select AIR3

3.2 Immer TaskFlow API verwenden — ⚠ Pflicht

Wir nutzen ausschließlich die TaskFlow API mit dem @task Dekorator

# Richtig
@task(retries=2)
def extract_data() -> list[str]:
    return ["file1.csv", "file2.csv"]

# Falsch
extract_data = PythonOperator(
    task_id="extract_data",
    python_callable=_extract_data,
)

3.3 Datenfabrik Hooks bevorzugen — ⚠ Pflicht

Im Rahmen der Datenfabrik entstehen eigene Hooks/Operators, da gewisse Standards gesetzt werden sollen. Alle Datenfabrik-Komponenten sind im Package datenfabrik-libraries (s. Custom Airflow Library der Datenfabrik) enthalten.

Kategorie	Einsatz	Airflow Standard	Datenfabrik Standard	Erklärung	Link zur weiterführenden Doku
Hook	S3	from airflow.providers.amazon.aws.hooks.s3 import S3Hook	from datenfabrik_libraries.hooks import DF_S3Hook	Der DF_S3Hook ist ein Wrapper um den Standard-Airflow S3Hook, der die Verwendung des data-type Tags verpflichtend macht. Dies gewährleistet eine konsistente Datenkategorisierung in der Datenfabrik SH. Erlaubte data-type Werte: • raw - Rohdaten aus externen Quellen • intermediate - Verarbeitete/transformierte Daten • output - Finale Ausgabedaten für den Konsum	DF_S3Hook Dokumentation

3.4 Kein Top-Level Code — ⚠ Pflicht

Airflow parst alle DAG-Dateien regelmäßig. Code, der außerhalb von Tasks und der DAG-Funktion steht, wird bei jedem Parse ausgeführt.

# FALSCH: Wird bei jedem Parse ausgeführt!
hook = PostgresHook("pg_conn")
results = hook.get_records("SELECT * FROM my_table;")

# RICHTIG: Nur innerhalb von Tasks
@dag(...)
def my_dag():

    @task()
    def get_data():
        hook = PostgresHook("pg_conn")
        return hook.get_records("SELECT * FROM my_table;")

Erlaubt auf Top-Level: Imports, Konstanten (Strings, Zahlen), statische Dictionaries (z.B. CSV_READ_PARAMS), default_args.

Nicht erlaubt auf Top-Level: Hook-Instanziierungen, API-Calls, Dateioperationen, ...

3.5 Nutzung von .airflowignore — 💡 Empfehlung

In der .airflowignore sollten Ausnahmen aufgeführt werden, welche der Parser nicht auf Airflow Dags untersuchen soll, was je nach Umfang - wenn nicht in der .airflowignore definiert - zu einer erheblichen Verlangsamung dieses Vorgangs führen kann (Dokumentation).

3.6 Dokumentation — 💡 Empfehlung

Datei-Docstring (ganz oben): Beschreibt den DAG grob, sodass man schnell erkennen kann was im DAG geschieht

"""
Airflow DAG Summary:
Last Step for Use Case: Serve all data as Parquet and create Mapdata JSON.

Task A) CSV → Parquet Konvertierung
Task B) Mapdata-JSON-Generation
Task C) FastAPI Reload
"""

DAG-Docstring (auf der Funktion): Markdown-Format, wird in der Airflow UI gerendert (DAG Docs):

@dag(...)
def csv_to_parquet_pipeline():
    """
    ## CSV to Parquet Conversion

    Converts all CSV files from the source folder to Parquet and writes
    them into the target folder within the same S3 bucket.

    **Dependency**: Waits for successful completion of `uc_zufish_02_transform` DAG.
    """

3.7 Task Dependencies — 💡 Empfehlung

# Standard Vorgehen mit "chain"
from airflow.sdk import task, dag, chain
chain(
    check_for_files,
    csv_files,
    converted,
    mapdata,
    reload
    )

3.8 Retries konfigurieren — ⚠ Pflicht

Jeder Task, der mit externen Systemen kommuniziert, muss Retries konfiguriert haben. retry_delay gibt die Wartezeit zwischen zwei Versuchen an — wichtig, um externe Systeme nicht sofort erneut zu belasten:

from datetime import timedelta

default_args = {
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    ...
}

# Einzelner Task überschreibt
@task(retries=4, retry_delay=timedelta(minutes=2))
def collect_csv_files() -> list[str]:
    ...

3.9 Naming Conventions — ⚠ Pflicht

Einheitliche Namen stellen sicher, dass DAGs, Connections und Assets verschiedener Use Cases in der gemeinsam genutzten Airflow-Instanz konfliktfrei koexistieren.

Connections & Variablen

Format: - AIRFLOW_CONN__UC_{Use Case Kürzel}__{Connection Type} - AIRFLOW_VAR__UC_{Use Case Kürzel}___{VERWENDUNG}

Hinweis: doppelter Unterstrich vor dem Use-Case-Kürzel, dreifacher Unterstrich zwischen Kürzel und Connection-ID. Das Format ist kompatibel mit dem Airflow Multi-Team-Setup.

AIRFLOW_CONN__UC_LBV_DATEX___HTTP

Es hilft der Übersicht im Code und späteren Maintainer/Reviewern, die Connections und Variablen möglichst weit oben in einem Block aufzulisten und nicht im Code zu verteilen.

DAG-Dateinamen

Die Benennung orientiert sich an der fachlichen Funktion des DAGs. Sie muss den Use-Case-Namen nicht wiederholen, da dieser bereits durch die Repository-Struktur gegeben ist. Sollte jedoch ansonsten der DAG-ID entsprechen.

Wenn DAGs aufeinander aufbauen oder sich gegenseitig triggern, kann die Reihenfolge als Nummern-Präfix dargestellt werden:

publish_roads_datex.py

1_download_data.py
2_transform.py

DAG-IDs (dag_id)

Format: uc_{Use Case Kürzel}_{Aktion}

Bei einem einzigen DAG pro Use Case:

uc_zit_zufish_gitlab_analytics_orchestration

Bei mehreren DAGs pro Use Case mit klarer Reihenfolge:

uc_zit_zufish_reporting_backend_1_download_data
uc_zit_zufish_reporting_backend_2_transform

Assets

Format: uc_{Use Case Kürzel}_{Connection Type}_{Asset Name}_{Asset Typ}

# Tabelle in einer Datenbank
db_table_asset = Asset("uc_zit_zufish_gitlab_analytics_postgres_tasks_table")

# Konkrete Datei in S3
raw_file_asset = Asset(
    uri="s3://uc-zit-zufish-gitlab-analytics/raw/tasks.json",
    name="uc_zit_zufish_gitlab_analytics_s3_tasks_file"
)

# Übergeordneter Ordner in S3
raw_folder_asset = Asset(
    uri="s3://uc-zit-zufish-reporting-backend/raw",
    name="uc_zit_zufish_reporting_backend_s3_raw_data_folder"
)

---

3.10 Keine Hardcoded Credentials — ⚠ Pflicht

Passwörter, API-Keys, Tokens und alle anderen Zugangsdaten dürfen nie im DAG-Code hardcoded sein. Credentials sollten über Airflow Connections referenziert werden.

---

4. Fortgeschrittene Patterns

Dieser Abschnitt behandelt Patterns, die über einen einfachen linearen ETL-DAG hinausgehen.

4.1 Dynamic Task Mapping

Wenn die Anzahl der zu verarbeitenden Elemente erst zur Laufzeit bekannt ist (z.B. die Anzahl CSV-Dateien in einem S3 Folder), nutzen wir Dynamic Task Mapping mit .expand().

@task()
def collect_csv_files() -> list[str]:
    """Collect all CSV file keys from S3."""
    return list_csv_files(MINIO_CONN_ID, BUCKET, CSV_PREFIX)

@task(retries=2, max_active_tis_per_dag=3)
def convert_to_parquet(file_key: str, **kwargs) -> dict:
    """Convert a CSV to Parquet and store it on S3."""
    # ... Logik ...
    return {"source": basename, "target": parquet_key, "rows": len(df)}

# Taskflow: 1 Task-Kopie pro Datei, automatisch parallelisiert
csv_files = collect_csv_files()
converted = convert_to_parquet.expand(file_key=csv_files)

Airflow erzeugt zur Laufzeit je eine Task-Instanz pro Element. Wenn eine Datei fehlschlägt, werden nur die fehlgeschlagenen Instanzen erneut ausgeführt, nicht der gesamte Batch.

4.2 Abhängigkeiten zwischen DAGs

Wenn eine Fachlogik mehrere DAGs umfasst (z.B. erst Daten beschaffen, dann transformieren, dann laden), gibt es unterschiedliche Mechanismen diese Abhängigkeit zu definieren:

4.2.1 Assets (Empfehlung)

Assets in Airflow repräsentieren Daten wie z.B. eine Datei in S3 oder eine Tabelle in einer Datenbank. Airflow selbst interagiert allerdings nicht mit dem dahinter liegenden Asset, sondern sieht es wie eine Art Flag, die ein Task setzt, wenn er erfolgreich war.

Für die Definition eines Assets reicht eigentlich ein Name, aus Gründen der Nachvollziehbarkeit ist allerdings empfohlen für Dateien, die einen fixen Pfad haben diesen auch als "uri" zu setzen. Werden viele Dateien innerhalb eines Ordners geändert kann hier auch der Pfad zum Ordner gewählt werden. Für die Unterscheidung der Assets von unterschiedlichen Usecases sollte auch bei den Assets immer der entsprechende Use-Case-spezifische Prefix im Namen verwendet werden.

from airflow.sdk import Asset

# einfachste Definition eines Assets z.B.einer Tabelle
db_table_asset = Asset("uc_usecasename_postgres_tasks_table")

# Referenzierung auf konkrete Datei in S3
raw_data_file_asset = Asset(
    uri="s3://usecases3bucket/raw/data.json",
    name="uc_usecasename_s3_raw_data_file"
)

# Referenzierung auf übergeordneten Ordner in S3
raw_data_folder_asset = Asset(
    uri="s3://usecases3bucket/raw",
    name="uc_usecasename_s3_raw_data_folder"
)

Wenn ein Task ein Asset aktualisiert kann dies beim Task unter "outlets" angegeben werden.

Ein DAG, das darauf wartet, dass dieses Asset aktualisiert wurde, also seinen "schedule" auf dieses Asset eingestellt hat, startet dann automatisch, sobald der Task, der das Asset aktualisiert, erfolgreich ausgeführt wurde.

from airflow.sdk import Asset

raw_data_file_asset = Asset(
    uri="s3://usecases3bucket/raw/data.json",
    name="uc_usecasename_s3_raw_data_file"
)

# DAG 1: Producer - aktualisiert Asset
@dag(
    dag_id="uc_usecasename_1_extract",
    schedule="0 * * * *", 
...
)
def extract_raw_data():

    @task(outlets=[raw_data_file_asset])
    def upload():
        """Schreibt Rohdaten nach S3."""
    ...

# DAG 2: Consumer – startet wenn Asset aktualisiert wurde
@dag(
    dag_id="uc_usecasename_2_transform"
    schedule=[raw_data_file_asset],
    ...
)
 ...

Ein DAG kann auch von mehreren Assets abhängig sein (also erst starten wenn alle dieser Assets aktualisert wurden).

from airflow.sdk import Asset

raw_data_file_asset_1 = Asset(
    uri="s3://usecases3bucket/raw/data1.json",
    name="uc_usecasename_s3_raw_data_file_1"
)

raw_data_file_asset_2 = Asset(
    uri="s3://usecases3bucket/raw/data2.json",
    name="uc_usecasename_s3_raw_data_file_2"
)

# Consumer – startet wenn alle Assets aktualisiert wurde
@dag(
    dag_id="uc_usecasename_2_transform"
    schedule=[raw_data_file_asset_1 & raw_data_file_asset_2],
    ...
)
 ...

Zusätzliche Daten, die das Asset betreffen, können entweder über Asset Extras (falls sie sich bei der Aktualisierung des Assets nicht ändern) oder Asset Event Extras (falls sie sich bei jeder Aktualisierung des Assets ändern, wie z.B. dynamische Pfade) an vom Asset abhängige DAGs übergeben werden.

Die Vorteile in der Benutzung von Assets gegenüber den nachfolgenden Optionen zur Auflösung von Abhängigkeiten zwischen einzelnen DAGs:

• Es besteht keine direkte Abhängigkeit zwischen den DAGs, sondern nur zu den aktualisierten Assets
• Im Airflow UI werden die DAGs, die Assets erzeugen und die von diesen Assets abhängigen DAGs übersichtlich im Bereich Assets dargestellt.

4.2.2 TriggerDagRunOperator (Alternative)

Über TriggerDagRunOperator kann innerhalb eines DAGs definiert werden, dass danach ein anderes bestimmtes DAG ausgeführt werden soll. Das ist eine direkte harte Kopplung, da DAG 1 die dag_id von DAG 2 kennen muss. Die Abhängigkeiten zwischen den DAGs werden im Airflow UI nicht angezeigt, anders als bei den Assets. Dafür ist so sichergestellt dass DAG 2 auf jeden Fall direkt nach DAG 1 ausgeführt wird. Wenn Assets nicht verwendet werden können ist das die Alternative.

    trigger_parquet = TriggerDagRunOperator(
        task_id="trigger_parquet_dag",
        trigger_dag_id=DOWNSTREAM_DAG_ID,
        wait_for_completion=False,
        reset_dag_run=False,
    )

4.2.3 ExternalTaskSensor (nicht nutzen)

Der ExternalTaskSensor wartet aktiv darauf, dass ein bestimmter Task in einem anderen DAG erfolgreich war. Da hier die Nachteile wie der Verbrauch von Worker-Slots während des Wartens, harte Kopplung an DAG-ID und Task-ID sowie keine Sichtbarkeit in der UI überwiegen sollte der ExternalTaskSensor nicht benutzt werden.

from airflow.providers.standard.sensors.external_task import ExternalTaskSensor

wait_for_pull = ExternalTaskSensor(
    task_id="wait_for_data_pull",
    external_dag_id="uc_xyz_1_pull_data",
    external_task_id="load_complete",
    timeout=3600,
    poke_interval=60,
)

4.3 Data Quality Checks

Jede Pipeline sollte vor dem Load in das Zielsystem die Daten validieren, um zu vermeiden dass ungültige Daten ausgeliefert werden, welche erst in der Folgeverarbeitung im angeschlossenen System sichtbar werden. Dies kann von einem reinen quantitativen Check bis hin zu einer XML-Schema Validierung gehen.

Als Mindestanforderung wird empfohlen, nach dem Transform-Schritt zumindest einen quantitativen Check durchzuführen (z.B. Zeilenzahl > 0, keine vollständig leeren Spalten), bevor Daten ins Zielsystem geschrieben werden.

---

5. Infrastruktur & Betrieb

5.1 Executor-Wahl

5.1.1 Aktueller Stand: Celery Executor Default

Unsere Airflow-Instanzen nutzen aktuell den Celery Executor als Default. Dabei laufen alle Tasks auf einem/mehreren gemeinschaftlich genutzten Worker-Pod(s). Das funktioniert gut für I/O-lastige Tasks (API-Calls, S3-Lese/Schreibvorgänge, SQL-Statements), birgt aber ein Risiko bei rechenintensiven Operationen insb. Python Transformationen mit bspw. pandas sowie der Separierung von Dags auf File-System Ebene.

Problem 1: Wenn ein Task große Pandas DataFrames im Worker-Prozess verarbeitet, teilt er sich den Arbeitsspeicher mit allen anderen Tasks auf demselben Worker Pod. Bei mehreren solcher Tasks gleichzeitig droht ein Out-of-Memory Zustand, der den gesamten Worker-Pod zum Absturz bringt – und damit alle darauf laufenden Tasks aller Teams (im besten Fall werden diese retried und laufen im nächsten Anlauf durch).

Problem 2: sollte ein Task Daten direkt ins Dateisystem schreiben, ist nicht sichergestellt dass nicht ein anderer Task / ein anderer DAG auf diese Daten zugreifen kann. Es sollte daher insgesamt vermieden werden, Daten im Filesystem abseits von S3/XCom zu persistieren.

5.1.2 KubernetesExecutor ist via Multi-Executor-Konfiguration verfügbar

Wenn schwergewichtige Tasks via KubernetesExecutor (Dokumentation) ausgeführt werden, kann für diesen Task entsprechend RAM vom Kubernetes Cluster reserviert werden, welcher nicht mit anderen Tasks geteilt wird

Airflow bietet das Feature "Multiple Executors Concurrently" . Damit lassen sich z.B. CeleryExecutor und KubernetesExecutor parallel betreiben, was aktuell durch die Datenfabrik unterstützt wird.

Das Routing erfolgt über den @Task Decorator via executor=

• Tasks ohne explizite Deklaration → laufen auf den Celery-Workern (Standard, schnell, kein Pod-Overhead).
• Tasks mit executor="KubernetesExecutor"' → laufen in einem eigenen Kubernetes-Pod, Task Abarbeitung dauert länger, da erst ein Pod erstellt werden muss

# Task auf den Celery-Workern (Standard)
@task(retries=2)
def extract_from_api() -> dict:
    ...

# Rechenintensiver Task → eigener Pod
@task(retries=2, executor="KubernetesExecutor")
def heavy_pandas_transform(data: dict) -> dict:
    ...

5.1.3 KubernetesPodOperator (KPO)

Der KubernetesPodOperator (Dokumentation) startet einen Task in einem separaten Kubernetes-Pod mit einem eigenen Docker-Image. Das ist nur nötig, wenn:

• Die Transformation nicht in Python möglich ist (z.B. R-Script, Java, ...).
• Ein Team ein eigenes Container-Image mit speziellen Dependencies verwaltet. (sollte die Ausnahme sein)
• Ein komplett anderes Python Runtime-Environment benötigt wird (andere Libraries, andere Python-Version). Wobei in diesem Fall bevorzugt ist, den Task besser via @task.virtualenv Decorator auszuführen (Dokumentation)

Für Python-basierte Transformationen (Pandas, etc.) reicht die Multi-Executor-Konfiguration (s.o.) mit executor="KubernetesExecutor" aus – der Task nutzt dasselbe Airflow-Image, bekommt aber seinen eigenen Pod mit isoliertem Speicher und wird ggf. optimiert auf einer freieren Worker Node gescheduled.

5.1.4 Entscheidungshilfe für DAG-Autoren

Unabhängig davon, welcher Executor konfiguriert ist, sind die wichtigsten Leitplanken:

• Airflow ist primär ein Orchestrator, kein spezialisierte Compute-Engine
• Der (Celery) Worker soll koordinieren und leichte Tasks abarbeiten, nicht rechnen.

In der Praxis heißt das:

• Kleine DataFrames (\< 100 MB) können im Worker verarbeitet werden.
• Mittlere Transformationen (100–500 MB) sollten per executor="KubernetesExecutor" isoliert werden
• Sehr große Datenmengen => ggf. andere Tooling nicht Airflow-basierend

5.2 Scaling-Parameter

Der DAG Autor sollte sich Gedanken über Skalierungsmöglichkeiten seiner DAGs und Tasks machen. Hier die wesentlichen Ebenen.

Environment-Level (gesamte Instanz, durch Datenfabrik Operations Team festgelegt):

• Wenn der default Celery Worker verwendet wird , haben wir aktuell eine Concurrency Begrenzung von "2"

DAG-Level (einzelner DAG):

Der DAG Author kann und sollte aus seiner Sicht festlegen:

• max_active_runs=1: Verhindert parallele DAG-Runs, des gleichen DAGs
• max_active_tasks: Max. gleichzeitige Tasks innerhalb eines DAG-Runs.

@dag(
    max_active_runs=1,
    max_active_tasks=5,
    ...
)

Task-Level (einzelne Tasks):

• max_active_tis_per_dag: Parallele Instanzen dieses Tasks begrenzen. Bei expand() nutzen

@task(retries=2, max_active_tis_per_dag=3)
def insert_into_db(file_key: str):
    ...

5.3 Logging

Für das Logging in Tasks sollte der Python-Standard-Logger (via logging.getLogger(__name__)) verwendet werden.

import logging

logger = logging.getLogger(__name__)

Log-Level nach Umgebung:

Umgebung	Angezeigtes Level
stage1, int (Entwicklungsumgebungen)	INFO und höher
alle weiteren Umgebungen (inkl. pilot)	WARNING und höher

Beispiel:

@task()
def transform_data(records: list) -> list:
    if not records:
        # Warning - Wird auf allen Umgebungen angezeigt
        logger.warning("No data records received for processing.")
        return []
    # Info - Wird nur auf Entwicklungsumgebungen angezeigt:
    logger.info("Processing %d data records.", len(records))
    result = data_transform(records)
    return result