Eine Callback-Funktion implementieren

du wurdest vor Kurzem damit beauftragt, Failure-Callbacks zu den von deinem Team erstellten Dags hinzuzufügen. Zum Einstieg möchtest du einen einfachen Failure-Callback hinzufügen, der beim Fehlschlagen des sales_etl_dag eine Nachricht ins Prüfprotokoll schreibt.

Die Objekte dag und task sind bereits importiert und die Tasks get_sales_data und process_sales_data wurden erstellt.

Diese Übung ist Teil des Kurses

<Kurs>Einführung in Apache Airflow mit Python</Kurs>

Kurs ansehen

Übungsanweisungen

Erstelle eine Callback-Funktion namens alert_on_failure.
Definiere die Funktion so, dass sie beliebige Objekte akzeptiert, die Airflow ihr übergibt.
Lege einen Failure-Callback mithilfe der Funktion alert_on_failure fest.

Interaktive praktische Übung

Versuche dich an dieser Übung, indem du diesen Beispielcode vervollständigst.

# 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()

Code bearbeiten und ausführen