KAL
Menu

Penguatan operasional

Memperkuat pipeline secara operasional: alert, prosedur backfill, dan dokumentasi.

Daftar isi

Pipeline yang sudah berjalan benar pada run pertama masih dapat gagal pada run keseratus, dan perbedaan antara keduanya ada pada penguatan operasionalnya. Topik ini memperkuat proyek statistik ujian pada tiga sisi, yaitu alert kegagalan yang benar-benar terpicu oleh task gagal, prosedur backfill yang menangani rentang dan run yang sudah ada, serta dokumentasi yang membuat DAG dapat dirawat oleh orang lain. Setiap sisi diuji dengan eksekusi nyata beserta keluarannya.

Alert yang terbukti terpicu#

Argumen on_failure_callback menerima fungsi Python yang dipanggil Airflow ketika task gagal, dengan satu argumen berupa kamus konteks eksekusi. Konteks itu memuat task_instance yang menyatakan DAG dan task yang gagal, serta kunci exception yang berisi penyebabnya. Pemasangan melalui default_args mencakup seluruh task DAG sekaligus, sehingga DAG baru tidak perlu mengulang pemasangannya per task.

DAG berikut memasang callback yang menuliskan baris alert ke berkas, sebuah penanda sederhana yang menggantikan kanal pemberitahuan sungguhan seperti email atau pesan ke kanal tim.

python dag_ekspor_ringkas.py 
"""DAG ekspor ringkas dengan alert kegagalan untuk latihan operasional.

Task kedua sengaja gagal sehingga on_failure_callback terpicu dan
menuliskan baris pemberitahuan ke berkas alert.
"""
import pendulum
from airflow.sdk import dag, task

BERKAS_ALERT = "/tmp/ra-operasional/alert.log"


def laporkan_gagal(context) -> None:
    """Menuliskan baris pemberitahuan kegagalan task ke berkas alert."""
    ti = context["task_instance"]
    with open(BERKAS_ALERT, "a", encoding="utf-8") as berkas:
        berkas.write(
            f"ALERT: DAG {ti.dag_id}, task {ti.task_id} gagal pada {context.get('exception')}\n"
        )


@dag(
    dag_id="dag_ekspor_ringkas",
    schedule=None,
    start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
    default_args={"on_failure_callback": laporkan_gagal},
    description="Ekspor ringkas dengan alert kegagalan pada berkas.",
)
def dag_ekspor_ringkas():
    @task
    def hitung_total() -> int:
        """Menghitung total nilai kelas."""
        return 82 + 74 + 91

    @task
    def ekspor_total(total: int) -> None:
        """Menuliskan total; sengaja membagi dengan nol pada latihan ini."""
        rata = total / 0
        print(f"rata-rata: {rata}")

    ekspor_total(hitung_total())


dag_ekspor_ringkas()

DAG disalin ke folder dags, basis data metadata disiapkan, lalu dag_ekspor_ringkas dijalankan sekali dengan task yang sengaja membagi dengan nol.

bash 
# lingkungan latihan terpisah untuk topik ini
export AIRFLOW_HOME=/tmp/ra-operasional
export AIRFLOW__CORE__LOAD_EXAMPLES=False
# lingkungan latihan bersih agar urutan langkah dapat diulang dari awal
rm -rf "$AIRFLOW_HOME"
bash 
# salin kedua berkas DAG topik ini ke folder yang dipindai Airflow
export AIRFLOW_HOME=/tmp/ra-operasional
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_ekspor_ringkas.py dag_rekap_harian.py "$AIRFLOW_HOME/dags/"
bash 
# siapkan skema basis data metadata
# log migrasi berisi stempel waktu, sehingga tidak ditampilkan
export AIRFLOW_HOME=/tmp/ra-operasional
airflow db migrate > /dev/null 2>&1
bash 
# jalankan DAG yang task terakhirnya sengaja gagal
# perintah diakhiri echo karena task gagal membuatnya berakhir gagal
export AIRFLOW_HOME=/tmp/ra-operasional
airflow dags test dag_ekspor_ringkas 2026-11-04 > /tmp/ra-operasional/alert-test.log 2>&1 || echo "dags test berakhir dengan task gagal"
text Keluaran 
dags test berakhir dengan task gagal

Berkas alert kini berisi satu baris yang menyebut DAG, task, dan penyebab kegagalannya, lengkap dengan nama exception yang sama seperti pada topik debugging.

bash 
# tampilkan isi berkas alert buatan callback
cat /tmp/ra-operasional/alert.log
text Keluaran 
ALERT: DAG dag_ekspor_ringkas, task ekspor_total gagal pada division by zero

Callback ini menunjukkan titik sambung yang dibutuhkan sistem produksi. Fungsi laporkan_gagal dapat digantikan isinya dengan pemanggilan webhook atau pengiriman pesan ke kanal tim tanpa mengubah satu pun task, karena semua task mewarisinya dari default_args.

