Pemrosesan inkremental

Memproses data bertahap dengan watermark, kursor, dan batas partisi per data interval.

Daftar isi

Pipeline yang memproses seluruh data dari awal pada setiap run cepat menjadi mahal seiring bertambahnya volume. Topik ini membahas pemrosesan inkremental, yaitu cara memproses data bertahap dengan watermark, kursor (cursor), dan batas partisi per data interval, beserta kaitan ketiganya dengan idempotensi task. Setelah membaca topik ini, batas query sebuah task dapat dibaca dari konteks dag run sehingga setiap run hanya menyentuh rentang datanya sendiri.

Full refresh dan pemrosesan inkremental#

Full refresh adalah strategi memproses ulang seluruh data sumber pada setiap run, sedangkan pemrosesan inkremental hanya memproses rentang data baru sejak batas terakhir yang sudah selesai. Pipeline harian dengan full refresh menghitung ulang seluruh riwayat meskipun hanya satu hari data yang bertambah. Biaya komputasinya tumbuh seiring panjang riwayat, sedangkan biaya pemrosesan inkremental relatif tetap karena cakupan kerjanya per interval.

Tabel berikut merangkum perbedaan keduanya.

Aspek	Full refresh	Inkremental
Cakupan per run	Seluruh data sumber	Hanya rentang data interval run
Biaya per run	Bertambah seiring panjang riwayat	Relatif tetap per interval
Batas yang diperlukan	Tidak ada	Watermark, kursor, atau data interval
Run ulang	Hasil akhirnya sama, tetapi biayanya diulang	Aman apabila batasnya eksplisit

Strategi inkremental menuntut kejelasan batas. Pipeline perlu mengetahui sampai titik mana data dianggap final dan dari titik mana pemrosesan dilanjutkan. Airflow menyediakan kedua batas tersebut melalui data interval yang menyertai setiap dag run, sehingga batas per run dapat dinyatakan oleh jadwal itu sendiri alih-alih oleh kode tambahan yang rawan keliru.

Watermark dan kursor#

Watermark adalah batas atas yang menyatakan bahwa data hingga titik tersebut dianggap final, yaitu tidak akan berubah lagi sehingga aman diproses. Data yang tanggalnya melampaui watermark belum dianggap lengkap karena masih mungkin ada keterlambatan masuk dari sumbernya. Pada pipeline terjadwal, watermark dapat dinyatakan eksplisit oleh data interval run, misalnya run dengan interval [2026-11-02, 2026-11-03) menyatakan bahwa hanya data pada rentang itu yang diproses.

Kursor adalah nilai yang menyatakan posisi terakhir pemrosesan, misalnya tanggal terakhir yang sudah direkap atau id terbesar yang sudah dimuat. Nilai ini disimpan terpisah dari kode, biasanya pada tabel kecil di basis data atau pada berkas state, lalu dibaca kembali oleh run berikutnya untuk melanjutkan pekerjaan. Kursor menentukan dari mana pemrosesan dilanjutkan, sedangkan watermark menentukan sampai mana data boleh diproses.

Keduanya bekerja berpasangan. Pemrosesan berjalan dari kursor hingga watermark, kemudian kursor dimajukan setelah pekerjaan selesai. Pada DAG terjadwal, Airflow sudah membawa pasangan batas serupa pada setiap dag run, yaitu data_interval_start dan data_interval_end, sehingga pola per hari tidak memerlukan pengelolaan state tambahan.

Data interval sebagai batas query#

Batas query yang eksplisit membuat setiap run hanya menyentuh rentangnya sendiri. Batas tersebut ditulis dalam format tanggal ISO, yaitu YYYY-MM-DD, karena perbandingan string pada format itu mengikuti urutan kalender. Tanggal "2026-11-02" bernilai lebih kecil daripada "2026-11-03" apabila dibandingkan sebagai string, sehingga notasi rentang [mulai, selesai) dapat diterjemahkan langsung menjadi perbandingan.

Skrip berikut memilih baris transaksi yang berada pada satu data interval dari kumpulan data kecil yang tersimpan sebagai daftar kamus.

python

