KAL
Menu

Percobaan ulang dan peringatan

Mengatur percobaan ulang pada task idempoten dan notifikasi kegagalan.

Daftar isi

Topik ini membahas percobaan ulang (retry) pada task yang gagal sesaat dan peringatan yang menyala ketika kegagalan itu menjadi permanen. Pembahasan mencakup argumen retries dan retry_delay, syarat idempotensi agar percobaan ulang aman, satu simulasi percobaan ulang dengan Python murni, lalu pemasangan peringatan lewat callback.

Mengatur percobaan ulang pada task#

Airflow mencoba ulang task yang gagal secara otomatis selama kuotanya masih ada, dan task yang menunggu percobaan berikutnya berstatus up_for_retry. Argumen retries menetapkan banyak percobaan ulang, sedangkan retry_delay menetapkan jeda antar percobaan dalam bentuk timedelta. Nilai retries=2 berarti task dijalankan hingga tiga kali, yaitu satu eksekusi pertama ditambah dua percobaan ulang.

python Contoh: task dengan kuota percobaan ulang 
# tidak-dijalankan
from datetime import timedelta

from airflow.sdk import task


@task(retries=2, retry_delay=timedelta(minutes=5))
def panggil_api_sumber() -> list[str]:
    """Mengambil data dari API sumber yang kadang tidak stabil."""
    respons = permintaan_ke_api()
    return respons["data"]

Kedua argumen itu dapat diberikan pada setiap task, atau sekali saja lewat default_args pada level DAG agar seluruh task mewarisinya. Airflow juga menyediakan on_retry_callback, yakni callback yang dipanggil setiap kali task memasuki percobaan ulang, sehingga percobaan yang sedang berlangsung pun dapat dicatat.

Kuota percobaan ulang menangani kegagalan yang muncul langsung, sedangkan batas waktu menangani task yang menggantung tanpa pernah selesai. Argumen execution_timeout menetapkan durasi maksimum satu eksekusi, dan task yang melewatinya dihentikan dengan exception bawaan Airflow lalu masuk ke alur percobaan ulang seperti kegagalan biasa. Keduanya sebaiknya dipasang bersama, karena sumber luar yang lambat sering berwujud sebagai penggantungan, bukan sebagai error yang muncul seketika.

Idempotensi agar percobaan ulang aman#

Percobaan ulang hanya aman pada task yang idempoten, yakni aman dijalankan berulang dengan hasil akhir yang sama. Task yang menambahkan baris tanpa pembersihan akan menggandakan data ketika dijalankan ulang, sedangkan task yang menimpa hasil lamanya tidak menambah apa pun. Karena itu, keluaran task sebaiknya ditulis ke jalur yang dihitung dari tanggal logis Dag run, misalnya berkas rekap per tanggal, sehingga eksekusi ulang hanya menimpa hasil tanggal yang sama.

Pola serupa berlaku pada tabel tujuan. Sebelum menulis hasil hari tertentu, hapus terlebih dahulu baris milik tanggal itu, lalu tulis ulang keseluruhannya. Berkas rekap pun dapat dinamai menyertakan tanggal logisnya, sehingga percobaan ulang pada hari yang sama selalu menimpa berkas yang sama dan tidak pernah menyentuh berkas hari lain. Dengan kebiasaan tersebut, percobaan ulang, pengisian ulang data lama, dan eksekusi manual berjalan tanpa risiko data ganda.

Meniru percobaan ulang dalam kode murni#

Perilaku percobaan ulang dapat disimulasikan tanpa Airflow agar mekanismenya terlihat. Skrip berikut menjalankan fungsi yang gagal pada dua percobaan pertama dan berhasil pada percobaan ketiga, dengan penghitung yang pasti alih-alih angka acak.

python percobaan-ulang.py 
"""Meniru percobaan ulang Airflow: fungsi gagal dua kali lalu berhasil."""

def panggil_sumber(nomor: int) -> str:
    """Menggantikan panggilan ke sumber data pada percobaan ke nomor."""
    if nomor < 3:
        raise ConnectionError("sumber belum siap")
    return "data diterima"


def jalankan_dengan_percobaan_ulang(maksimal: int) -> str:
    """Menjalankan panggilan_sumber hingga maksimal percobaan, meniru retries."""
    for nomor in range(1, maksimal + 1):
        try:
            hasil = panggil_sumber(nomor)
        except ConnectionError as pesan:
            print(f"percobaan {nomor}: gagal ({pesan})")
            if nomor < maksimal:
                print("  menunggu lalu mencoba ulang")
        else:
            print(f"percobaan {nomor}: berhasil, {hasil}")
            return hasil
    raise RuntimeError("task gagal permanen setelah semua percobaan habis")


