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

Einführung in Apache Airflow

Einen Task in Airflow testen

Airflow-Befehle untersuchen

Airflow-DAGs

Einen einfachen DAG definieren

Mit DAGs und der Airflow-Shell arbeiten

Fehlerbehebung bei der DAG-Erstellung

Airflow-Weboberfläche

Den Airflow-Webserver starten

In der Airflow-UI navigieren

DAGs mit der Airflow-UI untersuchen

Einführung in Airflow

Was geht, DAG? Jetzt lernst du die Grundlagen zur Implementierung von Airflow-DAGs. In praktischen Übungen richtest du Operatoren, Tasks und Scheduling ein und setzt sie ein.

Airflow-Operatoren

Eine BashOperator-Task definieren

Mehrere BashOperators

Airflow-Aufgaben

Reihenfolge von BashOperators festlegen

Die Reihenfolge von Tasks bestimmen

Fehlerbehebung bei DAG-Abhängigkeiten

Weitere Operatoren

Verwendung des PythonOperator

Weitere PythonOperators

EmailOperator und Abhängigkeiten

Airflow-Scheduling

Einen DAG per Python planen

Airflow-Zeitpläne entschlüsseln

Fehlerbehebung bei DAG-Runs

Airflow-DAGs implementieren

In diesem Kapitel lernst du, wie du mit Airflow-Komponenten wie Sensoren und Executorn Zeit sparst – während du Airflow-Workflows überwachst und Fehler behebst.

Airflow-Sensoren

Options

Sensors

Both

Operators

Sensoren vs. Operatoren

Sensorentzug

Airflow-Executors

Den Executor bestimmen

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2024,1,20),
    mode='poke',
    dag=report_dag
)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2024,1,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2024,1,20),
    mode='reschedule',
    dag=report_dag
)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2024,1,20),
    dag=report_dag
)

precheck >> generate_report_task


Auswirkungen des Executors

Debugging und Fehlerbehebung in Airflow

DAGs im Sack

from airflow import DAG
#from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='poke',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

sample_dag = DAG(
    dag_id = 'sample_dag',
    schedule_interval = "0 0 * * *"
)

sample_task = BashOperator(
    task_id='sample',
    bash_command='generate_sample.sh',
    start_date=datetime(2023,2,20),
    dag=sample_dag
)


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *"
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


Fehlender DAG

SLAs und Reporting in Airflow

Eine SLA definieren

Eine Task-SLA definieren

Einen Bericht erstellen und per E-Mail versenden

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

default_args={
    '____': [____],
    '____': True,
    'email_on_success': ____
}

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *",
    default_args=default_args
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.filesystem import FileSensor
from datetime import datetime

default_args={
    'email': ['airflowalerts@datacamp.com', 'airflowadmin@datacamp.com'],
    'email_on_failure': True,
    'email_on_success': True
}

report_dag = DAG(
    dag_id = 'execute_report',
    schedule_interval = "0 0 * * *",
    default_args=default_args
)

precheck = FileSensor(
    task_id='check_for_datafile',
    filepath='salesdata_ready.csv',
    start_date=datetime(2023,2,20),
    mode='reschedule',
    dag=report_dag)

generate_report_task = BashOperator(
    task_id='generate_report',
    bash_command='generate_report.sh',
    start_date=datetime(2023,2,20),
    dag=report_dag
)

precheck >> generate_report_task


Status-E-Mails hinzufügen

Airflow-Workflows warten und überwachen

Jetzt kommt alles zusammen. In diesem letzten Kapitel wendest du alles an, was du gelernt hast, um in Airflow einen produktionsreifen Workflow zu bauen.

Mit Templates arbeiten

Einen templated BashOperator erstellen

Templates mit mehreren Argumenten

Mehr Vorlagen

Listen mit Templates verwenden

Parameteroptionen verstehen

Vorlagenbasierte E-Mails senden

Verzweigung

Einen BranchPythonOperator definieren

Fehlerbehebung bei Branching

Eine Produktions-Pipeline erstellen

Eine produktionsreife Pipeline erstellen #1

Eine produktionsreife Pipeline erstellen #2

Die letzten Änderungen an deiner Pipeline vornehmen

Glückwunsch!

Produktions-Pipelines in Airflow aufbauen

Jetzt aktualisiert auf Apache Airflow 2.7 – Daten nach Zeitplan bereitzustellen, kann ein manueller Prozess sein. Du schreibst Skripte, pflegst komplexe Cron-Jobs und probierst verschiedene Wege aus, um ständig wechselnde Anforderungen zu erfüllen – und im Team wird die Verwaltung noch kniffliger. Apache Airflow nimmt dir diese Kopfschmerzen, indem es deinen Workflows Planung, Fehlerbehandlung 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 lernst du, wie du Directed Acyclic Graphs (DAGs) nutzt, Data-Engineering-Workflows automatisierst und Aufgaben im Data Engineering einfach und wiederholbar implementierst – damit du einen kühlen Kopf bewahrst.

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

Den Executor bestimmen

Einführung in Apache Airflow mit Python

Interaktive Übung