Airflow-Operatoren

Einführung in Apache Airflow mit Python

Mike Metzger

Data Engineer

Operatoren

Repräsentiert eine einzelne Aufgabe im Workflow
Läuft meist unabhängig
Teilt i. d. R. keine Daten
Verschiedene Operatoren für verschiedene Aufgaben

@dag(
  dag_id="Example_Dag"
)
def example_dag():
  @task
  def task1():
    return "The result from task1"

  task1()

example_dag()

@task (PythonOperator)

Führt eine Python-Funktion aus
Beliebige Python-Funktion kann mit @task dekoriert werden
Kann Daten mit anderen Funktionen/Tasks austauschen

from airflow.sdk import task

@task
def printme():
    print("This goes in the logs!")

printme()

@task-Argumente

Argumente wie bei normalen Python-Funktionen übergeben

@task
def printme(name: str):
    print(f"Hi {name} - This goes in the logs!")


printme(name='DataCamp')


# Fügt Folgendes zu den Airflow-Logs hinzu:
#   Hi DataCamp - This goes in the logs!

@task.bash (BashOperator)

@task.bash
def bash_example():
  return "echo 'Example!'"

bash_example()

@task.bash
def run_cleanup():
  return "runcleanup.sh"

run_cleanup()

Führt einen Bash-Befehl oder ein Skript aus
Läuft im temporären Verzeichnis
Umgebungsvariablen können gesetzt werden

Task-Abhängigkeiten

Jede DAG hat Aufgaben (Tasks)
Abhängigkeiten legen die Reihenfolge fest
Mehrere Arten, Abhängigkeiten zu setzen

Airflow-DAG mit verbundenen Tasks, die ihre Abhängigkeitsreihenfolge zeigen

Bitshift-Syntax

>> und <<

task1() >> task2()

# task1 endet, bevor task2 startet


task1() >> task2() >> task3()

# task1 endet vor task2 und task2 endet vor task3


task1() >> task3()
task2() >> task3()

# task1 und task2 können parallel laufen, beide müssen vor task3 fertig sein

Beispiel: Bitshift-Syntax

# Verkaufsdaten laden vor dem Abgleich
download_sales_data() >> reconcile()

# Bestandsdaten laden vor dem Abgleich
download_inventory_data() >> reconcile()

Beide müssen vor dem Abgleich fertig sein; Reihenfolge egal

Lass uns üben!

Einführung in Apache Airflow mit Python