Implementera en callback-funktion
Du har nyligen fått i uppdrag att lägga till felåterrop (failure callbacks) i de DAG:ar som ditt team skapat. Som ett första steg vill du lägga till ett enkelt felåterrop som skriver ett meddelande till revisionsloggen när sales_etl_dag misslyckas.
dag- och task-objekten är redan importerade, och uppgifterna get_sales_data och process_sales_data har redan skapats.
Den här övningen är en del av kursen
Introduktion till Apache Airflow i Python
Övningsinstruktioner
- Skapa en callback-funktion med namnet
alert_on_failure. - Definiera funktionen så att den accepterar alla objekt som Airflow skickar till den.
- Ange ett felåterrop med hjälp av funktionen
alert_on_failure.
Interaktiv övning med praktiskt arbete
Testa den här övningen genom att slutföra den här exempelkoden.
# 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()