Implémenter une fonction de rappel (callback)

Vous devez récemment ajouter des callbacks d’échec aux Dags créés par votre équipe. Pour commencer, vous souhaitez ajouter un simple callback d’échec qui écrit un message dans le journal d’audit lorsque le sales_etl_dag échoue.

Les objets dag et task sont déjà importés et les tâches get_sales_data et process_sales_data ont été créées.

Cet exercice fait partie du cours

<cours>Introduction à Apache Airflow en Python</cours>

Voir le cours

Instructions de l’exercice

Créez une fonction de callback nommée alert_on_failure.
Définissez la fonction pour accepter tous les objets qu’Airflow lui transmet.
Indiquez un callback d’échec en utilisant la fonction alert_on_failure.

Exercice interactif pratique

Essayez cet exercice en complétant ce code d’exemple.

# Create the callback function
def ____(____):
  dag_id = context["dag"].dag_id
  task_id = context["task_instance"].task_id
  print(f"Task {task_id} in Dag {dag_id} has failed.")
  
# Specify the Dag with a failure callback
@dag(dag_id='sales_etl_dag',
     ____=alert_on_failure
)
def sales_etl_dag():
  get_sales_data() >> process_sales_data()
  
sales_etl_dag()

Modifier et exécuter le code