Airflow-operators

Introductie tot Apache Airflow in Python

Mike Metzger

Data Engineer

Operators

Stelt één taak in een workflow voor
Draait (meestal) onafhankelijk
Deelt doorgaans geen informatie
Diverse operators voor verschillende taken

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

  task1()

example_dag()

@task (PythonOperator)

Voert een Python-functie uit
Elke Python-functie kan met @task worden gedecoreerd
Kan data uitwisselen met andere functies/taken

from airflow.sdk import task

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

printme()

@task-argumenten

Je kunt argumenten aan de taak/functie meegeven zoals bij een gewone Python-functie

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


printme(name='DataCamp')


# Voegt toe:
#   Hi DataCamp - This goes in the logs!
# aan de Airflow-logs

@task.bash (BashOperator)

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

bash_example()

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

run_cleanup()

Voert een opgegeven Bash-opdracht of -script uit
Draait de opdracht in een tijdelijke map
Kan omgevingsvariabelen voor de opdracht meegeven

Taakafhankelijkheden

Elke Dag heeft een set taken om te voltooien
Taakafhankelijkheden bepalen de volgorde
Meerdere manieren om afhankelijkheden te definiëren

Airflow-Dag met verbonden taken die de afhankelijkheidsvolgorde tonen

Bitshift-syntaxis

>> en <<

task1() >> task2()

# task1 is klaar voordat task2 start


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

# task1 is klaar voor task2 en task2 is klaar voor task3


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

# task1 en task2 kunnen tegelijk draaien, maar beide moeten klaar zijn voor task3 draait

Voorbeeld bitshift-syntaxis

# Verkoopdata downloaden voor reconciliatie
download_sales_data() >> reconcile()

# Voorraaddata downloaden voor reconciliatie
download_inventory_data() >> reconcile()

Beide moeten klaar zijn voor reconciliatie; volgorde is niet gespecificeerd

Laten we oefenen!

Introductie tot Apache Airflow in Python