Apache Airflow

Cron klasik menjalankan pipeline pada jam yang tetap, tetapi cron tidak tahu apa yang terjadi setelah skrip dibiarkan berjalan: dependensi antarlangkah, percobaan ulang saat gagal, sampai riwayat run untuk audit. Apache Airflow lahir untuk mengisi kekosongan tersebut — orchestrator open-source yang menyatakan pipeline sebagai kode, lengkap dengan penjadwalan, retry, dan antarmuka pemantauan. Airflow bermula di Airbnb pada 2014, menjadi project Apache pada 2019, dan kini menjadi orchestrator data yang paling banyak dipakai di industri. Versi stabil saat riset course ini ditulis adalah Airflow 3.3.x, yang mengubah gaya penulisan DAG secara signifikan dibandingkan Airflow 2. Kita akan membedah arsitekturnya, menulis DAG lengkap, lalu meninjau konsep deployment dan posisi Airflow di ekosistem.

Arsitektur Airflow #

Airflow berjalan sebagai kumpulan komponen yang bekerja di atas satu metadata database:

• Scheduler — membuat DAG run saat jadwalnya tiba dan mengantrekan task yang dependensinya sudah terpenuhi.
• DAG Processor — memparsing berkas DAG secara berkala ke metadata database; sejak Airflow 3 komponen ini dapat berjalan terpisah dari scheduler.
• Executor dan Worker — executor menentukan di mana task dieksekusi: LocalExecutor pada satu mesin, CeleryExecutor pada sekelompok worker, atau KubernetesExecutor dengan satu pod per task.
• Metadata DB — umumnya PostgreSQL; menyimpan definisi DAG, status run, task instance, dan rujukan ke log.
• API Server dan UI — sejak Airflow 3, webserver dilebur ke API server yang sekaligus menyajikan REST API dan antarmuka web.

Alur kerjanya berpusat pada metadata database. DAG processor memparsing berkas DAG dan menyimpan definisinya; scheduler membaca definisi tersebut, dan saat jadwal tiba, scheduler membuat DAG run lalu mengantrekan task ke executor. Worker menjalankan task dan menuliskan statusnya kembali, sementara UI cukup membaca metadata database untuk menampilkan riwayat run.

Konsep Inti: DAG, Task, dan TaskFlow #

DAG (directed acyclic graph) adalah definisi urutan kerja: task menjadi simpul dan dependency menjadi sisi — materi yang telah kita bahas pada Chapter Dependency dan Task. Cara klasik menulis task memakai operator seperti PythonOperator atau BashOperator yang dirangkai dengan >>. Airflow modern menganjurkan TaskFlow API: task didefinisikan sebagai fungsi Python biasa lewat dekorator @task, dan nilai balik satu task otomatis dialirkan ke argumen task berikutnya.

Pengaliran nilai tersebut terjadi lewat XCom — penyimpanan kecil di metadata database untuk data ringkas antar task. XCom cocok untuk membawa ringkasan seperti jumlah baris atau path berkas; dataset besar tetap ditulis ke storage eksternal dan hanya lokasinya yang dialirkan.

Menulis DAG Gaya Airflow 3 #

Sejak Airflow 3, import dekorator berpindah ke airflow.sdk. Bentuk DAG paling sederhana:

# Sketsa DAG minimal — Apache Airflow 3.
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.

import pendulum
from airflow.sdk import dag, task


@dag(
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["contoh"],
)
def dag_minimal():
    @task
    def sapa():
        print("pipeline penjualan berjalan")


dag_minimal()

Parameter schedule=None berarti DAG hanya berjalan saat dipicu manual dari UI atau CLI — titik awal yang aman untuk menguji DAG baru, sebelum pindah ke jadwal cron dari Chapter Scheduling. Sekarang kita naik ke bentuk yang lebih realistis: pipeline penjualan harian dengan tiga task dan retry.

# DAG pipeline penjualan harian — gaya Airflow 3 (TaskFlow API).
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.

import pendulum
from airflow.sdk import dag, task