transaksi = [
    {"id": 1, "tanggal": "2026-11-01", "jumlah": 20000},
    {"id": 2, "tanggal": "2026-11-02", "jumlah": 15000},
    {"id": 3, "tanggal": "2026-11-02", "jumlah": 30000},
    {"id": 4, "tanggal": "2026-11-03", "jumlah": 12000},
    {"id": 5, "tanggal": "2026-11-04", "jumlah": 25000},
]

# batas partisi yang diminta satu dag run
mulai = "2026-11-02"
selesai = "2026-11-03"

# hanya baris dengan tanggal di dalam [mulai, selesai) yang diproses
terpilih = [t for t in transaksi if mulai <= t["tanggal"] < selesai]
for t in terpilih:
    print(t["id"], t["tanggal"], t["jumlah"])
print("total:", sum(t["jumlah"] for t in terpilih))

text Keluaran

2 2026-11-02 15000
3 2026-11-02 30000
total: 45000

Keluaran hanya memuat baris dengan tanggal 2026-11-02. Baris 2026-11-01 berada di bawah batas mulai, sedangkan baris 2026-11-03 dan 2026-11-04 berada pada interval run lain. Seleksi semacam ini berlaku sama pada query basis data melalui WHERE dengan dua batas, atau pada pembacaan partisi penyimpanan objek melalui path bertanggal.

DAG dengan batas data interval#

Berkas dag_inkremental.py berikut menghubungkan batas konteks dengan seleksi baris. Timetable-nya sengaja menggunakan CronDataIntervalTimetable, yaitu timetable data interval klasik yang mengisi data_interval_start dan data_interval_end dengan rentang penuh satu hari. Pilihan ini berbeda dari string cron biasa pada Airflow 3 yang menghasilkan interval lebar nol. Task proses_interval membaca kedua batas dari konteks melalui get_current_context, memilih baris transaksi yang jatuh pada rentang tersebut, lalu mencetak setiap baris yang diproses beserta ringkasannya.

python dag_inkremental.py

import pendulum
from airflow.sdk import dag, task, get_current_context
from airflow.timetables.interval import CronDataIntervalTimetable

TRANSAKSI = [
    {"id": 1, "tanggal": "2026-11-01", "jumlah": 20000},
    {"id": 2, "tanggal": "2026-11-02", "jumlah": 15000},
    {"id": 3, "tanggal": "2026-11-02", "jumlah": 30000},
    {"id": 4, "tanggal": "2026-11-03", "jumlah": 12000},
    {"id": 5, "tanggal": "2026-11-04", "jumlah": 25000},
]


@dag(
    dag_id="dag_inkremental",
    schedule=CronDataIntervalTimetable("0 0 * * *", timezone=pendulum.timezone("UTC")),
    start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
    description="Memproses transaksi hanya pada data interval dag run.",
)
def dag_inkremental():
    @task
    def proses_interval():
        context = get_current_context()
        mulai = context["data_interval_start"].to_date_string()
        selesai = context["data_interval_end"].to_date_string()
        terpilih = [
            t for t in TRANSAKSI
            if mulai <= t["tanggal"] < selesai
        ]
        for t in terpilih:
            print(f"memproses {t['id']} tanggal {t['tanggal']} jumlah {t['jumlah']}")
        print(f"data interval [{mulai}, {selesai}): {len(terpilih)} baris")

    proses_interval()


dag_inkremental()

Perintah berikut menyiapkan folder metadata Airflow yang terpisah, lalu menyalin berkas DAG ke folder dags-nya. 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-pemrosesan-inkremental
mkdir -p "$AIRFLOW_HOME/dags"
cp dag_inkremental.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-pemrosesan-inkremental
airflow db migrate > /dev/null 2>&1

Struktur DAG yang sudah dimuat diperiksa sebelum dijalankan. Keluaran menegaskan nama timetable yang menentukan lebar data intervalnya.

bash

# muat modul DAG lalu cetak dag_id, timetable, dan seluruh task_id
export AIRFLOW_HOME=/tmp/ra-pemrosesan-inkremental
python3 -c "import dag_inkremental; d = dag_inkremental.dag_inkremental(); print(d.dag_id); print(type(d.timetable).__name__); print(sorted(t.task_id for t in d.tasks))"