Prosedur backfill yang terkendali#

Backfill mengisi run untuk rentang data interval lama, dan prosedurnya berjalan aman apabila tiga syarat dipenuhi. Pertama, task yang di-backfill harus idempoten, karena run lama dijalankan ulang dan hasilnya menimpa partisi yang sama. Kedua, rentangnya harus diperiksa terhadap run yang sudah ada, karena backfill create secara bawaan tidak membuat ulang run untuk tanggal yang sudah memiliki run. Ketiga, hanya satu pekerjaan backfill boleh aktif untuk satu DAG, sehingga pekerjaan yang masih berjalan tidak boleh ditumpuk.

DAG dag_rekap_harian menjadi bahan latihannya, yaitu DAG berjadwal cron 0 6 * * * yang menuliskan berkas rekap terpartisi per tanggal.

python dag_rekap_harian.py 
"""DAG rekap harian proyek statistik ujian sekolah.

Menuliskan rekap jumlah siswa dan rata-rata nilai per data interval
ke berkas terpartisi, sehingga setiap interval menghasilkan berkas
yang sama ketika dijalankan ulang.
"""
import csv
import os

import pendulum
from airflow.sdk import dag, task

FOLDER_REKAP = "/tmp/ra-operasional/rekap"

NILAI = [82, 74, 91, 66, 78, 85]


@dag(
    dag_id="dag_rekap_harian",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 2},
    description="Rekap jumlah siswa dan rata-rata nilai per data interval.",
    doc_md="""Rekap harian proyek statistik ujian.

DAG ini menuliskan satu berkas rekap untuk setiap data interval,
berisi jumlah siswa dan rata-rata nilai. Berkasnya terpartisi per
tanggal, sehingga run ulang untuk tanggal yang sama aman dilakukan.
""",
)
def dag_rekap_harian():
    @task
    def tulis_rekap(ds_nodash: str) -> str:
        """Menuliskan berkas rekap partisi tanggal data interval."""
        folder = f"{FOLDER_REKAP}/{ds_nodash}"
        os.makedirs(folder, exist_ok=True)
        path = f"{folder}/rekap.csv"
        with open(path, "w", newline="", encoding="utf-8") as berkas:
            penulis = csv.writer(berkas)
            penulis.writerow(["jumlah_siswa", "rata_nilai"])
            penulis.writerow([len(NILAI), round(sum(NILAI) / len(NILAI), 1)])
        print(f"rekap ditulis ke {path}")
        return path

    tulis_rekap()


dag_rekap_harian()

Backfill diminta untuk rentang 1 sampai 3 November 2026. Perhatikan batas atas rentangnya. Run dengan jadwal cron pukul 06.00 milik 4 November masih berada di luar rentang karena --to-date berhenti pada awal hari tersebut.

bash 
# buat backfill untuk rentang tiga data interval
# log panjangnya diarahkan ke berkas
export AIRFLOW_HOME=/tmp/ra-operasional
airflow dags reserialize > /dev/null 2>&1
airflow backfill create --dag-id dag_rekap_harian --from-date 2026-11-01 --to-date 2026-11-04 > /tmp/ra-operasional/backfill.log 2>&1
bash 
# hitung run yang dibuat pekerjaan backfill
grep -c "Created backfill Dag run" /tmp/ra-operasional/backfill.log
text Keluaran 
3

Daftar run memperlihatkan tiga run backfill yang menunggu dieksekusi scheduler, dengan run_id yang berasal dari tanggal data intervalnya.

bash 
# tampilkan daftar dag run milik DAG rekap harian
export AIRFLOW_HOME=/tmp/ra-operasional
airflow dags list-runs dag_rekap_harian 2>&1 | grep -v alembic | head -n 7
text Keluaran 
dag_id           | run_id                              | state  | run_after                 | logical_date              | start_date | end_date
=================+=====================================+========+===========================+===========================+============+=========
dag_rekap_harian | backfill__2026-11-03T06:00:00+00:00 | queued | 2026-11-03T06:00:00+00:00 | 2026-11-03T06:00:00+00:00 |            |
dag_rekap_harian | backfill__2026-11-02T06:00:00+00:00 | queued | 2026-11-02T06:00:00+00:00 | 2026-11-02T06:00:00+00:00 |            |
dag_rekap_harian | backfill__2026-11-01T06:00:00+00:00 | queued | 2026-11-01T06:00:00+00:00 | 2026-11-01T06:00:00+00:00 |            |

Status queued menyatakan run sudah tercatat pada basis data metadata dan menunggu scheduler, karena eksekusi pekerjaan backfill adalah tugas scheduler, bukan tugas perintah yang membuatnya. Pada instalasi produksi dengan scheduler aktif, ketiga run dieksekusi berurutan dan berkas rekap ketiga partisinya terbentuk. Untuk memverifikasi perilaku task tanpa menunggu scheduler, satu tanggal dijalankan langsung dengan dags test, dan berkas rekap partisi 20261102 terbentuk dengan isinya.

