KAL
Menu

Task idempoten

Merancang task yang aman dijalankan ulang: operasi idempoten dan penulisan berkas terpartisi.

Daftar isi

Task yang aman dijalankan ulang adalah syarat dasar pipeline data yang dapat dipercaya. Topik ini membahas sifat idempoten (idempotent), yaitu sifat yang membuat efek setelah n kali eksekusi sama dengan efek satu kali eksekusi. Pola penerapannya dibahas untuk penulisan berkas, penulisan basis data, dan efek samping di luar penyimpanan data. Pemahaman ini menjadi fondasi backfill yang aman karena backfill pada dasarnya adalah menjalankan ulang run untuk interval tertentu.

Definisi idempotensi#

Sebuah operasi disebut idempoten apabila menjalankannya beberapa kali menghasilkan keadaan akhir yang sama dengan menjalankannya sekali. Definisi ini menyangkut efek pada keadaan di luar kode, misalnya isi berkas atau isi tabel, dan tidak menyangkut nilai kembalian operasi. Perbedaan operasi idempoten dan non-idempoten mudah dilihat pada operasi aritmetika sederhana. Penjumlahan pada nilai yang sama tidak idempoten karena hasilnya bertambah setiap kali diulang, sedangkan penimpaan nilai bersifat idempoten karena keadaan akhirnya ditentukan sepenuhnya oleh operasi terakhir.

Skrip berikut membandingkan keduanya pada daftar saldo. Fungsi tambah_saldo menambah nilai yang sama setiap kali dipanggil, sedangkan fungsi tetapkan_saldo menulis nilai yang sama setiap kali dipanggil. Keduanya dipanggil dua kali berurutan dengan nilai awal yang sama.

python 
def tambah_saldo(saldo, kenaikan):
    for i in range(len(saldo)):
        saldo[i] += kenaikan
    return saldo


def tetapkan_saldo(saldo, nilai):
    for i in range(len(saldo)):
        saldo[i] = nilai
    return saldo


# pola non-idempoten: efek bertambah pada setiap eksekusi
saldo = [100, 200]
pertama = list(tambah_saldo(saldo, 10))
kedua = list(tambah_saldo(saldo, 10))
print("non-idempoten:", pertama, "lalu", kedua)

# pola idempoten: efek tetap pada setiap eksekusi
saldo = [100, 200]
pertama = list(tetapkan_saldo(saldo, 110))
kedua = list(tetapkan_saldo(saldo, 110))
print("idempoten    :", pertama, "lalu", kedua)
text Keluaran 
non-idempoten: [110, 210] lalu [120, 220]
idempoten    : [110, 110] lalu [110, 110]

Keluaran memperlihatkan perbedaan kedua pola. Dua kali penambahan menghasilkan nilai 120 dan 220, sedangkan dua kali penimpaan tetap memberikan 110. Task Airflow berada pada sisi penimpaan karena scheduler dapat menjalankan task yang sama lebih dari sekali, baik karena retry, backfill, maupun pemicuan manual, tanpa pemberitahuan apa pun. Task yang efeknya menumpuk pada setiap eksekusi menghasilkan data yang berbeda setiap kali dijalankan ulang.

Berkas hasil terpartisi per data interval#

Task yang menulis keluaran berkas menjadi idempoten melalui dua kebiasaan yang saling melengkapi. Pertama, hasil ditulis dengan mode timpa, bukan mode tambah. Kedua, nama berkas menyertakan data interval run. Nama berkas yang mengandung data interval membuat run untuk interval berbeda menulis ke berkas yang berbeda, sehingga tidak ada run yang menimpa hasil run lain. Run untuk interval yang sama, termasuk run ulang, menimpa berkas yang sama dengan isi yang sama.

Berkas dag_nilai.py berikut menerapkan kedua kebiasaan itu. Task tulis_hasil membaca nilai ds dari konteks, yaitu tanggal data interval run dalam format tahun-bulan-tanggal, lalu menulis hasil agregasi ke berkas yang namanya menyertakan nilai tersebut. Data mentahnya disimulasikan sebagai daftar tetap pada berkas DAG, yaitu lima nilai ujian. Pemanggilan get_current_context mengembalikan kamus variabel konteks task, dan nilai ds yang sama juga dapat ditulis sebagai {{ ds }} pada field operator yang mendukung templat.

python dag_nilai.py 
import csv
from pathlib import Path

import pendulum
from airflow.sdk import dag, task, get_current_context

