Bagian 2: Menyiapkan Environment

Sebelum menulis DAG, kita mengunci dua keputusan yang menentukan bagaimana seluruh proyek diverifikasi. Pertama, logika ETL tidak tinggal di dalam berkas DAG, melainkan di paket src/ sebagai fungsi Python murni. Kedua, karena Apache Airflow tidak terpasang di lingkungan penulisan, kita menetapkan sejak awal jalur verifikasi apa yang tersedia: py_compile untuk sintaks DAG, ast untuk strukturnya, dan pytest untuk logika. Bagian ini menyiapkan kedua hal itu beserta konfigurasi environment-nya.

Struktur Direktori #

Pemetaan berikut dihasilkan dari direktori proyek yang sebenarnya setelah seluruh bagian selesai:

Output yang Dihasilkan:

orkestrasi-pipeline-dengan-airflow/
|-- config
|   |-- connections.example.yaml
|   `-- requirements.txt
|-- dags
|   `-- etl_pesanan.py
|-- part-1-memahami-workflow
|   `-- index.md
|-- part-2-menyiapkan-environment
|   `-- index.md
|-- part-3-membuat-dag
|   `-- index.md
|-- part-4-mengatur-dependency-task
|   `-- index.md
|-- part-5-mengatur-scheduling
|   `-- index.md
|-- part-6-menerapkan-retry-dan-timeout
|   `-- index.md
|-- part-7-menjalankan-backfill
|   `-- index.md
|-- part-8-memantau-eksekusi
|   `-- index.md
|-- part-9-menguji-workflow
|   `-- index.md
|-- src
|   |-- __init__.py
|   |-- state.py
|   `-- tasks.py
|-- tests
|   |-- conftest.py
|   |-- test_backfill_sim.py
|   |-- test_dag_structure.py
|   |-- test_state.py
|   `-- test_tasks.py
`-- _index.md

Empat folder pendukung (dags, src, tests, config) berdiri berdampingan dengan sembilan halaman bagian. Folder dags/ adalah satu-satunya yang dikenali scheduler Airflow; src/ diimpor oleh berkas DAG; tests/ dijalankan dengan python3 -m pytest tests/ -q dari direktori proyek; dan config/ berisi dependensi serta contoh koneksi untuk deployment sungguhan.

Memisahkan Logika dari Orchestrator #

Alasan pemisahan src/ dari dags/ bukan gaya, melainkan testability dan portabilitas. Fungsi di src/tasks.py hanya mengenal Path, csv, dan dataclass — tidak ada import Airflow — sehingga bisa diuji dalam 0,3 detik tanpa infrastruktur. Berkas DAG tinggal menjadi lapisan deklarasi yang membungkus fungsi tersebut:

Kepala berkas src/tasks.py menyatakan kontrak ini secara eksplisit:

"""Logika task ETL pesanan harian TokoKita — murni Python tanpa framework.

Setiap fungsi publik di modul ini dibungkus satu task pada
``dags/etl_pesanan.py``, tetapi tidak ada import Airflow di sini. Pemisahan
ini disengaja: logika ETL bisa diuji cepat dengan pytest tanpa memasang
orchestrator, sedangkan DAG menjadi lapisan tipis yang mengatur urutan,
penjadwalan, retry, dan pemantauan.
"""

from __future__ import annotations

import csv
from dataclasses import dataclass
from datetime import date
from pathlib import Path

from . import state as state_etl

KOLOM_WAJIB = ("pesanan_id", "kota", "status", "total")
STATUS_DIKENAL = ("selesai", "diproses", "dikirim")

Konstanta KOLOM_WAJIB dan STATUS_DIKENAL di baris 19-20 adalah kontrak data sekecil apa pun tetap perlu ditulis: kolom yang hilang atau status asing harus gagal keras, bukan lolos diam-diam.

Konfigurasi Environment #

Dua berkas di config/ menyiapkan lingkungan deployment. Yang pertama mendeklarasikan dependensi:

# Dependensi untuk menjalankan proyek orkestrasi pada environment yang
# memiliki Apache Airflow. Kode pada src/ hanya memakai pustaka standar
# Python, jadi daftar di bawah dibutuhkan oleh berkas DAG dan environment
# airflow — bukan oleh pengujian pytest (yang berjalan tanpa Airflow).
#
# Pasang di virtual environment terpisah; Airflow berat dan punya banyak
# dependensi. Ikuti panduan pemasangan resmi untuk versi yang dipakai.

apache-airflow>=3.3,<4          # orchestrator; TaskFlow API dari airflow.sdk
pendulum>=3                     # tanggal dan zona waktu untuk start_date DAG

# Pengujian lokal (tanpa Airflow):
pytest>=8

Yang kedua, config/connections.example.yaml, berisi contoh koneksi dan variabel untuk deployment: bagaimana kredensial warehouse dipisahkan dari kode dan bagaimana jalur zona data bisa disimpan sebagai Airflow Variable. Berkas tersebut tidak dibaca oleh kode proyek — jalur zona data pada berkas DAG dibaca dari variabel lingkungan TOKOKITA_RAW_DIR, TOKOKITA_GOLD_DIR, dan TOKOKITA_STATE_PATH dengan nilai bawaan data/ — melainkan menjadi pola yang siap dipakai bila pipeline berkembang menulis ke database. Pada klaster sungguhan, koneksi didaftarkan lewat UI atau CLI Airflow dan tersimpan di metadata database; penamaan kunci persisnya bergantung versi, jadi periksa --help pada CLI versi yang dipakai.

Cara Airflow Menemukan DAG #

Scheduler Airflow memindai folder yang dikonfigurasi sebagai dags_folder (bawaannya ~/airflow/dags/) dan mengurai setiap berkas Python di dalamnya. Karena folder tersebut yang masuk sys.path, sementara paket src/ kita berada di luar folder DAG, berkas DAG perlu menambahkan akar proyek sendiri:

# 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

Pola tiga baris sys.path ini standar untuk proyek yang logikanya tidak tinggal di dalam folder dags/; pengujian test_bootstrap_sys_path_agar_src_bisa_diimpor menjaganya agar tidak terhapus tanpa sengaja. Alternatif lain di produksi adalah memasang src/ sebagai paket (misalnya pip install -e .) sehingga import bekerja tanpa manipulasi jalur.

Yang Bisa Dijalankan Sekarang #

Tanpa Airflow, tiga perintah tetap bermakna dan semuanya dipakai berulang di bagian-bagian berikut: python3 -m py_compile dags/etl_pesanan.py memverifikasi sintaks berkas DAG tanpa mengeksekusinya, python3 -m pytest tests/ -q menguji seluruh logika di src/, dan python3 -m pytest tests/test_dag_structure.py -q memeriksa struktur DAG lewat ast. Berikut yang pertama:

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

Dengan struktur, konfigurasi, dan jalur verifikasi yang siap, kita sekarang bisa menulis berkas DAG-nya sendiri pada Bagian 3. Konsep orchestrator dan arsitektur Airflow dibahas lebih dalam pada chapter Konsep Workflow Orchestration dan Apache Airflow, dan panduan pemasangan resmi ada di dokumentasi Apache Airflow .