bash 
# jalankan satu tanggal secara lokal untuk memverifikasi hasil task
export AIRFLOW_HOME=/tmp/ra-operasional
airflow dags test dag_rekap_harian 2026-11-02 > /tmp/ra-operasional/rekap-test.log 2>&1
cat /tmp/ra-operasional/rekap/20261102/rekap.csv
text Keluaran 
jumlah_siswa,rata_nilai
6,79.3

Dokumentasi DAG dan repo#

Dokumentasi DAG berada di dua tempat. Di dalam berkas DAG, argumen doc_md pada @dag dan @task menyimpan dokumen Markdown yang ditampilkan pada panel detail DAG di antarmuka web. README.md pada folder repo menjelaskan hal yang tidak tampak dari kode, misalnya daftar DAG beserta tanggung jawab dan kebijakan operasionalnya.

Argumen doc_md pada dag_rekap_harian sudah terpasang pada kode di atas, dan isinya terbaca langsung dari atribut DAG yang dimuat.

bash 
# tampilkan isi doc_md dari DAG yang dimuat
export AIRFLOW_HOME=/tmp/ra-operasional
python3 -c "import dag_rekap_harian; d = dag_rekap_harian.dag_rekap_harian(); print(d.doc_md)"
text Keluaran 
Rekap harian proyek statistik ujian.

DAG ini menuliskan satu berkas rekap untuk setiap data interval,
berisi jumlah siswa dan rata-rata nilai. Berkasnya terpartisi per
tanggal, sehingga run ulang untuk tanggal yang sama aman dilakukan.

README berikut merangkum repo dari sudut pandang pengelola, bukan dari sudut pandang kode, sehingga anggota tim baru mengetahui pembagian DAG dan kebijakan yang berlaku sebelum membaca satu pun berkas DAG.

markdown README.md 
# Repo DAG proyek statistik ujian sekolah

Folder `dags/` pada deployment ini memuat tiga DAG:

| DAG | Jadwal | Tanggung jawab |
|---|---|---|
| `dag_ingesta_harian` | `0 6 * * *` | Menyalin sumber CSV mentah ke folder ingesta terpartisi tanggal. |
| `dag_olah_nilai` | tanpa jadwal | Gerbang mutu lalu agregasi nilai dan kehadiran per kelas. |
| `dag_ringkasan_ekspor` | tanpa jadwal | Sensor ingesta, gerbang mutu, ringkasan lintas kelas, ekspor CSV. |

## Kebijakan operasional

- Setiap task menulis hasil ke jalur yang memuat `ds_nodash` dengan mode
  timpa, sehingga run ulang dan backfill aman dilakukan.
- Keberhasilan diperiksa lewat `dags test` sebelum deployment, dan kegagalan
  task memicu `on_failure_callback` yang menuliskan baris alert.
- Backfill dibuat dengan `airflow backfill create` setelah rentangnya
  diperiksa terhadap daftar run yang sudah ada.

Kesalahan umum#

  • Menumpuk pekerjaan backfill. backfill create kedua untuk DAG yang pekerjaannya masih aktif diakhiri AlreadyRunningBackfill, dan pesan errornya baru muncul setelah seluruh validasi selesai. Periksa daftar run terlebih dahulu, lalu gunakan rentang yang belum diisi atau hapus run yang masih queued.
  • Menetapkan batas --to-date tanpa menghitung jam jadwalnya. DAG dengan cron 0 6 * * * memiliki run pukul 06.00 setiap hari, sedangkan --to-date 2026-11-03 berhenti pada awal hari itu, sehingga run 3 November tidak termasuk rentang. Hitung rentang berdasarkan run_after, bukan hanya tanggalnya.
  • Menuliskan alert ke berkas yang namanya tetap. Berkas alert yang sama memuat kegagalan lintas run tanpa penanda tanggal, sehingga sulit ditelusuri. Sertakan konteks eksekusi pada isi pesannya sebagaimana task_instance pada callback, atau partisikan berkas penyimpanannya per tanggal.

Ringkasan#

  • on_failure_callback yang dipasang melalui default_args mencakup seluruh task DAG, dan callback yang dieksekusi pada task gagal menuliskan baris alert berisi DAG, task, serta penyebabnya.
  • Prosedur backfill memerlukan task idempoten, rentang yang dihitung dari run_after, dan satu pekerjaan aktif per DAG, sedangkan run hasilnya menunggu scheduler dalam status queued.
  • doc_md pada DAG ditampilkan antarmuka web dan terbaca dari atribut DAG yang dimuat, sedangkan README.md menjelaskan repo dari sudut pengelola.

Referensi#