FOLDER_HASIL = Path("/tmp/ra-task-idempoten")
NILAI_UJIAN = [72, 85, 90, 65, 78]


@dag(
    dag_id="dag_nilai",
    schedule="@daily",
    start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
    description="Agregasi nilai ujian ke berkas CSV terpartisi tanggal data.",
)
def dag_nilai():
    @task
    def tulis_hasil():
        context = get_current_context()
        ds = context["ds"]
        FOLDER_HASIL.mkdir(parents=True, exist_ok=True)
        path = FOLDER_HASIL / f"hasil-{ds}.csv"
        with open(path, "w", newline="") as berkas:
            penulis = csv.writer(berkas)
            penulis.writerow(["jumlah_siswa", "nilai_rata_rata"])
            penulis.writerow([len(NILAI_UJIAN), round(sum(NILAI_UJIAN) / len(NILAI_UJIAN), 2)])
        print(f"Hasil ditulis ke {path}")

    tulis_hasil()


dag_nilai()

Perintah berikut mengarahkan basis data metadata Airflow ke folder sementara, menyiapkan folder dags, lalu menyalin berkas DAG ke sana. Variabel AIRFLOW_HOME ditetapkan pada setiap blok yang memanggil CLI agar blok dapat dijalankan ulang secara terpisah.

bash 
# siapkan folder dags lalu salin berkas DAG ke sana
export AIRFLOW_HOME=/tmp/ra-task-idempoten
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_nilai.py "$AIRFLOW_HOME/dags/"

Skema basis data metadata disiapkan terlebih dahulu melalui db migrate. Log migrasinya berisi stempel waktu sehingga tidak ditampilkan.

bash 
# siapkan skema basis data metadata pada folder yang sama
export AIRFLOW_HOME=/tmp/ra-task-idempoten
airflow db migrate > /dev/null 2>&1

Struktur DAG yang sudah dimuat diperiksa tanpa menjalankan task sama sekali. Pemeriksaan ini memastikan berkas DAG memuat satu DAG bernama dag_nilai dengan satu task sebelum DAG dijalankan.

bash 
# muat modul DAG lalu cetak dag_id, jadwal, dan seluruh task_id
export AIRFLOW_HOME=/tmp/ra-task-idempoten
python3 -c "import dag_nilai; d = dag_nilai.dag_nilai(); print(d.dag_id); print(d.schedule); print(sorted(t.task_id for t in d.tasks))"
text Keluaran 
dag_nilai
@daily
['tulis_hasil']

DAG dijalankan dua kali dengan tanggal pemicuan yang sama, yaitu 2026-11-02, untuk menunjukkan bahwa run ulang aman. Argumen tanggal menetapkan waktu pemicuan dag run, dan nilai ds pada konteks mengikuti tanggal tersebut. Log perintahnya panjang dan bertanda waktu, sehingga diarahkan ke berkas.

bash 
# run pertama untuk pemicuan 2026-11-02
export AIRFLOW_HOME=/tmp/ra-task-idempoten
airflow dags test dag_nilai 2026-11-02 > "$AIRFLOW_HOME/dags-test-1.log" 2>&1

Run kedua menggunakan tanggal pemicuan yang sama persis. Apabila task-nya idempoten, isi berkas hasil setelah run kedua harus sama dengan isi setelah run pertama.

bash 
# run kedua untuk pemicuan yang sama
export AIRFLOW_HOME=/tmp/ra-task-idempoten
airflow dags test dag_nilai 2026-11-02 > "$AIRFLOW_HOME/dags-test-2.log" 2>&1

Folder hasil diperiksa bersama isi berkasnya. Setelah dua eksekusi, hanya ada satu berkas hasil dan isinya tetap sama.

bash 
# tampilkan berkas hasil beserta isinya setelah dua kali eksekusi
ls /tmp/ra-task-idempoten/hasil-*
cat /tmp/ra-task-idempoten/hasil-2026-11-02.csv
text Keluaran 
/tmp/ra-task-idempoten/hasil-2026-11-02.csv
jumlah_siswa,nilai_rata_rata
5,78.0

Keluaran memenuhi definisi idempoten pada bagian pertama, yaitu dua kali eksekusi menghasilkan satu keadaan akhir yang sama. Run untuk pemicuan lain, misalnya 2026-11-03, akan menulis ke hasil-2026-11-03.csv dan tidak menyentuh berkas ini. Kebalikan pola ini adalah menambah baris ke berkas yang sama pada setiap eksekusi, kebiasaan yang sering dipilih karena terasa lebih murah, tetapi efeknya menumpuk pada setiap run.

