In diesem Kapitel erhältst du eine umfassende Einführung in die Komponenten von Apache Airflow und lernst, wie und warum du sie einsetzen solltest.

Einführung in Apache Airflow

Dag-Zähler

Deinen ersten Dag ausführen

Airflow-DAGs

Einen einfachen Dag erstellen

Mit Dags und der Airflow-Shell arbeiten

Airflow-Befehle untersuchen

Einen Task in Airflow testen

Airflow-Weboberfläche

In der Airflow-UI navigieren

Dags mit der Airflow-UI untersuchen

Dag-Fehler

Einführung in Airflow

Was geht, DAG? Jetzt lernst du die Grundlagen der Implementierung von Airflow-DAGs. In praktischen Übungen erfährst du, wie du Operatoren, Aufgaben und Zeitpläne einrichtest und bereitstellst.

Airflow-Operatoren

Einen Task dekorieren

Einen BashOperator mit @task.bash definieren

Reihenfolge der Tasks festlegen

Mehr @tasks

Dags planen

Einen Dag per Python planen

Airflow-Zeitpläne entschlüsseln

Probleme mit Dag-Runs beheben

Daten zwischen Tasks mit XCom übergeben

Daten mit XCom übergeben

XCom-Variablen

Airflow-Sensoren

Reihenfolge von Tasks bestimmen

Einen FileSensor erstellen

Sensorische Deprivation

DAGs in Airflow erstellen

In diesem Kapitel lernst du, wie du mit Airflow-Komponenten wie Sensoren und Executoren Zeit sparst und dabei Airflow-Workflows überwachst und Fehler behebst.

Monitoring, Benachrichtigungen und Callbacks

Eine Callback-Funktion implementieren

Einen SmtpNotifier implementieren

Was steht in den Logs?

Templating mit Jinja

Mit Jinja schreiben

Austauschen

Airflow-Variablen

Die Antwort ist Variable

Eine Variable in Python lesen

Variablen nicht ganz korrekt

Debugging und Fehlerbehebung in Airflow

Dag-Fehlersuche

Fehlender Dag

Airflow-Workflows warten und überwachen

Jetzt alles zusammenführen. In diesem letzten Kapitel setzt du alles Gelernte ein, um einen produktionsreifen Workflow in Airflow zu bauen.

Trigger und Fehlertoleranz

Eine Trigger-Regel implementieren

Retries hinzufügen

Einen Child-Dag auslösen

Verzweigungen

Definiere ein @task.branch

Fehlerbehebung bei Branches

Der HITL-Operator

Menschliche Freigabe hinzufügen

ApprovalOperator-Task

Eine Produktions-Pipeline erstellen

# Imports
from airflow.sdk import ____, ____, ____
from airflow.providers.standard.sensors.filesystem import ____
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='____',
  on_success_callback=____(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.____}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
    
  initialized = initialize()
  initialized >> filewatcher
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
      
  initialized = initialize()
  initialized >> filewatcher 
  
sales_etl()

Die Sales-ETL-Pipeline einrichten

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @____
  def validate_data():
    sales_path = Variable.____('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = '____'
    if '-09-' in sales_path:
      next_task = '____'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >> validated
  
  
sales_etl()

Monatliches Branching hinzufügen

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ____

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ____(
    task_id="____",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), ____] >> push_to_warehouse()
  
  
sales_etl()

# Imports
from airflow.sdk import dag, task, Variable
from airflow.providers.standard.sensors.filesystem import FileSensor
from airflow.providers.smtp.notifications.smtp import SmtpNotifier
from airflow.providers.standard.operators.hitl import ApprovalOperator

from pendulum import datetime

# Dag definition
@dag(
  start_date=datetime(2026,5,1),
  schedule='@monthly',
  on_success_callback=SmtpNotifier(
    to="airflow_team@mycompany.com",
    from_email="airflow_production@mycompany.com",
    subject="sales_etl run succeeded!"
  )
)
def sales_etl():
  
  @task
  def initialize(ds):
    print('Cleaning data directories, prepping for processing...')

    # Use a variable to set the current data path
    Variable.set("sales_etl_data_path", f'/data/sales_etl/{ds}')
  
  
  # Setup a file sensor to wait for the filescopied.txt file to be present
  filewatcher = FileSensor(
    task_id="WaitForDataFiles",
    filepath="{{var.value.sales_etl_data_path}}/filescopied.txt",
    poke_interval=15,
    timeout=300
  )
  
  
  # Create a branching task to validate the data. 
  @task.branch
  def validate_data():
    sales_path = Variable.get('sales_etl_data_path')
    print(f"Parsing data at {sales_path}")
    next_task = 'regular_monthly_task'
    if '-09-' in sales_path:
      next_task = 'yearend_approval_task'
    return next_task
  
  @task
  def regular_monthly_task():
    print(f'Processing data and automatically updating dataset')
    
  yearend_approval_task = ApprovalOperator(
    task_id="yearend_approval_task",
    subject="Sales data processing - Approval Required",
    body=(
            "Please review the sales data produced by the *validate_data* task "
            "on {{ds}}."
            "Approve to push to the data warehouse, or Reject to halt the run."
    ),
  )
  
  @task
  def push_to_warehouse():
    print(f'Pushing processed data to warehouse')
  
  initialized = initialize()
  validated = validate_data()
  
  initialized >> filewatcher >>validated
  
  validated >> [regular_monthly_task(), yearend_approval_task] >> push_to_warehouse()
  
  
sales_etl()

Einen manuellen Freigabeschritt hinzufügen

Glückwunsch!

DAG-Logik steuern

Jetzt aktualisiert auf Apache Airflow 3.1.6 – Daten nach Zeitplan zu liefern, kann ein manueller Kraftakt sein. Du schreibst Skripte, ergänzt komplexe Cron-Jobs und versuchst auf verschiedene Arten, wechselnde Anforderungen zu erfüllen – und im Team alles zu managen, ist noch schwieriger. Apache Airflow nimmt dir diese Kopfschmerzen ab, indem es deinen Workflows Planung, Fehlermanagement und Reporting hinzufügt. In diesem Kurs lernst du die Grundlagen von Apache Airflow und wie du komplexe Data-Engineering-Pipelines produktiv umsetzt. Außerdem erfährst du, wie du Directed Acyclic Graphs (DAGs) nutzt, Data-Engineering-Workflows automatisierst und Aufgaben einfach und wiederholbar implementierst – damit du einen kühlen Kopf behältst.

Intermediate Python

Introduction to Shell

Lerne, Daten-Workflows mit Airflow und Python zu planen, Fehler zu behandeln und Berichte zu erstellen.

Einführung in Apache Airflow mit Python

Dieser Kurs zeigt dir, wie du Data-Engineering-Workflows implementierst und planst.

Dateningenieur in Python

Monatliches Branching hinzufügen

Interaktive praktische Übung