Bagian 5: Membangun Pipeline

Satu perintah harus cukup untuk membangun seluruh sistem dari dua berkas mentah sampai muatan dashboard — itulah standar yang kita tetapkan pada bagian ini. Orkestrator src/pipeline.py merangkai enam tahap: kumpulkan, bersihkan, model, marts, validasi, ekspor. Dua prinsip dari chapter Idempotency dan Retries diterapkan penuh: eksekusi berulang pada data sumber yang sama harus menghasilkan keluaran identik, dan basis data dibangun ulang dari nol agar tidak ada sisa tabel usang dari eksekusi lama. Materi perancangan alurnya dibahas pada chapter Merancang Data Pipeline dan Idempotent Transformation.

Menjalankan Model dari Python #

src/models_runner.py adalah jembatan antara berkas model SQL dan DuckDB. Fungsinya hanya dua puluh baris:

def jalankan_model(con, nama_model: str, direktori_models) -> int:
    """Materialisasi satu model SQL menjadi tabel dengan nama yang sama."""
    sql = (direktori_models / f"{nama_model}.sql").read_text(encoding="utf-8")
    con.execute(f'CREATE OR REPLACE TABLE "{nama_model}" AS\n{sql}')
    return con.execute(f'SELECT COUNT(*) FROM "{nama_model}"').fetchone()[0]

Dua baris tersebut menggantikan sebagian pekerjaan dbt. Pemetaan konsepnya:

CREATE OR REPLACE adalah kunci idempotensi: model boleh dijalankan berapa pun kali, tabel selalu dibuat ulang dengan isi yang sama. Berbeda dengan CREATE TABLE polos yang galat ketika tabel sudah ada, atau INSERT INTO yang menduplikasi baris pada eksekusi kedua.

Orkestrator Enam Tahap #

Inti src/pipeline.py adalah fungsi jalankan yang menerima akar proyek sebagai parameter — parameter inilah yang membuat uji end-to-end dapat menjalankan pipeline pada salinan direktori tanpa menyentuh bundle:

def jalankan(akar=AKAR_PROYEK, tampilkan=True) -> dict:
    """Jalankan enam tahap pipeline dan kembalikan ringkasan eksekusi."""
    akar = Path(akar)
    direktori_data = akar / "data"
    direktori_raw = direktori_data / "raw"
    direktori_staging = direktori_data / "staging"
    direktori_gold = direktori_data / "gold"
    direktori_dashboard = direktori_data / "dashboard"
    direktori_models = akar / "models"
    berkas_db = direktori_data / "analytics.duckdb"

    # Tahap 1: kumpulkan dari dua sumber.
    df_pesanan = collect.baca_pesanan(direktori_raw / "pesanan.csv")
    events, karantina_baca = collect.baca_events(direktori_raw / "events.jsonl")

    # Tahap 2: bersihkan dan simpan checkpoint staging.
    hasil_pesanan = clean.bersihkan_pesanan(df_pesanan)
    hasil_events = clean.bersihkan_events(events, karantina_baca)
    ringkasan_bersih = clean.simpan_staging(direktori_staging, hasil_pesanan, hasil_events)

    # Basis data dibangun ulang dari nol agar idempoten.
    if berkas_db.exists():
        berkas_db.unlink()
    con = duckdb.connect(str(berkas_db))

Mari kita telaah baris-baris kode di atas (nomor baris merujuk berkas src/pipeline.py):

1. Baris 20 (akar=AKAR_PROYEK): akar proyek adalah parameter, bukan konstanta global yang dikunci; uji end-to-end memakai titik ini untuk mengarahkan pipeline ke direktori sementara.
2. Baris 31-33 (tahap 1): pengumpulan mengembalikan data mentah persis seperti Bagian 2.
3. Baris 35-38 (tahap 2): hasil pembersihan tidak hanya dipakai di memori — simpan_staging menuliskannya sebagai checkpoint sehingga tahap model bisa dikerjakan ulang tanpa membersihkan lagi.
4. Baris 40-42 (unlink): berkas database lama dihapus sebelum koneksi dibuka. Tanpa langkah ini, tabel yang dihapus dari models/ akan tetap hidup di database lama dan membingungkan pembaca.
5. try/finally pada berkas aslinya: koneksi DuckDB selalu ditutup meskipun tahap mana pun melempar galat, sehingga eksekusi berikutnya tidak terhalang berkas database yang terkunci.

Tahap 3 sampai 6 berlanjut di dalam koneksi yang sama: models_runner.jalankan_semua untuk staging dan fact, marts.bangun_marts plus marts.ekspor_gold, validate.jalankan_pemeriksaan yang berhenti membatalkan pipeline bila ada pemeriksaan gagal, dan export_dashboard.ekspor sebagai tahap penutup.

Menjalankan Pipeline Lengkap #

Dari akar proyek, satu perintah menjalankan semuanya:

python3 -m src.pipeline

Output yang Dihasilkan:

[1/6] kumpulkan : pesanan.csv 38 baris (12 kolom); events.jsonl 116 baris
[2/6] bersihkan : pesanan 33 bersih, 2 duplikat, 3 karantina; events 112 bersih, 1 duplikat, 3 karantina
[3/6] model     : sumber_pesanan 33, sumber_events 112, stg_pesanan 33, stg_events 112, fct_pesanan 33
[4/6] marts     : mkt_gmv_harian 14, mkt_kategori 4, mkt_funnel 4, mkt_pelanggan 11; 4 berkas gold
[5/6] validasi  : 8/8 pemeriksaan lulus (data/gold/validation_report.json)
[6/6] ekspor    : data/dashboard/data.json (14 titik GMV harian)
selesai: data/analytics.duckdb

Setiap tahap dikerjakan fungsi pada modul yang berbeda sehingga masing-masing bisa diuji sendiri:

Enam baris tersebut adalah riwayat lengkap perjalanan data: 38 baris pesanan mentah menjadi 33 baris fact, 116 baris event menjadi 112 baris staging, lalu 33 baris fact mengerucut menjadi empat tabel analitik berisi 14+4+4+11 baris. Tahap validasi lolos penuh sebelum ekspor dashboard diizinkan — urutan ini bukan kebetulan, melainkan gerbang: dashboard tidak boleh diperbarui dari data yang gagal validasi.

Membuktikan Idempotensi #

Klaim idempoten harus dibuktikan, bukan ditulis. Kita jalankan pipeline dua kali lalu bandingkan sidik digital data.json:

python3 -m src.pipeline > /dev/null && md5sum data/dashboard/data.json
python3 -m src.pipeline > /dev/null && md5sum data/dashboard/data.json

Output yang Dihasilkan:

5c95d1c6b4eaab9e50bd91d0e2554b04  data/dashboard/data.json
5c95d1c6b4eaab9e50bd91d0e2554b04  data/dashboard/data.json

Sidik digital identik karena muatan dashboard tidak memuat waktu eksekusi — seluruh isinya murni turunan data. Uji test_idempoten pada tests/test_pipeline_e2e.py memeriksa hal yang sama secara otomatis: dua eksekusi penuh harus menghasilkan data.json dan validation_report.json yang byte-identik.

Dengan orkestrator yang terbukti idempoten, tahap berikutnya adalah mengisi lapisan marts: tabel-tabel agregat yang dibaca dashboard. Dokumentasi resmi DuckDB untuk pola CREATE OR REPLACE TABLE tersedia di https://duckdb.org/docs/stable/sql/statements/create_table.html .