Pada proyek Proyek: Data Warehouse Penjualan, kita membangun warehouse yang memuat dimensi dan fact setiap malam. Pertanyaannya kini berpindah: siapa yang menjalankan pekerjaan itu setiap malam tanpa diawasi, apa yang terjadi bila satu tahap gagal pukul tiga pagi, dan bagaimana memproses ulang tiga hari yang terlewat saat mesin scheduler mati? Itu bukan pekerjaan pipeline, melainkan pekerjaan orchestration. Proyek ini memindahkan tanggung jawab eksekusi harian TokoKita ke Apache Airflow: satu berkas DAG mendeklarasikan urutan task, jadwal, retry, dan cara memantau setiap eksekusi.
Pipeline yang diorkestrasi adalah versi sederhana dari Proyek: Pipeline ETL Terjadwal: satu partisi pesanan harian dibaca dari zona data/raw, dibersihkan, diagregasi per kota, ditulis ke data/gold, lalu watermark di data/state/state.json dimajukan. Seluruh logika ETL hidup di src/ sebagai fungsi Python murni tanpa framework; berkas dags/etl_pesanan.py hanya membungkus fungsi-fungsi tersebut menjadi task Airflow.
Arsitektur Singkat #
Satu DAG run memproses satu tanggal partisi melalui lima task. Setiap task adalah pembungkus tipis di atas satu fungsi murni:
| Task | Fungsi murni di src/ | Berkas yang dihasilkan atau dibaca |
|---|---|---|
ekstrak | tasks.baca_pesanan | membaca data/raw/pesanan_YYYY-MM-DD.csv |
transform | tasks.bersihkan_pesanan, tasks.agregasi_harian | ringkasan ringan lewat XCom |
muat | tasks.muat_ringkasan, state.catat_sukses | data/gold/summary_harian_YYYY-MM-DD.csv, data/state/state.json |
laporan | tasks.laporan_dari_state | satu baris laporan di log task |
alarm_kegagalan | — (stub print) | baris peringatan bila task hulu gagal |
Pemisahan src/ dari dags/ adalah keputusan arsitektur utama proyek ini: logika ETL dapat diuji dengan pytest dalam hitungan detik tanpa memasang Airflow, sedangkan berkas DAG hanya mengatur urutan, jadwal, retry, dan pemantauan.
Kontrak Verifikasi yang Jujur #
Apache Airflow tidak terpasang di lingkungan penulisan proyek ini karena Airflow adalah tool berat dengan banyak dependensi. Oleh karena itu proyek ini membedakan dengan tegas antara yang benar-benar dijalankan dan yang dijelaskan:
| Artefak | Cara diverifikasi |
|---|---|
dags/etl_pesanan.py | python3 -m py_compile untuk sintaks, plus tests/test_dag_structure.py yang mengurai berkas dengan modul ast |
src/tasks.py, src/state.py | diuji nyata dengan pytest (ekstraksi, pembersihan, agregasi, watermark, simulasi backfill) |
| Perilaku scheduler, UI, retry Airflow | dijelaskan dalam prosa dan ditandai sebagai perilaku yang diharapkan, tanpa keluaran karangan |
Tahapan Pengerjaan #
Proyek dikerjakan dalam sembilan bagian berurutan. Setiap bagian menghasilkan berkas yang dirujuk bagian berikutnya:
| Bagian | Fokus |
|---|---|
| 1 | Masalah orkestrasi, istilah inti Airflow, dan lingkup proyek |
| 2 | Struktur direktori, pemisahan logika dari orchestrator, dan konfigurasi |
| 3 | Anatomi DAG dengan TaskFlow API Airflow 3 |
| 4 | Dependency antar task: XComArg, operator >>, dan trigger rule |
| 5 | Scheduling: cron, zona waktu, data interval, catchup |
| 6 | Retry, retry delay, dan execution timeout per task |
| 7 | Backfill: semantik catchup dan simulasi lokal memutar ulang tiga tanggal |
| 8 | Pemantauan eksekusi: UI, state watermark, dan alarm |
| 9 | Strategi pengujian dan penutup proyek |
Mulailah dari Bagian 1 karena setiap bagian dibangun di atas keputusan bagian sebelumnya. Seluruh berkas sumber tersedia pada bagian Berkas Sumber di bawah halaman ini, python3 -m pytest tests/ -q yang dijalankan dari direktori proyek harus selalu hijau, dan dokumentasi Apache Airflow
menjadi rujukan pendamping untuk perilaku yang tidak dijalankan di lingkungan penulisan.
Petakan masalah orkestrasi harian TokoKita, istilah inti Airflow, dan lingkup proyek sebelum menulis berkas DAG.
Buka → # Bagian 2: Menyiapkan EnvironmentSusun struktur direktori, pisahkan logika ETL dari orchestrator, dan siapkan konfigurasi environment Airflow beserta jalur verifikasi lokal.
Buka → # Bagian 3: Membuat DAGTulis anatomi berkas DAG dengan TaskFlow API Airflow 3, dari import airflow.sdk sampai instansiasi dan verifikasi sintaks.
Buka → # Bagian 4: Mengatur Dependency TaskRangkai lima task dengan XComArg dan operator >>, pahami trigger rule, dan kunci wiring lewat pengujian ast.
Buka → # Bagian 5: Mengatur SchedulingPahami cron 30 2 * * *, zona waktu, data interval, dan ds, lalu simulasikan interval harian secara lokal.
Buka → # Bagian 6: Menerapkan Retry dan TimeoutAtur retries, retry_delay, dan execution_timeout di default_args, pelajari override per task, dan buktikan prasyarat idempotensi lewat eksekusi nyata.
Buka → # Bagian 7: Menjalankan BackfillPahami semantik catchup, jelaskan jalur backfill pada Airflow 3, lalu buktikan pemrosesan ulang tiga tanggal lewat simulasi lokal yang nyata.
Buka → # Bagian 8: Memantau EksekusiBaca grid eksekusi di UI Airflow secara konseptual, dan jadikan state watermark sebagai sumber pemantauan yang bisa dibuktikan tanpa Airflow.
Buka → # Bagian 9: Menguji WorkflowRangkai strategi pengujian: unit test fungsi murni, uji struktur DAG via ast, simulasi backfill, lalu tutup proyek dengan suite pytest hijau.
Buka →