Opérateurs Airflow

Introduction à Apache Airflow en Python

Mike Metzger

Data Engineer

Opérateurs

  • Représentent une tâche unique dans un workflow
  • S’exécutent de façon indépendante (généralement)
  • Partagent rarement des informations
  • Divers opérateurs pour différents types de tâches
@dag(
  dag_id="Example_Dag"
)
def example_dag():
  @task
  def task1():
    return "The result from task1"

  task1()

example_dag()
Introduction à Apache Airflow en Python

@task (PythonOperator)

  • Exécute une fonction Python
  • Toute fonction Python peut être décorée avec @task
  • Peut échanger des données avec d’autres fonctions/tâches
from airflow.sdk import task

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

printme()
Introduction à Apache Airflow en Python

Arguments @task

  • Peut passer des arguments à la tâche/fonction comme une fonction Python ordinaire
@task
def printme(name: str):
    print(f"Hi {name} - This goes in the logs!")


printme(name='DataCamp')


# Ajoute :
#   Hi DataCamp - This goes in the logs!
# aux logs Airflow
Introduction à Apache Airflow en Python

@task.bash (BashOperator)

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

bash_example()

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

run_cleanup()
  • Exécute une commande ou un script Bash donné
  • Lance la commande dans un répertoire temporaire
  • Peut définir des variables d’environnement pour la commande
Introduction à Apache Airflow en Python

Dépendances entre tâches

  • Chaque DAG a un ensemble de tâches à exécuter
  • Les dépendances définissent l’ordre d’exécution
  • Plusieurs méthodes pour définir les dépendances

 

DAG Airflow de tâches connectées montrant leur ordre de dépendance

Introduction à Apache Airflow en Python

Syntaxe bitshift

>> et <<

task1() >> task2()

# task1 se termine avant le démarrage de task2


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

# task1 se termine avant task2 et task2 avant task3


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

# task1 et task2 peuvent s’exécuter ensemble mais doivent toutes deux finir avant task3
Introduction à Apache Airflow en Python

Exemple de syntaxe bitshift

# Télécharger les ventes avant le rapprochement
download_sales_data() >> reconcile()

# Télécharger les stocks avant le rapprochement
download_inventory_data() >> reconcile()

 

  • Les deux doivent finir avant le rapprochement, sans ordre imposé
Introduction à Apache Airflow en Python

Passons à la pratique !

Introduction à Apache Airflow en Python

Preparing Video For Download...