Mengimplementasikan fungsi callback
Baru-baru ini Anda ditugaskan untuk menambahkan failure callback pada Dag yang dibuat oleh tim Anda. Untuk memulai, Anda ingin menambahkan failure callback sederhana yang menuliskan sebuah pesan ke audit log ketika sales_etl_dag gagal.
Objek dag dan task sudah diimpor dan tugas get_sales_data serta process_sales_data telah dibuat.
Latihan ini merupakan bagian dari kursus
Pengantar Apache Airflow dengan Python
Instruksi latihan
- Buat fungsi callback bernama
alert_on_failure. - Definisikan fungsi agar menerima objek apa pun yang diteruskan Airflow kepadanya.
- Tentukan failure callback dengan menggunakan fungsi
alert_on_failure.
Latihan interaktif langsung praktik
Cobalah latihan ini dengan melengkapi kode contoh ini.
# 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()