Apache Airflow adalah orchestrator: program yang menjalankan langkah-langkah pipeline kamu sesuai jadwal, dalam urutan yang benar, mengulang yang gagal, dan mencatat setiap run. Bayangin stasiun kereta. Jadwalnya bilang kereta paket berangkat tiap hari jam 00.00. Kepala stasiun melihat jam, memberangkatkan kereta, menahan kereta berikutnya kalau relnya belum kosong, dan mencatat tiap keberangkatan di buku. Kereta yang berangkat 2 Oktober jam 00.00 membawa paket tanggal 1 Oktober, jadi di karcisnya tertulis 1 Oktober. Jadwal itu DAG, kepala stasiun itu scheduler, satu keberangkatan itu satu run.
🧪 Tebakan cepat: DAG Toko Buku jalan tiap hari jam 00.00 WIB dan tiap run memuat penjualan satu hari. Run yang memuat penjualan 1 Oktober jalan kapan: 1 Oktober jam 00.00, atau 2 Oktober jam 00.00? Tebak dulu, jawabannya ada di bagian "Data interval dan logical date".
DAG, task, operator, scheduler
- DAG (Directed Acyclic Graph): kumpulan task beserta panahnya. Directed: panahnya searah, dari task yang harus selesai dulu ke task sesudahnya. Acyclic: gak ada putaran; task gak bisa menunggu dirinya sendiri. Satu DAG ditulis sebagai satu file Python.
- Task: satu unit kerja dalam DAG, misalnya "muat penjualan satu tanggal".
- Operator: cetakan untuk membuat task yang menjalankan aksi:
BashOperatormenjalankan perintah shell,PythonOperator(atau@task) memanggil fungsi Python,SQLExecuteQueryOperatormenjalankan query. - Sensor: operator khusus yang menunggu kondisi terpenuhi, misalnya
FileSensormenunggu file dari marketplace muncul, baru task sesudahnya boleh jalan. - Scheduler: proses yang membaca DAG, membuat run saat waktunya tiba, dan mengirim task ke worker. Di Airflow 3 dia ditemani dag processor (membaca file DAG), API server (UI dan API untuk task), dan triggerer.
Versi terbaru Airflow per 11 Oktober 2026 adalah 3.3.2 (rilis 17 September 2026). Untuk pelajaran ini Airflow dipasang di virtual environment Python 3.13 dengan pip install "apache-airflow==3.3.2" --constraint ... (file constraint resmi untuk versi itu), sekitar 185 MB. Semua output di bawah dari instalasi itu.
DAG pertama Toko Buku
DAG ini menjalankan pipeline.py dari pelajaran sebelumnya, satu tanggal per run. Ditulis dengan TaskFlow API: fungsi Python biasa yang diberi decorator @dag dan @task.
from datetime import date, timedelta
import pendulum
from airflow.sdk import CronDataIntervalTimetable, dag, task
import pipeline # pipeline.py dari pelajaran sebelumnya, satu folder dengan file ini
WIB = "Asia/Jakarta"
@dag(
schedule=CronDataIntervalTimetable("0 0 * * *", timezone=WIB),
start_date=pendulum.datetime(2026, 9, 28, tz=WIB),
catchup=True,
max_active_runs=1,
default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)
def toko_buku_harian():
@task
def tentukan_tanggal(data_interval_start=None) -> str:
return data_interval_start.in_timezone(WIB).date().isoformat()
@task
def muat(tanggal: str) -> int:
hari = date.fromisoformat(tanggal)
bersih = pipeline.transform(pipeline.extract(hari))
pipeline.load(bersih, hari)
return len(bersih)
@task
def laporkan(tanggal: str, baris: int) -> None:
print(f"penjualan {tanggal}: {baris} baris")
tanggal = tentukan_tanggal()
laporkan(tanggal, muat(tanggal))
toko_buku_harian()
schedule=CronDataIntervalTimetable("0 0 * * *", timezone=WIB): tiap hari jam 00.00 WIB, dan tiap run punya interval satu hari penuh."0 0 * * *"adalah ekspresi cron: menit 0, jam 0, tiap tanggal, tiap bulan, tiap hari.start_dateadalah awal interval pertama.catchupdanretriesdibahas di bawah.- Memanggil
muat(tanggal)dengan hasiltentukan_tanggal()membuat panah:tentukan_tanggalharus selesai dulu. Bentuk klasiknya ditulis dengan>>, misalnyaextract >> transform >> load. data_interval_start=Nonedi parameter task membuat Airflow mengisinya dari run yang sedang jalan.
Dijalankan sekali tanpa scheduler dengan airflow dags test toko_buku_harian 2026-10-02T00:00:00+07:00 (run manual memproses interval lengkap terakhir sebelum waktu itu, yaitu 1 Oktober), task muat mencatat log pipeline.py yang sama dengan pelajaran sebelumnya: extract 2026-10-01: 3 baris, 1 baris dobel dibuang, load 2026-10-01: 2 baris. Lalu laporkan mencetak penjualan 2026-10-01: 2 baris.
XCom: untuk nilai kecil saja
Nilai yang di-return sebuah task disimpan Airflow sebagai XCom (cross-communication) di database metadata-nya, lalu diberikan ke task yang membutuhkannya. Di run tadi, log Airflow mencatat Done. Returned value was: 2026-10-01 untuk tentukan_tanggal dan Done. Returned value was: 2 untuk muat: dua XCom, sebuah tanggal dan sebuah angka.
XCom dirancang untuk nilai kecil: tanggal, jumlah baris, path file. Data besarnya tetap di warehouse atau di file, dan task berikutnya membacanya dari sana. DataFrame jutaan baris yang dilewatkan lewat XCom membuat database metadata Airflow membengkak.
Data interval dan logical date
Tiap run terjadwal punya empat waktu:
- data interval (
data_interval_startsampaidata_interval_end): rentang data yang menjadi jatah run itu. - logical date (
logical_date): label run, sama dengan awal interval. - jalan: run baru boleh jalan setelah intervalnya selesai, yaitu di
data_interval_end.
Dihitung oleh Airflow 3.3.2 untuk DAG di atas (semua jam WIB):
| Run jalan | logical_date |
Data interval | ds |
|---|---|---|---|
| 29 Sep 00.00 | 28 Sep 00.00 | 28 Sep 00.00 sampai 29 Sep 00.00 | 2026-09-27 |
| 2 Okt 00.00 | 1 Okt 00.00 | 1 Okt 00.00 sampai 2 Okt 00.00 | 2026-09-30 |
| 11 Okt 00.00 | 10 Okt 00.00 | 10 Okt 00.00 sampai 11 Okt 00.00 | 2026-10-09 |
Run berlabel 1 Oktober jalan 2 Oktober jam 00.00, setelah harinya habis, karena baru saat itu semua pesanan 1 Oktober sudah ada. Itu jawaban tebakannya. Label yang tertinggal satu hari dari jam jalannya ini kebingungan klasik di Airflow.
Dua hal yang bikin makin licin:
dsitu tanggal UTC. Variabel template{{ ds }}adalahlogical_datedalam UTC. 1 Oktober 00.00 WIB = 30 September 17.00 UTC, jadidsbilang2026-09-30. Karena itu DAG di atas mengubahdata_interval_startke WIB dulu baru mengambil tanggalnya.- Di Airflow 3,
schedule="@daily"gak punya interval. Cron biasa sekarang memakaiCronTriggerTimetable: run 2 Oktober 00.00 berlabel 2 Oktober 00.00, dandata_interval_startsama dengandata_interval_end, yaitu 2 Oktober 00.00. Task di atas akan memuat 2 Oktober, hari yang baru mulai: 0 baris. Di Airflow 2,"@daily"masih memakai interval. Jadi tulis timetable-nya terang-terangan seperti di DAG tadi, atau beriCronTriggerTimetablesebuahinterval=timedelta(days=1).
Catchup dan backfill
DAG di atas dinyalakan 11 Oktober jam 21.00 WIB, padahal start_date-nya 28 September.
catchup=True: scheduler langsung membuat run untuk tiap interval yang sudah selesai: 13 run, 28 September sampai 10 Oktober, dijalankan berurutan (max_active_runs=1).catchup=False(default di Airflow 3): cuma interval terakhir, 10 Oktober. Sejarah sebelumnya gak diisi.
Tanggal lama bisa diisi kapan saja dengan backfill:
airflow backfill create --dag-id toko_buku_harian \
--from-date 2026-09-28T00:00:00+07:00 --to-date 2026-10-09T00:00:00+07:00
Hasilnya 12 run, logical date 28 September sampai 9 Oktober, yang dijalankan scheduler dan kelihatan di UI. Tulis jam dan zona waktunya: tanggal polos 2026-09-28 dibaca sebagai tengah malam UTC (07.00 WIB), dan --dry-run menunjukkan run 28 September WIB ikut terlewat. Backfill dan catchup aman karena load di pipeline.py idempoten: tanggal yang dimuat ulang dihapus dulu.
Retry dan status
default_args={"retries": 2, "retry_delay": timedelta(minutes=5)} berlaku untuk semua task: yang gagal dicoba lagi 2 kali (total 3 percobaan) dengan jeda 5 menit. Selama menunggu, statusnya up_for_retry; kalau tetap gagal, failed, dan task sesudahnya jadi upstream_failed tanpa dijalankan. Hati-hati, retry-nya berlapis: pipeline.extract sendiri mencoba konek 3 kali, jadi sumber yang mati total dicoba 3 × 3 = 9 kali sebelum run dinyatakan gagal.
💡 Yang baru di Airflow 3
Airflow 3.0 keluar April 2025. Perubahan yang terasa bagi penulis DAG, menurut panduan resmi "Upgrading to Airflow 3":
- UI baru, dan DAG versioning: tiap run mencatat versi kode DAG yang menjalankannya.
- Asset (dulu dataset): DAG bisa dijadwalkan "jalan setiap tabel ini diperbarui" (
schedule=[Asset(...)]), bukan cuma per jam. - Import untuk penulis DAG pindah ke
airflow.sdk, dan task gak boleh lagi mengakses database metadata Airflow langsung. - Backfill dijalankan scheduler (
airflow backfill create),catchupdefaultFalse, dan cron biasa jadiCronTriggerTimetable. - Dihapus: SubDAG (ganti TaskGroup atau Asset), SLA (ganti Deadline Alerts),
execution_date(pakailogical_date), dan argumenschedule_interval(pakaischedule).
⚠️ Jebakan yang sering ketemu
- Menganggap
dstanggal WIB.dsitu UTC: di WIB, run jam 00.00 selalu punyadssehari sebelumnya. Ambil tanggal daridata_interval_startyang diubah ke zona waktumu. "@daily"di Airflow 3 untuk load per hari. Gak ada interval, jadi task memuat hari yang baru mulai. PakaiCronDataIntervalTimetableatau beri interval.- Data besar lewat XCom. Lewatkan tanggal atau path, bukan DataFrame.
- Kerja berat di level atas file DAG. Kode di luar fungsi task jalan tiap kali file DAG dibaca ulang, jauh lebih sering dari jadwalnya. Query dan panggilan API taruh di dalam task.
- Task yang gak idempoten. Retry, catchup, dan backfill menjalankan ulang task; tanpa
DELETElaluINSERT, angkanya dobel.
TL;DR: Airflow menjalankan pipeline sesuai jadwal: DAG (Directed Acyclic Graph) adalah task beserta urutannya tanpa putaran; operator menjalankan aksi, sensor menunggu kondisi; scheduler membuat run. Versi terbaru 3.3.2. DAG Toko Buku memakai TaskFlow (@dag, @task, import dari airflow.sdk) dan CronDataIntervalTimetable("0 0 * * *", timezone="Asia/Jakarta"), jadi run berlabel 1 Oktober (logical_date = awal interval) jalan 2 Oktober 00.00 setelah intervalnya selesai. ds itu tanggal UTC (2026-09-30 untuk run itu), dan "@daily" di Airflow 3 gak punya interval. XCom cuma untuk nilai kecil seperti tanggal dan jumlah baris. catchup=True membuat run untuk tiap interval yang terlewat sejak start_date (13 run), catchup=False (default) cuma yang terakhir, dan airflow backfill create dengan jam dan zona waktu mengisi tanggal lama. retries=2 berarti 3 percobaan; task harus idempoten.
