Bagian 3: Membuat DAG

Struktur proyek sudah berdiri; sekarang kita menulis jantungnya: dags/etl_pesanan.py. Kita memakai TaskFlow API gaya Airflow 3, yang menandai task sebagai fungsi Python biasa berdekorator @task alih-alih objek operator eksplisit. Sejak Airflow 3, import dekorator berpindah ke modul airflow.sdk — perubahan kecil yang membuat banyak contoh lama di internet tidak lagi relevan. Bagian ini membedah berkas DAG dari baris import pertama sampai instansiasi terakhir.

Anatomi Bagian Awal Berkas DAG #

from __future__ import annotations

import os
import sys
from datetime import timedelta
from pathlib import Path

import pendulum
from airflow.sdk import dag, task

# dags/ berada satu tingkat di bawah akar proyek. Airflow hanya menaruh
# folder dags di sys.path, jadi akar proyek ditambahkan manual agar paket
# src/ bisa diimpor oleh berkas DAG ini.
PROYEK_DIR = Path(__file__).resolve().parents[1]
if str(PROYEK_DIR) not in sys.path:
    sys.path.insert(0, str(PROYEK_DIR))

from src import state as state_etl  # noqa: E402
from src import tasks as etl  # noqa: E402

DAG_ID = "etl_pesanan_tokokita"

RAW_DIR = Path(os.environ.get("TOKOKITA_RAW_DIR", PROYEK_DIR / "data" / "raw"))
GOLD_DIR = Path(os.environ.get("TOKOKITA_GOLD_DIR", PROYEK_DIR / "data" / "gold"))
STATE_PATH = Path(
    os.environ.get("TOKOKITA_STATE_PATH", PROYEK_DIR / "data" / "state" / "state.json")
)

Mari kita telaah baris-baris kode di atas:

1. Baris 20-21 (import pendulum dan from airflow.sdk import dag, task): Dua import ini adalah sidik khas Airflow 3. Modul airflow.sdk menjadi rumah dekorator TaskFlow sejak Airflow 3, menggantikan airflow.decorators pada Airflow 2. pendulum dipakai untuk start_date yang sadar zona waktu.
2. Baris 26-28 (PROYEK_DIR dan sys.path.insert): Airflow hanya menaruh folder dags/ di sys.path, sedangkan paket src/ kita berada satu tinggi di atasnya. Tiga baris ini menambahkan akar proyek ke sys.path sehingga from src import tasks berhasil saat scheduler mengurai berkas DAG.
3. Baris 30-31 (import src): Logika ETL diimpor sebagai modul murni. Task pada DAG hanya akan memanggil etl.baca_pesanan dan kawan-kawannya — tidak ada logika ETL yang ditulis ulang di berkas DAG.
4. Baris 33 (DAG_ID): Identitas DAG dalam satu konstanta, dipakai dekorator dan pesan alarm, sehingga tidak ada nama yang tersalin berbeda.
5. Baris 35-39 (RAW_DIR, GOLD_DIR, STATE_PATH): Jalur zona data dibaca dari variabel lingkungan dengan nilai bawaan data/ di samping proyek. Pola ini membuat deployment produksi cukup mengatur variabel lingkungan tanpa menyunting berkas DAG.

Dekorator @dag dan Argumennya #

Dekorator @dag mengubah fungsi biasa menjadi pabrik DAG. Seluruh kebijakan orchestration — jadwal, kebijakan catchup, retry bawaan — dinyatakan sebagai argumen dekorator:

@dag(
    dag_id=DAG_ID,
    description="ETL harian pesanan TokoKita: ekstrak, bersihkan, agregasi, muat",
    schedule="30 2 * * *",  # setiap hari 02:30 WIB
    start_date=pendulum.datetime(2026, 11, 1, tz="Asia/Jakarta"),
    catchup=True,  # run yang terlewat sejak start_date diproses ulang
    max_active_runs=1,  # state.json tunggal: jangan menimpanya secara paralel
    default_args={
        "owner": "tim-data-tokokita",
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "execution_timeout": timedelta(minutes=15),
    },
    tags=["tokokita", "etl", "pesanan"],
    doc_md=__doc__,
)
def etl_pesanan_tokokita():
    """Lima task; setiap task membungkus fungsi murni di src/."""

Mari kita telaah argumen-argumen pentingnya:

1. Baris 45 (schedule="30 2 * * *" ): Ekspresi cron — setiap hari pukul 02:30. Airflow 3 menerima cron string, timetables, None (manual), atau preset seperti "@daily". Makna interval datanya dibahas tuntas pada Bagian 5.
2. Baris 46 (start_date): Tanggal pertama periode data. Dibuat dengan pendulum.datetime(..., tz="Asia/Jakarta") agar seluruh perhitungan jadwal mengikuti zona waktu toko, bukan zona server.
3. Baris 47 (catchup=True): Scheduler diperbolehkan membuat run untuk setiap interval yang terlewat sejak start_date — pintu masuk backfill pada Bagian 7.
4. Baris 48 (max_active_runs=1): Hanya satu DAG run aktif pada satu waktu. State kita satu berkas state.json; dua run yang menulis bersamaan akan saling menimpa, jadi serialisasi ini melindungi state.
5. Baris 49-54 (default_args): Kamus argumen bawaan untuk semua task: pemilik, dua percobaan ulang dengan jeda lima menit, dan batas waktu 15 menit. Bagian 6 membedah perilakunya dan menunjukkan override per task.
6. Baris 55-56 (tags dan doc_md): Tag memudahkan penyaringan di UI; doc_md=__doc__ menautkan docstring modul sebagai dokumentasi DAG yang terbaca dari UI.

