Proyek: Pipeline ETL Terjadwal

Pada proyek Proyek: Pipeline Ingestion API, kita mengambil data dari sebuah API dan menaruhnya di zona pendaratan. TokoKita kini punya kebutuhan yang lebih terstruktur: setiap malam, berkas pesanan harian harus dibersihkan, diagregasi, dan dimuat ke sebuah warehouse analitik agar tim bisnis bisa membaca laporan tanpa menunggu. Pekerjaan itu bukan sekali jalan, melainkan berulang setiap hari; eksekusi ulang harus aman, dan setiap kegagalan harus dilaporkan lewat exit code yang bisa dibaca mesin. Proyek ini merangkai semuanya menjadi satu pipeline ETL terjadwal yang utuh.

Kita memakai DuckDB sebagai warehouse berbentuk satu berkas data/warehouse.duckdb, pandas untuk pembersihan baris, dan SQL untuk pembebanan serta agregasi. Seluruh kode berada di src/, pengujian di tests/, konfigurasi di config/, dan data di data/ yang terbagi menjadi zona raw, gold, state, dan quarantine. Setiap bagian proyek merujuk ke berkas yang sama, sehingga kondisi proyek tetap utuh di setiap tahap.

Arsitektur Singkat #

Satu eksekusi pipeline memproses satu tanggal partisi, dari berkas mentah sampai berkas gold yang siap dibaca:

flowchart LR A["data/raw partisi CSV harian"] --> B["extract.py read_partition"] B --> C["transform.py clean_orders"] C -->|"baris valid"| D["load.py MERGE + summary"] C -->|"baris ditolak"| Q["data/quarantine dead-letter"] D --> W["data/warehouse.duckdb fact + summary"] D --> G["data/gold CSV harian"] D --> S["data/state watermark + manifest"]

Baris yang gagal aturan pembersihan tidak dibuang diam-diam; baris tersebut ditulis ke zona karantina beserta alasan dan nomor baris asalnya. Warehouse menyimpan dua tabel: fact_pesanan dengan grain satu pesanan, dan summary_harian yang diagregasi per kota per tanggal. Zona state mencatat watermark per tanggal sehingga eksekusi ulang dapat dilewati atau dipaksakan dengan aman.

Perilaku yang Diharapkan #

Contoh sesi berikut memproses partisi 2026-11-03 yang sengaja memuat spasi berlebih, harga berformat Indonesia, status huruf besar, satu duplikat, dan dua baris yang tidak lengkap:

bash

python3 -m src.pipeline --date 2026-11-03

Output yang Dihasilkan:

text
[extract] pesanan_2026-11-03.csv: 10 baris, 927 bytes
[transform] 10 baris -> 7 fact, 2 karantina, 1 duplikat
[karantina] 2 baris ditulis ke data/quarantine/pesanan_2026-11-03.csv
[load] MERGE fact: 7 baris; summary 2026-11-03: 4 baris
[validate] OK  reconcile kota: 7 baris di 4 kota (total sumber 7 = total target 7)
[validate] OK  agregat harian: fact: 7 baris, Rp3,613,000.00 bruto, Rp3,613,000.00 bersih; summary: 7 baris, Rp3,613,000.00 bruto, Rp3,613,000.00 bersih
[validate] OK  gold csv: 4 baris gold vs 4 baris summary
[gold] data/gold/summary_harian_2026-11-03.csv
[selesai] run-2026-11-03-075012 status=success durasi=51ms exit=0

Sepuluh baris mentah menjadi tujuh baris fact, dua baris karantina, dan satu duplikat yang dibuang. Setiap tahap mencetak satu baris log berprefiks tahapannya, dan tiga pemeriksaan validasi harus lulus sebelum transaksi di-commit.

Kontrak Exit Code #

Pipeline adalah program yang dipanggil mesin, bukan hanya manusia. Karena itu hasil eksekusinya dilaporkan lewat exit code yang tetap:

Exit code	Arti
`0`	Tanggal berhasil diproses, atau dilewati karena sudah sukses
`1`	Argumen CLI tidak valid
`2`	Konfigurasi hilang atau rusak
`3`	Ekstraksi gagal, misalnya partisi tidak ada
`4`	Transformasi gagal
`5`	Pemuatan ke warehouse gagal
`6`	Validasi hasil gagal

Tahapan Pengerjaan #

Proyek dikerjakan dalam sembilan bagian berurutan. Setiap bagian menghasilkan kode yang benar-benar berjalan dan diuji:

Bagian	Fokus
1	Arsitektur, struktur direktori, konfigurasi, dan kontrak CLI
2	Ekstraksi partisi harian dari zona raw
3	Pembersihan baris dan pembentukan fact
4	Pemuatan MERGE ke DuckDB dan ekspor gold
5	Watermark state JSON dan bukti eksekusi ulang identik
6	Penjadwalan: cron, due runs, loop Python, systemd
7	Exit code, karantina dead-letter, dan alert stub
8	Manifest eksekusi, checksum, dan laporan status
9	Rekonsiliasi sumber-target dan penutupan proyek

Mulailah dari Bagian 1 karena setiap bagian dibangun di atas bagian sebelumnya. Seluruh berkas sumber tersedia pada bagian Berkas Sumber di bawah halaman ini dan setiap bagian, dan python3 -m pytest tests/ -q dari direktori proyek harus selalu hijau.

# Bagian 1: Merancang Alur ETL

Petakan zona data, struktur direktori, konfigurasi, dan kontrak CLI sebelum menulis satu baris kode ETL.

Buka → # Bagian 2: Mengambil Data Sumber

Baca partisi harian CSV dari zona raw sebagai teks mentah dan tolak partisi dengan skema yang salah.

Buka → # Bagian 3: Membersihkan dan Mentransformasi

Bersihkan baris pesanan di pandas: pangkas spasi, karantina baris tidak valid, dan tangani duplikat order_id.

Buka → # Bagian 4: Memuat Data ke Target

Muat fact ke DuckDB dengan MERGE, bangun summary harian dari SQL, lalu ekspor hasil ke zona gold.

Buka → # Bagian 5: Membuat Pipeline Idempotent

Catat watermark per tanggal di state JSON dan buktikan bahwa eksekusi ulang menghasilkan warehouse identik.

Buka → # Bagian 6: Menjadwalkan Pipeline

Jalankan pipeline setiap malam: parsing cron, due runs, loop penjadwal Python, serta konfigurasi cron dan systemd.

Buka → # Bagian 7: Menangani Kegagalan

Tangani kegagalan dengan kontrak exit code, berkas karantina dead-letter, dan alert stub tanpa jaringan.

Buka → # Bagian 8: Menambahkan Monitoring

Catat manifest eksekusi berisi jumlah baris, durasi, dan checksum, lalu susun laporan status.

Buka → # Bagian 9: Memvalidasi Hasil

Rekonsiliasi baris sumber dengan warehouse, cek agregat fact-summary-gold, dan tutup proyek dengan pytest hijau.

Buka →