Bagian 1: Merancang Alur ETL

Proyek Proyek: Pipeline Ingestion API mengajarkan kita menarik data dari satu sumber dan menaruhnya di zona pendaratan. TokoKita sekarang melangkah lebih jauh: data yang sudah mendarat harus dibersihkan, diagregasi, dan dimuat ke warehouse analitik secara terjadwal, lalu dilengkapi penjadwal, pencatat kegagalan, dan validasi hasil. Sebelum menulis logika ETL, kita mengunci dulu empat keputusan: zona data apa saja yang dipakai, berkas tinggal di mana, konfigurasi dibaca dari satu sumber mana, dan bagaimana pipeline dipanggil dari mesin. Keputusan-keputusan ini yang menjadi pagar di dalamnya seluruh bagian berikutnya bekerja.

Peta Zona Data #

Satu eksekusi pipeline memindahkan data melalui zona-zona dengan peran yang berbeda. Pemisahan zona membuat setiap keputusan bisa diaudit dari lokasinya:

Zona atau berkas	Isi	Ditulis oleh
`data/raw/`	Partisi CSV harian dari sistem kasir, apa adanya	Sumber (bukan pipeline)
`data/warehouse.duckdb`	Tabel `fact_pesanan` dan `summary_harian`	Tahap load
`data/gold/`	Ekspor CSV summary per tanggal, siap dibaca konsumen	Tahap load
`data/quarantine/`	Baris yang ditolak aturan pembersihan beserta alasannya	Tahap transform
`data/state/`	Watermark per tanggal, manifest per eksekusi, log alert	Semua tahap

Zona raw tidak pernah diubah oleh pipeline; state mencatat keadaan sehingga eksekusi ulang bisa dilewati atau dipaksakan; quarantine menjaga baris yang ditolak tetap bisa diperiksa. Pola raw-curated seperti ini dibahas pada chapter Zona Data Raw dan Curated, dan perbedaan ETL dari ELT di Pengenalan ETL dan ELT.

Struktur Direktori #

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

Output yang Dihasilkan:

text
pipeline-etl-terjadwal/
|-- config/
|   |-- alert.example.yaml
|   |-- cron
|   |-- etl-nightly.service
|   |-- etl-nightly.timer
|   `-- pipeline.yaml
|-- data/
|   |-- gold/
|   |   |-- summary_harian_2026-11-02.csv
|   |   |-- summary_harian_2026-11-03.csv
|   |   `-- summary_harian_2026-11-04.csv
|   |-- quarantine/
|   |   |-- pesanan_2026-11-03.csv
|   |   `-- pesanan_2026-11-04.csv
|   |-- raw/
|   |   |-- pesanan_2026-11-02.csv
|   |   |-- pesanan_2026-11-03.csv
|   |   `-- pesanan_2026-11-04.csv
|   |-- state/
|   |   |-- runs/
|   |   |   |-- run-2026-11-02-075011.json
|   |   |   |-- run-2026-11-02-075015.json
|   |   |   |-- run-2026-11-02-075017.json
|   |   |   |-- run-2026-11-03-075012.json
|   |   |   |-- run-2026-11-04-075014.json
|   |   |   |-- run-2026-11-05-075018.json
|   |   |   `-- run-2026-11-05-075019.json
|   |   |-- alerts.log
|   |   `-- state.json
|   `-- warehouse.duckdb
|-- part-1-merancang-alur-etl/
|   `-- index.md
|-- part-2-mengambil-data-sumber/
|   `-- index.md
|-- part-3-membersihkan-dan-mentransformasi/
|   `-- index.md
|-- part-4-memuat-data-ke-target/
|   `-- index.md
|-- part-5-membuat-pipeline-idempotent/
|   `-- index.md
|-- part-6-menjadwalkan-pipeline/
|   `-- index.md
|-- part-7-menangani-kegagalan/
|   `-- index.md
|-- part-8-menambahkan-monitoring/
|   `-- index.md
|-- part-9-memvalidasi-hasil/
|   `-- index.md
|-- src/
|   |-- sql/
|   |   |-- reconcile.sql
|   |   |-- schema.sql
|   |   `-- summary_harian.sql
|   |-- __init__.py
|   |-- extract.py
|   |-- failures.py
|   |-- idempotency.py
|   |-- load.py
|   |-- monitoring.py
|   |-- pipeline.py
|   |-- scheduler.py
|   |-- transform.py
|   `-- validate.py
|-- tests/
|   |-- conftest.py
|   |-- test_extract.py
|   |-- test_failures.py
|   |-- test_idempotency.py
|   |-- test_load.py
|   |-- test_monitoring.py
|   |-- test_pipeline_e2e.py
|   |-- test_scheduler.py
|   |-- test_transform.py
|   `-- test_validate.py
`-- _index.md