Pola pada basis data#

Pada basis data, idempotensi berarti satu data interval menghasilkan satu keadaan tabel yang sama berapa kali pun task dijalankan. Praktik terbaik resmi Airflow menyarankan task diperlakukan seperti transaksi basis data, yaitu tidak pernah meninggalkan hasil yang setengah jadi. Pernyataan INSERT yang dijalankan ulang menambah baris duplikat, sehingga panduan tersebut menganjurkan penggunaan upsert, yaitu operasi yang memperbarui baris yang sudah ada atau menyisipkan baris baru apabila belum ada.

Dua pola yang paling sering digunakan sama-sama bekerja per partisi. Pola pertama adalah DELETE lalu INSERT, yaitu menghapus baris milik satu data interval sebelum menyisipkan hasil baru untuk interval itu. Misalnya, pernyataan DELETE FROM rekap_harian WHERE tanggal = '{{ ds }}' diikuti penyisipan hasil untuk tanggal yang sama. Pola kedua adalah pernyataan MERGE atau INSERT ... ON CONFLICT DO UPDATE yang menyatukan penyisipan dan pembaruan dalam satu pernyataan. Kedua pola membatasi cakupan kerjanya pada nilai kunci partisi, sehingga run untuk interval lain tidak tersentuh.

Pembatasan cakupan per partisi inilah yang membuat backfill aman. Run backfill untuk interval lama hanya menulis pada partisi intervalnya, sedangkan run harian menulis pada partisi intervalnya sendiri. Tanpa batas partisi yang eksplisit, run ulang dapat menyentuh data di luar cakupannya dan merusak hasil interval lain.

Efek samping global#

Idempotensi juga berlaku pada efek yang tidak tersimpan di berkas maupun tabel, misalnya mengirim email, mengirim pesan ke kanal chat, atau memanggil API eksternal. Eksekusi berikutnya tidak dapat menarik kembali efek tersebut. Apabila task yang mengirim email dijalankan ulang, email yang sama dikirim lagi, sehingga beberapa kali retry menghasilkan beberapa email untuk hasil perhitungan yang sama.

Ada dua pendekatan untuk mengurangi efek samping semacam ini. Pendekatan pertama adalah memindahkan efek samping ke task terpisah yang berada paling hilir, sehingga task yang berpeluang dijalankan ulang, misalnya task perhitungan dengan jumlah retry besar, tidak ikut membawa efek tersebut. Pendekatan kedua adalah menjadikan efek samping itu sendiri idempoten dengan kunci yang mengandung data interval, misalnya subjek email yang menyertakan tanggal data, sehingga duplikatnya mudah dikenali oleh penerima.

Kesalahan umum#

Dua kekeliruan berikut muncul paling sering ketika pola di atas dipraktikkan.

  • Nama berkas keluaran menggunakan waktu eksekusi nyata melalui datetime.now(). Path keluaran berubah pada setiap eksekusi, sehingga run ulang menulis ke berkas lain, sedangkan berkas lama beserta isinya tetap tertinggal. Gunakan nilai konteks seperti ds sebagai nama partisi, karena nilainya ditetapkan oleh data interval run dan tidak berubah antar eksekusi.
  • Berkas keluaran dibiarkan setengah jadi ketika task gagal di tengah penulisan. Eksekusi berikutnya menimpanya, tetapi sistem hilir yang membaca berkas di antara dua eksekusi dapat memproses isi yang belum lengkap. Tulis hasilnya ke berkas sementara terlebih dahulu, lalu ganti namanya menjadi nama akhir melalui satu operasi rename pada akhir task.

Ringkasan#

  • Task idempoten menghasilkan keadaan akhir yang sama setelah dijalankan n kali maupun satu kali, dan sifat ini menjadi syarat retry maupun backfill yang aman.
  • Keluaran berkas aman dijalankan ulang apabila ditulis dengan mode timpa dan namanya menyertakan data interval run, sehingga run berbeda menulis ke partisi berbeda dan run ulang menimpa isi yang sama.
  • Pada basis data, cakupan kerja dibatasi per partisi melalui DELETE lalu INSERT atau melalui upsert seperti MERGE, menggantikan INSERT yang menambah baris duplikat ketika dijalankan ulang.
  • Efek samping di luar penyimpanan data dipindahkan ke task paling hilir atau diberi kunci yang mengandung data interval agar duplikatnya dikenali.

Referensi#