Konsep Airflow

Topik ini menjelaskan konsep dasar Apache Airflow versi 3, yaitu DAG, task, operator, scheduler, dan Dag run. Pembahasan bergerak dari cara DAG dinyatakan dalam kode, struktur pekerjaan di dalamnya, komponen server yang menjalankannya, sampai cara satu eksekusi dibingkai oleh tanggal. Setelah membaca, berkas DAG dan diagram arsitektur Airflow dapat dibaca dengan memahami peran setiap bagian.

DAG sebagai kode#

Airflow mendefinisikan pipeline sebagai DAG, yakni graf dependensi berarah yang tidak membentuk putaran tertutup, sehingga urutan task selalu dapat diselesaikan. Definisi tersebut ditulis dalam berkas Python biasa, sehingga pipeline diperlakukan seperti kode program lain, yaitu dapat dibaca, diuji sebagian, dan diperiksa dengan versi kontrol. Airflow memanggil pendekatan ini workflow as code, yakni alur kerja yang dinyatakan sepenuhnya sebagai kode.

Sebuah DAG dinyatakan lewat tiga bentuk, yaitu context manager, konstruktor biasa, dan dekorator @dag. Context manager adalah blok with DAG(...) yang mengaitkan segala task di dalamnya pada DAG itu. Konstruktor biasa menerima objek DAG secara eksplisit, sedangkan dekorator @dag ditempelkan pada sebuah fungsi. Ketiganya menghasilkan DAG yang sama, sehingga pilihan bentuk biasanya mengikuti gaya contoh yang dibaca. Bentuk context manager berikut menjadi titik banding dengan contoh lain di dokumentasi.

# tidak-dijalankan
import datetime

from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import DAG

with DAG(
    dag_id="pembersihan_arsip",
    schedule="@daily",
    start_date=datetime.datetime(2026, 1, 1),
    catchup=False,
) as dag:
    EmptyOperator(task_id="pindahkan_berkas_lama")

Argumen yang paling sering dipakai ada tiga, yaitu dag_id sebagai nama unik DAG, schedule sebagai pola penjadwalan seperti @daily atau ekspresi cron, dan start_date sebagai batas awal tanggal yang dipertimbangkan scheduler. Dua bentuk selanjutnya, yaitu dekorator @dag dan fungsi task, ditunjukkan berkas berikut, dan bentuk inilah yang dipakai pada topik menulis DAG.

# tidak-dijalankan
import datetime

from airflow.sdk import dag, task


@dag(
    dag_id="rekap_harian",
    schedule="@daily",
    start_date=datetime.datetime(2026, 1, 1),
    catchup=False,
)
def rekap_harian():
    @task
    def baca_sumber() -> list[str]:
        """Mengembalikan baris data yang sudah dibaca dari sumber."""
        return ["2026-01-01,12", "2026-01-01,30"]

    @task
    def hitung_jumlah(data: list[str]) -> int:
        """Menghitung jumlah baris yang diterima dari task sebelumnya."""
        return len(data)

    hitung_jumlah(baca_sumber())


rekap_harian()

Task, operator, dan statusnya#

Task adalah satu unit pekerjaan di dalam DAG, sedangkan operator adalah templat untuk jenis pekerjaan tertentu, dan task dibuat dari templat itu. PythonOperator membungkus fungsi Python, BashOperator menjalankan perintah shell, dan EmptyOperator tidak mengerjakan apa pun selain menjadi simpul struktur. Dari satu task dalam definisi, Airflow membuat task instance, yakni wujud task yang dijalankan pada satu Dag run, dan setiap task instance memiliki status tersendiri.

Tabel status di atas penting untuk dibaca pada UI dan log, karena pesan kegagalan sering berasal dari status task lain. Status upstream_failed, misalnya, berarti task yang sedang diperiksa tidak pernah dijalankan, sehingga pencarian penyebabnya perlu naik ke task di hulunya.

Scheduler dan komponen yang menjalankan DAG#

Scheduler adalah komponen server yang memicu eksekusi DAG sesuai jadwalnya, kemudian mengirimkan task yang siap ke executor, yakni pengatur antrian eksekusi yang menjadi bagian konfigurasi scheduler. Berdasarkan dokumentasi arsitektur versi 3, komponen pendukungnya ada empat. Dag processor membaca berkas DAG dan menuliskan hasilnya ke basis data metadata. Worker menjalankan task yang dikirim executor, API server melayani UI dan permintaan programatik, sedangkan triggerer menangani task yang menunggu kejadian secara asinkron.

