Airflow-operatorer

Introduktion till Apache Airflow i Python

Mike Metzger

Data Engineer

Operatorer

Representerar en enskild uppgift i ett arbetsflöde
Körs oberoende av varandra (vanligtvis)
Delar i allmänhet inte information
Olika operatorer utför olika uppgifter

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

  task1()

example_dag()

@task (PythonOperator)

Kör en Python-funktion
Valfri Python-funktion kan dekoreras med @task
Kan skicka data till och från andra funktioner/uppgifter

from airflow.sdk import task

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

printme()

@task-argument

Argument kan skickas till uppgiften/funktionen precis som till en vanlig Python-funktion

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


printme(name='DataCamp')


# Adds:
#   Hi DataCamp - This goes in the logs!
# to the 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()

Kör ett givet Bash-kommando eller skript
Kör kommandot i en tillfällig katalog
Miljövariabler kan anges för kommandot

Uppgiftsberoenden

Varje DAG har en uppsättning uppgifter att slutföra
Uppgiftsberoenden styr körningsordningen
Det finns olika sätt att ange beroenden

Airflow-DAG med sammankopplade uppgifter som visar beroendeordningen

Bitshift-syntax

>> och <<

task1() >> task2()

# task1 completes before task2 starts


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

# task1 completes before task2 and task2 completes before task3


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

# task1 and task2 can run together but both must complete before task3 runs

Exempel på bitshift-syntax

# Download sales data before reconciling
download_sales_data() >> reconcile()

# Download inventory data before reconciling
download_inventory_data() >> reconcile()

Båda måste slutföras innan avstämning, men inbördes ordning är inte specificerad

Nu kör vi en övning!

Introduktion till Apache Airflow i Python