jalankan_dengan_percobaan_ulang(3)
text Keluaran 
percobaan 1: gagal (sumber belum siap)
  menunggu lalu mencoba ulang
percobaan 2: gagal (sumber belum siap)
  menunggu lalu mencoba ulang
percobaan 3: berhasil, data diterima

Keluaran memperlihatkan pola yang sama dengan retries=2 pada Airflow, yaitu dua kegagalan yang diikuti jeda, lalu keberhasilan pada eksekusi ketiga. Apabila percobaan ketiga juga gagal, skrip melempar RuntimeError sebagaimana Airflow menetapkan status failed permanen setelah kuota habis. Status permanen itulah pemicu peringatan pada bagian berikutnya.

Struktur skripnya meniru dua sisi Airflow sekaligus, yaitu fungsi task yang bisa gagal dan lapisan penjadwal yang mengulanginya. Fungsi panggil_sumber mewakili task, sedangkan jalankan_dengan_percobaan_ulang mewakili pengaturan retries beserta jeda di antara percobaan. Pemisahan semacam ini juga berguna saat menguji DAG, karena perilaku task dapat diperiksa tanpa menyalakan penjadwal.

Peringatan kegagalan dengan callback#

Peringatan dipasang lewat callback, yakni fungsi yang dipanggil Airflow pada kejadian tertentu. Argumen on_failure_callback menerima fungsi tersebut pada level task atau level DAG, dan setiap callback menerima kamus konteks yang memuat informasi task instance, misalnya task_instance_key_str untuk mengenali task yang gagal. Callback dapat diberikan satu fungsi atau daftar fungsi sekaligus.

python Contoh: callback peringatan pada task 
# tidak-dijalankan
from airflow.sdk import task


def peringatan_gagal(konteks) -> None:
    """Dipanggil Airflow ketika task gagal, membawa informasi task instance."""
    print(f"task gagal: {konteks['task_instance_key_str']}")
    # di sini peringatan dikirim, misalnya ke webhook atau layanan pesan


@task(on_failure_callback=peringatan_gagal)
def tulis_rekap(total: int) -> None:
    """Menuliskan hasil rekap ke tabel tujuan."""
    tulis_ke_tabel_tujuan(total)

Isi callback biasanya berupa pengiriman pemberitahuan, misalnya pesan ke kanal tim atau webhook, yakni URL yang menerima pemberitahuan otomatis. Jaga callback tetap sederhana, karena pekerjaan berat di dalamnya menambah jalur kegagalan baru pada saat sistem sedang bermasalah.

Peringatan pada level DAG melengkapi peringatan per task. Argumen on_failure_callback yang diberikan pada DAG dipanggil ketika Dag run gagal, dan konteksnya memuat run_id yang mengenali eksekusi yang bermasalah. Peringatan per task mempercepat penemuan penyebabnya, sedangkan peringatan per DAG memastikan kegagalan yang melibatkan banyak task tetap dilaporkan sebagai satu kejadian.

Kesalahan umum#

  1. Mengandalkan percobaan ulang pada task yang tidak idempoten. Penyebabnya, keluaran ditambahkan tanpa pembersihan, sehingga setiap percobaan ulang menggandakan baris. Perbaikannya, tulis hasil ke partisi tanggal logis dengan pola ganti-tulis sebelum kuota percobaan diperbesar.
  2. Memilih retry_delay yang terlalu kecil. Penyebabnya, jeda detik membuat seluruh kuota percobaan habis sebelum sumber luar selesai pulih. Perbaikannya, sesuaikan jeda dengan pola pemulihan sumber, misalnya beberapa menit untuk API yang mengalami gangguan singkat.
  3. Menulis callback yang bisa gagal sendiri. Penyebabnya, pengiriman peringatan memakai kode yang tidak dilindungi penanganan error, sehingga peringatan tidak terkirim dan bekasnya hanya ada di log dag processor. Perbaikannya, bungkus pengiriman peringatan dengan penanganan error dan jaga isinya tetap sederhana.

Ringkasan#

  • Argumen retries dan retry_delay mengatur percobaan ulang otomatis, dan task yang menunggu percobaan berikutnya berstatus up_for_retry.
  • Percobaan ulang hanya aman pada task idempoten yang menimpa hasil milik rentang data yang sama, bukan menambah baris baru.
  • on_failure_callback pada level task atau DAG mengirim peringatan saat kegagalan menjadi permanen, dengan kamus konteks sebagai informasinya.
  • Error di dalam callback tampil pada log dag processor, bukan pada log task yang gagal.

Referensi#