Pemisahan ini menjelaskan mengapa berkas DAG tidak langsung dieksekusi saat disimpan. Berkas itu dipindai dag processor secara berkala, lalu DAG yang valid tersedia untuk dijadwalkan. Perubahan pada kode DAG pun tidak berlaku seketika, tetapi menunggu siklus pemindaian berikutnya, sehingga daftar DAG pada UI bisa tertinggal beberapa saat dari isi folder. Keberadaan dag processor sebagai komponen tersendiri juga menjadi salah satu perbedaan yang menonjol dibanding versi 2.x, sehingga diagram arsitektur lama tidak dapat dibaca apa adanya.

Dag run dan tanggal logis#

Satu eksekusi DAG disebut Dag run, yakni salinan DAG yang hidup bersama rentang data (data interval) yang diwakilinya. Untuk DAG harian yang menjalankan rekap tanggal 1 Januari, Dag run-nya membawa rentang data satu hari tersebut, sehingga kode task yang sama menghasilkan data yang berbeda untuk setiap hari tanpa perlu mengubah kode. Tanggal yang menandai tujuan eksekusi disebut logical date, yakni tanggal logis yang menandai awal rentang data, dan tanggal itu berbeda dari saat Dag run sungguhan dijalankan.

Pemisahan antara waktu eksekusi dan rentang data inilah yang membuat pengisian ulang data lama (backfill) menjadi mungkin. Airflow dapat menjalankan salinan Dag run untuk tanggal-tanggal sebelumnya secara berurutan, dan tiap salinan bekerja pada rentang datanya sendiri. Parameter catchup=False pada contoh sebelumnya menghentikan pengejaran otomatis atas tanggal-tanggal yang terlewat sejak start_date. Tanpa pengaturan itu, DAG harian yang baru dipasang dapat langsung mengantrekan Dag run untuk banyak hari sebelumnya, sehingga pemasangan pertama menjadi mahal tanpa disadari.

Kesalahan umum#

1. Mengira dekorator @task mengeksekusi fungsi seketika. Penyebabnya, bentuk pemanggilan hitung_jumlah(baca_sumber()) memang menyerupai pemanggilan fungsi biasa. Perbaikannya, pahami bahwa pemanggilan itu mendaftarkan task beserta dependensinya, sedangkan eksekusinya menunggu jadwal dari Airflow.
2. Memberi task_id yang sama untuk dua task dalam satu DAG. Penyebabnya, task_id menjadi kunci identitas task pada UI, log, dan dependensi. Perbaikannya, pastikan setiap task memiliki nama unik dan deskriptif.
3. Menunggu perubahan kode DAG berlaku seketika. Penyebabnya, pemindaian berkas DAG oleh dag processor berjalan berkala, bukan pada setiap penyimpanan berkas. Perbaikannya, tunggu siklus pemindaian berikutnya atau periksa pesan impor error pada halaman daftar DAG sebelum menyimpulkan kode gagal.
4. Menyamakan tanggal penjadwalan dengan rentang data. Penyebabnya, tanggal pada jadwal terlihat seperti waktu pekerjaan berjalan, padahal keduanya berbeda pada Dag run terjadwal. Perbaikannya, baca rentang data pada rincian Dag run sebelum menyimpulkan bahwa task dijalankan pada tanggal yang keliru.

Ringkasan#

• DAG dinyatakan sebagai kode Python lewat context manager, konstruktor biasa, atau dekorator @dag, dengan schedule dan start_date sebagai argumen penjadwalan utamanya.
• Operator menjadi templat pekerjaan, task menjadi unit pekerjaan, dan setiap task instance membawa status seperti success, failed, skipped, up_for_retry, dan upstream_failed.
• Scheduler memicu Dag run sesuai jadwal, dag processor memindai berkas DAG, dan worker menjalankan task yang dikirim executor.
• Setiap Dag run membawa rentang data dan tanggal logisnya sendiri, sehingga kode yang sama dapat dijalankan ulang untuk tanggal-tanggal lama.

Referensi#

• DAGs — Apache Airflow — Dokumentasi resmi konsep DAG, cara penulisannya, jadwal, dan Dag run.
• Tasks — Apache Airflow — Dokumentasi resmi task, task instance, daftar statusnya, dan batas waktu eksekusi.
• Airflow architecture — Apache Airflow — Dokumentasi komponen scheduler, dag processor, executor, worker, dan triggerer pada versi 3.