Task Pertama: ekstrak #

Di dalam fungsi pabrik, setiap task didefinisikan sebagai fungsi dalam berdekorator @task:

    @task
    def ekstrak(ds=None) -> int:
        """Baca partisi pesanan untuk tanggal ds; gagal bila berkas hilang."""
        hasil = etl.baca_pesanan(RAW_DIR, ds)
        print(f"[ekstrak] {ds}: {hasil.jumlah_baris} baris dari {hasil.berkas}")
        return hasil.jumlah_baris

Tiga hal penting pada potongan ini. Pertama, parameter ds=None: Airflow menyuntikkan nilai konteks ke parameter task yang namanya cocok dengan kunci konteks — ds berisi logical date run dalam format YYYY-MM-DD, yaitu persis label partisi yang mau kita proses (ekuivalen dengan templating {{ ds }}). Kedua, isi task satu panggilan fungsi murni etl.baca_pesanan plus satu baris print yang akan tampil di log task. Ketiga, nilai kembalian (jumlah_baris) otomatis tersimpan sebagai XCom sehingga task lain bisa membacanya.

Fungsi baca_pesanan sendiri tinggal di src/tasks.py dan bertanggung jawab penuh atas validasi berkas:

def baca_pesanan(raw_dir: str | Path, tanggal: str) -> HasilEkstraksi:
    """Baca partisi pesanan_YYYY-MM-DD.csv dari zona raw apa adanya.

    Galat yang mungkin terjadi: format tanggal salah (ValueError), partisi
    belum ada (FileNotFoundError), atau kolom wajib hilang (ValueError).
    """
    try:
        date.fromisoformat(tanggal)
    except ValueError as e:
        raise ValueError(f"tanggal harus format YYYY-MM-DD: {tanggal!r}") from e
    berkas = Path(raw_dir) / f"pesanan_{tanggal}.csv"
    if not berkas.exists():
        raise FileNotFoundError(f"partisi tidak ditemukan: {berkas}")
    with berkas.open(newline="", encoding="utf-8") as f:
        pembaca = csv.DictReader(f)
        kolom = set(pembaca.fieldnames or ())
        hilang = [k for k in KOLOM_WAJIB if k not in kolom]
        if hilang:
            raise ValueError(f"kolom wajib hilang pada {berkas}: {', '.join(hilang)}")
        baris = tuple(dict(r) for r in pembaca)
    return HasilEkstraksi(
        tanggal=tanggal, berkas=str(berkas), jumlah_baris=len(baris), baris=baris
    )

Jenis galat yang dibangkitkan pada baris 78, 81, dan 87 bukan detail kecil: jenis galat menentukan apakah retry pada Bagian 6 berguna atau hanya membuang waktu.

Instansiasi dan Penemuan DAG #

Dekorator tidak menjalankan apa pun sampai fungsinya dipanggil:

etl_pesanan_tokokita()  # instansiasi agar scheduler menemukan DAG ini

Tanpa baris ini, berkas hanya berisi definisi fungsi dan scheduler tidak menemukan DAG apa pun — kesalahan klasik saat pertama kali menulis TaskFlow. Setelah instansiasi, scheduler yang mengurai folder dags/ akan mendaftarkan etl_pesanan_tokokita sehingga perintah airflow dags list di environment sungguhan menampilkannya bersama jadwal dan pemiliknya; kita tidak menjalankan perintah itu di sini karena Airflow tidak terpasang, jadi perilakunya cukup dipahami sebagai prosa.

Verifikasi Sintaks #

Yang bisa kita jalankan tanpa Airflow adalah verifikasi sintaks. py_compile mengompilasi sumber menjadi bytecode tanpa mengeksekusi satu baris pun, sehingga from airflow.sdk import ... aman walau modulnya tidak ada:

python3 -m py_compile dags/etl_pesanan.py && echo "py_compile OK: dags/etl_pesanan.py"

Output yang Dihasilkan:

py_compile OK: dags/etl_pesanan.py

Tanpa keluaran galat berarti seluruh berkas lolos kompilasi. Verifikasi struktural yang lebih dalam — apakah lima task, argumen schedule, catchup, retries, dan wiring dependency benar-benar ada — dipegang oleh tests/test_dag_structure.py yang membedah berkas dengan modul ast, dan dibahas pada Bagian 4 serta Bagian 9. Bacaan lanjut: tutorial TaskFlow resmi yang memakai gaya import yang sama dengan proyek ini, serta dokumentasi Airflow Task SDK .