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

Einführung in Apache Airflow

Eine Aufgabe in Airflow testen

Prüfung der Luftstrombefehle

Luftstrom DAGs

Definieren einer einfachen DAG

Arbeiten mit DAGs und der Airflow-Schale

Fehlersuche DAG Erstellung

Airflow Web Interface

Den Airflow Webserver starten

Navigieren im Luftstrom UI

Untersuche DAGs mit dem Airflow UI

Einführung in die Luftströmung

Was ist los DAG? Jetzt ist es an der Zeit, die Grundlagen der Implementierung von Airflow DAGs zu lernen. In praktischen Übungen lernst du, wie du Operatoren, Aufgaben und Planungen einrichtest und einsetzt.

Luftstrombetreiber

Definieren einer BashOperator Aufgabe

Mehrere BashOperatoren

Aufgaben im Luftstrom

Reihenfolge der BashOperatoren festlegen

Festlegen der Reihenfolge der Aufgaben

Fehlersuche DAG Abhängigkeiten

Zusätzliche Betreiber

Verwendung des PythonOperators

Mehr PythonOperatoren

EmailOperator und Abhängigkeiten

Luftstrom-Planung

Planen Sie eine DAG über Python

Entschlüsselung von Luftstromplänen

Fehlersuche DAG läuft

Implementierung des Luftstroms DAGs

In diesem Kapitel erfährst du, wie du mit Airflow-Komponenten wie Sensoren und Executors bei der Überwachung und Fehlerbehebung von Airflow-Workflows Zeit sparen kannst.

Luftstromsensoren

Optionen

Sensoren

Beides

Betreiber

Sensoren vs. Bediener

Sensorische Deprivation

Luftstrom-Executors

Bestimmung des Vollstreckers

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 auf den Vollstrecker

Fehlersuche und Fehlerbehebung in Airflow

DAGist in der Tasche

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


Fehlt DAG

SLAs und Berichte in Airflow

Definieren einer SLA

Definieren einer Aufgabe SLA

Erstelle einen Bericht und sende ihn per E-Mail

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


Hinzufügen von Status-E-Mails

Pflege und Überwachung von Airflow-Workflows

Nimm alles zusammen. In diesem letzten Kapitel wendest du alles an, was du gelernt hast, um einen produktionsgerechten Workflow in Airflow aufzubauen.

Arbeiten mit Vorlagen

Einen BashOperator mit Vorlage erstellen

Vorlagen mit mehreren Argumenten

Mehr Vorlagen

Listen mit Vorlagen verwenden

Verstehen der Parameteroptionen

Versenden von E-Mail-Vorlagen

Verzweigung

Define a BranchPythonOperator

Fehlersuche in der Branche

Erstellen einer Produktionspipeline

Erstellen einer Produktionspipeline #1

Erstellen einer Produktionspipeline #2

Hinzufügen der letzten Änderungen zu deiner Pipeline

Herzlichen Glückwunsch!

Aufbau von Produktionspipelines in Airflow

Jetzt aktualisiert auf Apache Airflow 2.7 - Die Bereitstellung von Daten nach einem Zeitplan kann ein manueller Prozess sein. Du schreibst Skripte, fügst komplexe Cron-Tasks hinzu und probierst verschiedene Möglichkeiten aus, um die sich ständig ändernden Anforderungen zu erfüllen - und es ist noch schwieriger, alles zu bewältigen, wenn du mit Teamkollegen zusammenarbeitest. Apache Airflow kann dir diese Kopfschmerzen ersparen, indem es deine Arbeitsabläufe um Zeitplanung, Fehlerbehandlung und Berichterstellung ergänzt. In diesem Kurs lernst du die Grundlagen von Apache Airflow kennen und erfährst, wie du komplexe Data-Engineering-Pipelines in der Produktion implementieren kannst. Außerdem lernst du, wie du gerichtete azyklische Graphen (DAGs) verwendest, Data-Engineering-Workflows automatisierst und Data-Engineering-Aufgaben auf einfache und wiederholbare Weise umsetzt - damit du deinen Verstand 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 in Python

Erfahren Sie, wie Sie Daten-Engineering-Workflows implementieren und planen.

Dateningenieur in Python

Sensorische Deprivation

Einführung in Apache Airflow in Python

Interaktive Übung zum Anfassen