Operatori Airflow

Introduzione ad Apache Airflow in Python

Mike Metzger

Data Engineer

Operatori

Rappresentano un singolo task in un workflow
Girano in modo indipendente (di solito)
In genere non condividono informazioni
Vari operator per svolgere task diversi

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

  task1()

example_dag()

@task (PythonOperator)

Esegue una funzione Python
Qualsiasi funzione Python può essere decorata con @task
Puoi passare dati ad altre funzioni/task

from airflow.sdk import task

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

printme()

Argomenti di @task

Puoi passare argomenti al task/funzione come a una normale funzione Python

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


printme(name='DataCamp')


# Aggiunge:
#   Hi DataCamp - This goes in the logs!
# ai log di Airflow

@task.bash (BashOperator)

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

bash_example()

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

run_cleanup()

Esegue un comando o script Bash dato
Esegue il comando in una cartella temporanea
Puoi specificare variabili d'ambiente per il comando

Dipendenze tra task

Ogni DAG ha un set di task da completare
Le dipendenze dei task definiscono l'ordine di esecuzione
Ci sono vari modi per definire le dipendenze

DAG di Airflow con task collegati che mostrano l'ordine delle dipendenze

Sintassi bitshift

>> e <<

task1() >> task2()

# task1 finisce prima che inizi task2


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

# task1 finisce prima di task2 e task2 prima di task3


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

# task1 e task2 possono girare insieme ma entrambi devono finire prima di task3

Esempio di sintassi bitshift

# Scarica i dati di vendita prima del riconcilio
download_sales_data() >> reconcile()

# Scarica i dati di inventario prima del riconcilio
download_inventory_data() >> reconcile()

Entrambi devono finire prima del riconcilio, ma l'ordine non è specificato

Passons à la pratique !

Introduzione ad Apache Airflow in Python