text Keluaran

dag_inkremental
CronDataIntervalTimetable
['proses_interval']

DAG dijalankan dengan pemicuan 2026-11-03. Pada timetable data interval klasik, pemicuan pada waktu tertentu dipetakan ke interval yang berakhir pada waktu itu, sehingga run ini memproses data interval [2026-11-02, 2026-11-03). Log perintahnya panjang dan bertanda waktu, sehingga diarahkan ke berkas.

bash

# jalankan run untuk pemicuan 2026-11-03
export AIRFLOW_HOME=/tmp/ra-pemrosesan-inkremental
airflow dags test dag_inkremental 2026-11-03 > "$AIRFLOW_HOME/dags-test.log" 2>&1

Baris yang dicetak task proses_interval tersimpan pada log tersebut. Pencarian dengan dua kata kunci memperlihatkan cakupan kerja run.

bash

# tampilkan baris yang diproses beserta ringkasan intervalnya
grep -E "memproses|data interval" /tmp/ra-pemrosesan-inkremental/dags-test.log

text Keluaran

memproses 2 tanggal 2026-11-02 jumlah 15000
memproses 3 tanggal 2026-11-02 jumlah 30000
data interval [2026-11-02, 2026-11-03): 2 baris

Keluaran memuat dua baris milik 2026-11-02 beserta ringkasannya, sedangkan baris tanggal lain tidak tersentuh sama sekali. Run pada pemicuan berikutnya membaca interval [2026-11-03, 2026-11-04) dan memilih baris id 4. Setiap run membawa batasnya sendiri, sehingga pemrosesan berjalan bertahap tanpa satu pun run yang mengerjakan dua interval sekaligus.

Run ulang yang aman#

Batas yang eksplisit per run adalah bentuk idempotensi pada tingkat pipeline. Run ulang untuk interval yang sama memilih baris yang sama dan menulis hasil pada partisi yang sama, sehingga keadaan akhirnya sama dengan hasil satu kali eksekusi. Sifat inilah yang membuat backfill aman karena run lama dapat diulang berkali-kali tanpa menumpuk hasil.

Sisi yang rawan justru ada pada kursor apabila pipeline mengelolanya sendiri. Kursor yang dimajukan terlalu awal membuat rentang data dianggap selesai padahal pemrosesannya belum selesai.

Pola pada dag_inkremental.py menghindari masalah itu dengan tidak menyimpan kursor sama sekali. Batas per run dinyatakan oleh data interval yang dibawa dag run, dan pencatatan keberhasilan run diurus basis data metadata Airflow. Pipeline yang membutuhkan kursor, misalnya untuk sumber yang tidak terpartisi tanggal, memindahkan pembaruan kursor ke akhir task, sehingga kursor hanya dimajukan ketika pemrosesan berhasil.

Ringkasan#

Pemrosesan inkremental memproses rentang data baru per run, sedangkan full refresh menghitung ulang seluruh riwayat pada setiap run dengan biaya yang tumbuh seiring panjang riwayat.
Watermark menyatakan batas atas data yang dianggap final, sedangkan kursor menyatakan posisi terakhir yang sudah diproses dan disimpan terpisah dari kode.
Pada DAG terjadwal, data_interval_start dan data_interval_end dari konteks menjadi batas query per run, sehingga setiap run hanya menyentuh rentangnya sendiri.
Batas eksplisit membuat run ulang memilih baris yang sama dan menulis partisi yang sama, sedangkan kursor yang dikelola sendiri hanya boleh dimajukan setelah pemrosesan berstatus success.

Referensi#

Best Practices — Apache Airflow — Saran resmi membaca dan menulis pada partisi tertentu, dengan data_interval_start sebagai contoh partisinya.
DAGs — Apache Airflow — Definisi data interval dag run beserta kaitannya dengan logical date.
Timetables — Apache Airflow — Uraian timetable dan varian CronDataIntervalTimetable yang mengisi data interval secara penuh.