@dag(
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["penjualan"],
)
def pipeline_penjualan_harian():
    @task
    def ekstrak_pesanan() -> list[dict]:
        """Baca pesanan baru dari tabel staging (ilustrasi)."""
        return [
            {"id": 1, "pelanggan": "Budi", "total": 150000},
            {"id": 2, "pelanggan": "Siti", "total": 75000},
        ]

    @task(retries=2)
    def hitung_ringkasan(pesanan: list[dict]) -> dict:
        total = sum(p["total"] for p in pesanan)
        return {"jumlah": len(pesanan), "pendapatan": total}

    @task
    def muat_ke_warehouse(ringkasan: dict) -> None:
        """Tulis ringkasan ke tabel warehouse (ilustrasi)."""
        print(f"dimuat {ringkasan['jumlah']} pesanan, "
              f"pendapatan {ringkasan['pendapatan']}")

    pesanan = ekstrak_pesanan()
    ringkasan = hitung_ringkasan(pesanan)
    muat_ke_warehouse(ringkasan)


pipeline_penjualan_harian()

Mari kita telaah baris-baris kode di atas:

1. Baris 4-5 (import): pendulum menyediakan tanggal sadar zona waktu untuk start_date; dekorator dag dan task diimpor dari airflow.sdk — pola Airflow 3 yang menggantikan airflow.decorators.
2. Baris 8-13 (@dag(...)): dekorator mengubah fungsi menjadi objek DAG. schedule="0 6 * * *" menjalankan DAG setiap hari pukul 06:00, sedangkan catchup=False mencegah Airflow mengejar jadwal yang terlewat sejak start_date — topik yang kita bahas tuntas pada Chapter Backfill.
3. Baris 15-21 (ekstrak_pesanan): dekorator @task membungkus fungsi menjadi task; nilai baliknya disimpan sebagai XCom sehingga task berikutnya dapat menerimanya.
4. Baris 23-26 (hitung_ringkasan): argumen pesanan berisi hasil task sebelumnya; retries=2 memerintahkan Airflow mengulang task hingga dua kali saat gagal, selaras dengan konsep Chapter Retries dan Timeouts.
5. Baris 34-36 (pemanggilan): urutan pemanggilan fungsi membentuk dependency ekstrak_pesanan → hitung_ringkasan → muat_ke_warehouse; Airflow menyusun graf dari pemanggilan ini tanpa operator >>.

Perilaku yang diharapkan: DAG muncul di UI setelah berkas diparsing, berjalan setiap pukul 06:00, dan ketika hitung_ringkasan gagal karena gangguan sementara, Airflow mengulang task tersebut lebih dahulu sebelum menandai run gagal. Contoh resmi gaya penulisan ini tersedia di tutorial TaskFlow Airflow .

Deployment Singkat dengan Docker Compose #

Dokumentasi resmi menyediakan satu berkas docker-compose.yaml yang memuat seluruh komponen arsitektur di atas. Struktur intinya, disederhanakan:

# Struktur disederhanakan dari docker-compose.yaml resmi Airflow 3.
x-airflow-common: &airflow-common
  environment: &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow

services:
  postgres:
    image: postgres:13

  airflow-apiserver:
    <<: *airflow-common
    command: api-server
    ports:
      - "8080:8080"

  airflow-scheduler:
    <<: *airflow-common
    command: scheduler

  airflow-dag-processor:
    <<: *airflow-common
    command: dag-processor

Anchor YAML (&airflow-common dan <<: *airflow-common) memastikan tiga service Airflow berbagi konfigurasi environment yang sama. Alur memulainya, secara konsep:

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'
docker compose up airflow-init
docker compose up -d

Service airflow-init membuat schema database dan pengguna awal; setelah semua service berjalan, UI terbuka di port 8080 dan berkas DAG tinggal diletakkan di folder dags. Untuk skala besar, berkas resmi yang sama menyediakan varian CeleryExecutor dengan service redis dan worker tambahan.

Kekuatan dan Kelemahan #

Airflow paling kuat ketika pipeline berbentuk batch terjadwal yang membutuhkan audit dan ekosistem integrasi luas. Ketika pertanyaan utama bergeser dari “langkah apa yang dijalankan” menjadi “data apa yang harus dijaga kesegarannya”, paradigma aset dari Chapter Dagster menjadi alternatif yang layak dinilai. Rujukan lengkapnya ada di dokumentasi Apache Airflow .