Kalender penuh script cron, job Python yang jalan jam tiga pagi, lalu gagal diam-diam tanpa ada yang tahu — kalau kamu pernah merangkai beberapa proses data yang harus berurutan tiap hari, kamu paham betapa cepatnya setup sederhana berubah jadi benang kusut. Apache Airflow menata ulang semua itu: pipeline didefinisikan sebagai kode Python, dan Airflow yang mengurus jadwal, urutan eksekusi, retry, sampai log per task.

Airflow vs Cron vs Celery

Cron bisa menjalankan satu perintah di jam tertentu, tapi tidak paham dependensi — kalau proses A belum selesai, cron tetap meluncurkan B. Airflow memodelkan alur kerja sebagai DAG (Directed Acyclic Graph): tiap task adalah node, dan panah antar task menentukan urutannya. Task yang gagal bisa di-retry otomatis, dan setiap task menyimpan lognya sendiri-sendiri.

Bedanya dengan task queue seperti Celery: Celery mengeksekusi task yang dikirim aplikasi secara asinkron, sedangkan Airflow fokus menjadwalkan dan mengorkestrasi alur yang berjalan rutin. Untuk ETL harian, training model mingguan, atau pipeline reporting, Airflow lebih pas. Kalau yang kamu butuh eksekusi cepat sebagai respons request, Celery tetap juaranya.

Jangan pakai Airflow untuk data streaming real-time. Airflow menjadwalkan batch; untuk event per detik, engine streaming seperti Kafka lebih cocok.

Instalasi

Airflow butuh Python 3.8 ke atas. Cara paling cepat adalah lewat venv lalu menjalankan perintah standalone, yang otomatis menginisialisasi database, membuat user admin, dan menyalakan webserver plus scheduler sekaligus.

bash
python -m venv airflow-env
source airflow-env/bin/activate
pip install "apache-airflow==2.10.3"
export AIRFLOW_HOME=~/airflow
airflow standalone

Tunggu beberapa menit sampai dependensi selesai terpasang — Airflow memang gemuk, instalasinya bisa ratusan megabyte. Setelah selesai, buka http://localhost:8080 dan login pakai kredensial yang tercetak di terminal (username-nya admin, password-nya muncul saat standalone siap).

airflow standalone menyatukan semua komponen dalam satu proses, cocok untuk belajar. Untuk produksi, webserver dan scheduler harus berjalan terpisah — biasanya lewat Docker atau Kubernetes.

DAG Pertama

DAG adalah inti Airflow: file Python berisi task dan aturan urutannya. Tidak boleh ada task yang membentuk siklus — kalau ada loop, Airflow menolak memuat file itu. Mulai dari yang sederhana.

python
from datetime import datetime

from airflow import DAG
from airflow.decorators import task
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="pipeline_pertama",
    start_date=datetime(2026, 10, 1),
    schedule="@daily",
    catchup=False,
    tags=["contoh"],
) as dag:

    mulai = BashOperator(
        task_id="mulai",
        bash_command="echo 'Pipeline dimulai: {{ ds }}'",
    )

    @task
    def ambil_data():
        return {"baris": 128, "sumber": "api"}

    @task
    def proses(data: dict) -> int:
        return data["baris"] * 2

    ambil = ambil_data()
    hitung = proses(ambil)

    mulai >> ambil >> hitung

Operator >> menandai dependensi: mulai jalan dulu, lalu ambil, baru hitung. Decorator @task mengubah fungsi Python biasa jadi task, dan nilai yang di-return otomatis diteruskan ke task berikutnya lewat XCom. Simpan file ini di $AIRFLOW_HOME/dags/ dengan nama bebas, misalnya pipeline_pertama.py, lalu Airflow akan memuatnya dalam beberapa detik.

Menjadwalkan DAG

Parameter schedule menerima preset atau ekspresi cron. Beberapa contoh yang sering dipakai:

python
schedule="@daily"            # preset: tiap hari
schedule="0 3 * * *"         # cron: jam 03.00 tiap hari
schedule="30 9 * * 1-5"      # cron: 09.30, Senin-Jumat

Dua hal yang paling sering bikin bingung pemula: start_date dan catchup. Airflow menjadwalkan DAG mulai dari start_date, dan kalau catchup=True, ia akan "mengejar" semua jadwal yang terlewat sejak tanggal itu. Untuk belajar, set catchup=False supaya Airflow tidak tiba-tiba memicu puluhan run lama.