Tiga berkas pada config/ tidak memiliki akhiran kode sehingga tidak tampil sebagai blok sintaks, tetapi tetap dapat diunduh dari bagian Berkas Sumber; isinya dibahas pada Bagian 6 untuk penjadwalan dan Bagian 7 untuk alert. Model SQL berada di src/sql/ agar skema warehouse dapat dibaca tanpa membuka kode Python. Pengujian di tests/ dijalankan dari direktori proyek dengan python3 -m pytest tests/ -q dan harus tetap hijau di setiap bagian.

Konfigurasi sebagai Satu Sumber #

Seluruh jalur dan parameter tinggal di config/pipeline.yaml:

yaml
config/pipeline.yaml (dipangkas)
nama: etl-pesanan-tokokita

# Zona dan target
raw_dir: data/raw              # partisi harian pesanan_YYYY-MM-DD.csv
warehouse: data/warehouse.duckdb
gold_dir: data/gold            # ekspor CSV summary per tanggal
state_dir: data/state          # state.json, runs/*.json, alerts.log
quarantine_dir: data/quarantine  # baris gagal validasi (dead-letter)

# Penjadwalan (dipakai src/scheduler.py, bukan oleh pipeline satu tanggal)
schedule:
  cron: "30 2 * * *"           # setiap hari 02:30
  lookback_days: 7             # jendela mundur yang diperiksa scheduler
  max_attempts: 3              # percobaan maksimum sebelum tanggal diblok

load_config() di src/pipeline.py memeriksa kunci wajib sebelum pipeline berjalan; konfigurasi yang hilang atau rusak berhenti di exit code 2. Kunci penjadwalan dibaca src/scheduler.py, bukan oleh eksekusi satu tanggal, sehingga mengubah jam cron tidak menyentuh jalur ETL sama sekali.

Kontrak CLI #

Pipeline dipanggil sebagai modul paket, bukan skrip lepas: python3 -m src.pipeline. Bentuk -m menjalankan src/ sebagai paket sehingga impor antarmodul konsisten antara CLI, scheduler, dan pengujian. Tiga argumennya:

bash

python3 -m src.pipeline --help

Output yang Dihasilkan:

text
usage: src.pipeline [-h] [--date DATE] [--config CONFIG] [--force]

Pipeline ETL harian pesanan TokoKita (lihat config/pipeline.yaml).

options:
  -h, --help       show this help message and exit
  --date DATE      tanggal partisi: YYYY-MM-DD, 'yesterday', atau 'today'
  --config CONFIG  berkas konfigurasi
  --force          proses ulang meskipun tanggal sudah sukses sebelumnya

Nilai --date diterjemahkan parse_run_date: kata yesterday menjadi tanggal sebelumnya, today menjadi hari ini, dan bentuk lain harus ISO YYYY-MM-DD. Nilai yang tidak dikenal berhenti di exit code 1 sebelum menyentuh data. Tiga modul lain juga punya titik masuk sendiri: python3 -m src.scheduler (daemon penjadwal), python3 -m src.monitoring (laporan status), dan python3 -m pytest untuk pengujian.

Dengan zona, direktori, konfigurasi, dan CLI yang sudah dikunci, kita siap menulis tahap pertama yang benar-benar menyentuh data: membaca partisi dari zona raw pada Bagian 2. Bacaan lanjut: argparse pada dokumentasi Python dan DuckDB .