Bagian 5: Membangun Pipeline Pemuatan

Loader pada Bagian 3 dan 4 membuktikan mekanismenya bekerja, tetapi menyusun pemanggilan fungsi secara manual setiap kali pemuatan berjalan adalah resep kelalaian. Pada bagian ini kita memindahkan transformasi ke berkas SQL murni pada folder models/, lalu src/pipeline.py merangkai seluruh tahap menjadi satu perintah. Bentuk model SQL mengikuti konvensi dbt yang kita pelajari pada chapter dbt dan Model Staging, Intermediate, dan Marts: satu berkas, satu SELECT, nama berkas menjadi nama tabel hasil.

Pemetaan Konsep dbt #

Prinsip yang diadopsi sama: transformasi dinyatakan sebagai SQL deklaratif yang dapat dibaca dari bawah ke atas, sementara orkestrasi hanya mengatur urutan eksekusi.

Mengapa repot-repot memisahkan staging? Karena tipe data CSV sering tidak sesuai harapan: DuckDB bisa menebak kolom baris sebagai BIGINT atau kolom tanggal sebagai teks, dan tebakan yang berubah-ubah membuat model tidak stabil. Tahap staging menetapkan tipe sekali di satu tempat, sehingga seluruh model di hilir bekerja atas tipe yang pasti. Prinsip layer staging ini persis pembahasan chapter Model Staging, Intermediate, dan Marts.

Model SQL Dimensi dan Fact #

Model dimensi menghitung surrogate key dan kedua hash langsung di SQL. Berikut models/dim_pelanggan.sql utuh:

-- Model dim_pelanggan: pemuatan awal SCD Type 2.
-- Seluruh pelanggan dimuat sebagai satu versi yang berlaku sejak 2025-01-01
-- (awal periode data sumber). Perubahan atribut berikutnya ditangani oleh
-- src/scd2.py, peran yang pada dbt dipegang oleh snapshot.
SELECT
    row_number() OVER (ORDER BY s.pelanggan_id)::INTEGER AS pelanggan_sk,
    md5(s.pelanggan_id) AS nk_hash,
    md5(concat_ws('|', s.nama, s.email, s.kota, s.segmen)) AS attr_hash,
    s.pelanggan_id,
    s.nama,
    s.email,
    s.kota,
    s.segmen,
    DATE '2025-01-01' AS valid_from,
    DATE '9999-12-31' AS valid_to,
    TRUE AS is_current
FROM stg_pelanggan AS s

Perhatikan bahwa md5(concat_ws('|', ...)) memakai skema penggabungan yang sama dengan hash_atribut() pada Bagian 3, sehingga loader manual dan model SQL menghasilkan hash identik untuk data yang sama; kesamaan ini diuji eksplisit pada tests/test_pipeline_e2e.py. Model fact kemudian memuat seluruh logika resolusi Bagian 4 dalam bentuk deklaratif:

SELECT
    s.pesanan_id,
    s.baris,
    t.tanggal_sk,
    p.pelanggan_sk,
    r.produk_sk,
    k.toko_sk,
    s.jumlah,
    s.harga_satuan,
    s.diskon,
    s.jumlah * s.harga_satuan - s.diskon AS total,
    s.status
FROM stg_pesanan AS s
JOIN dim_tanggal AS t
  ON t.tanggal = s.tanggal
JOIN dim_pelanggan AS p
  ON p.pelanggan_id = s.pelanggan_id
 AND s.tanggal BETWEEN p.valid_from AND p.valid_to
JOIN dim_produk AS r
  ON r.produk_id = s.produk_id
JOIN dim_toko AS k
  ON k.toko_id = s.toko_id

Tidak ada INSERT, tidak ada DROP: materialisasi ditangani oleh runner. Kolom komentar pada berkas aslinya menjelaskan peran tiap join, termasuk join BETWEEN yang menjadi kunci SCD Type 2.

Orkestrator pipeline #

src/pipeline.py menjalankan lima tahap berurutan. Tahap staging membaca CSV dengan tipe eksplisit:

    for tabel, berkas, kolom in SPESIFIKASI_STAGING:
        jalur = direktori_data / berkas
        daftar_kolom = ", ".join(
            f"{nama}::{tipe} AS {nama}" for nama, tipe in kolom.items()
        )
        con.execute(f'DROP TABLE IF EXISTS "{tabel}"')
        con.execute(
            f"CREATE TABLE {tabel} AS "
            f"SELECT {daftar_kolom} FROM read_csv('{jalur}', header = true)"
        )
        ringkasan[tabel] = con.execute(f"SELECT COUNT(*) FROM {tabel}").fetchone()[0]

Tahap berikutnya mematerialisasi model satu per satu:

def _materialisasi_model(con, nama_model: str, direktori_models: Path):
    """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}')

Urutan lengkapnya dilihat paling jelas sebagai diagram:

Satu urutan yang perlu digarisbawahi: fact_pesanan sengaja dimaterialisasi setelah SCD Type 2 diterapkan. Model fact memilih versi dimensi yang sah pada tanggal transaksi, sehingga riwayat pelanggan harus sudah tersedia lebih dulu; bila urutan ini dibalik, pesanan setelah tanggal perpindahan akan menunjuk versi lama.

Menjalankan Pipeline #

Seluruh tahap terjadi dengan satu perintah dari akar proyek:

python3 -m src.pipeline

Output yang Dihasilkan:

[1/5] staging      : stg_pelanggan=10, stg_pelanggan_update=2, stg_produk=12, stg_toko=5, stg_pesanan=32
[2/5] model dimensi: dim_tanggal, dim_toko, dim_produk, dim_pelanggan dimuat
[3/5] SCD type 2   : 2 perubahan diterapkan, 0 dilewati
[4/5] model fact   : 32 baris fact_pesanan
[5/5] kualitas data: 5/5 pemeriksaan lulus
                     laporan: data/dq_reports/dq_report.json
gudang data siap  : data/warehouse.duckdb

Angka pada tahap pertama adalah jumlah baris per tabel staging; stg_pelanggan_update=2 menandakan ada dua perubahan atribut pelanggan menunggu. Menjalankan perintah yang sama berulang kali menghasilkan isi gudang yang sama karena seluruh tahap membangun ulang dari sumber, perilaku yang diuji pada tests/test_pipeline_e2e.py. Konsep materialisasi table dan perintah dbt run yang kita tiru polanya terdokumentasi resmi di https://docs.getdbt.com/docs/build/models dan https://docs.getdbt.com/reference/dbt-commands .

Pipeline berjalan, tetapi dua perubahan pada tahap ketiga belum kita bedah: bagaimana satu pelanggan dapat memiliki dua versi tanpa merusak analisis historis adalah pokok bahasan Bagian 6.