Kirim Data Antar Task dengan XCom

Task di Airflow terisolasi — output satu task tidak otomatis terlihat task lain. XCom (cross-communication) adalah mekanisme untuk menyelipkan nilai kecil antar task. Dengan decorator @task, return value sudah ditangani otomatis; kalau pakai operator klasik, kamu push dan pull manual seperti ini.

python
from datetime import datetime

from airflow import DAG
from airflow.operators.python import PythonOperator

def dorong(**context):
    context["ti"].xcom_push(key="konfigurasi", value={"batch": 500})

def tarik(**context):
    nilai = context["ti"].xcom_pull(key="konfigurasi", task_ids="dorong")
    print("Nilai diterima:", nilai)

with DAG(
    dag_id="contoh_xcom",
    start_date=datetime(2026, 10, 1),
    schedule=None,
    catchup=False,
) as dag:
    t1 = PythonOperator(task_id="dorong", python_callable=dorong)
    t2 = PythonOperator(task_id="tarik", python_callable=tarik)
    t1 >> t2

XCom cocok untuk nilai kecil seperti ID, URL, atau path file. Jangan dipakai untuk memindahkan dataset raksasa — untuk data besar, simpan ke file atau object storage, lalu lempar path-nya lewat XCom.

Pipeline ETL Sederhana

Sekarang gabungkan semuanya: ekstrak, transform, load. Contoh berikut mensimulasikan tarikan data penjualan, menghitung total per baris, lalu "memuatnya" — di sini cukup dicetak ke log.

python
from datetime import datetime

from airflow import DAG
from airflow.decorators import task

with DAG(
    dag_id="etl_penjualan",
    start_date=datetime(2026, 10, 1),
    schedule="@daily",
    catchup=False,
) as dag:

    @task
    def ekstrak() -> list:
        # simulasi tarik data dari API atau CSV
        return [
            {"produk": "laptop", "qty": 3, "harga": 12000000},
            {"produk": "mouse", "qty": 40, "harga": 150000},
            {"produk": "keyboard", "qty": 25, "harga": 350000},
        ]

    @task
    def transform(baris: list) -> list:
        hasil = []
        for r in baris:
            r = dict(r)
            r["total"] = r["qty"] * r["harga"]
            hasil.append(r)
        return hasil

    @task
    def load(baris: list):
        total = sum(r["total"] for r in baris)
        print(f"{len(baris)} baris diproses, total pendapatan: Rp {total:,}")

    data = ekstrak()
    tabel = transform(data)
    load(tabel)

Di kasus nyata, langkah transform biasanya memakai Pandas untuk agregasi, dan load menulis ke database atau data warehouse. Struktur dasarnya tetap sama: tiap langkah jadi task, XCom jadi jembatannya.

Tulis setiap task agar idempoten — bisa dijalankan ulang tanpa menghasilkan data ganda. Kalau task load menulis ke tabel tanpa dedup, retry otomatis Airflow bisa menggandakan baris. Gunakan INSERT ... ON CONFLICT atau hapus dulu data periode yang sama sebelum insert.

Pantau di Dashboard

Buka kembali http://localhost:8080. Grid view menampilkan status tiap run: hijau sukses, merah gagal, biru sedang jalan. Klik task yang gagal untuk membaca log lengkapnya, lalu perbaiki kode DAG dan trigger ulang dari tombol Trigger DAG. Dari CLI kamu bisa memakai airflow dags list, airflow dags trigger etl_penjualan, atau airflow dags backfill untuk menjalankan ulang rentang tanggal tertentu.

Untuk produksi, jangan jalankan standalone. Setup yang wajar memakai Docker (Airflow menyediakan docker-compose resmi) dengan webserver, scheduler, dan database metadata yang terpisah. Untuk belajar, standalone sudah lebih dari cukup.

Langkah Selanjutnya

Mulai dari satu pipeline kecil yang benar-benar kamu butuhkan — misalnya job yang mengunduh CSV harian lalu mengirim ringkasan ke Telegram. Setelah DAG pertama jalan, tambahkan default_args={'retries': 3} untuk retry otomatis, atau alert email saat task gagal. Begitu kamu punya masalah nyata, konsep seperti backfill dan dataset scheduling akan terasa jauh lebih masuk akal.