Warehouse kita adalah satu berkas DuckDB di data/warehouse.duckdb yang menyimpan dua tabel: fact_pesanan bergrain satu pesanan dan summary_harian bergrain satu kota per tanggal. Pemuatan harus menjawab dua pertanyaan sekaligus: bagaimana memasukkan baris baru tanpa menduplikasi baris lama, dan bagaimana tabel agregat tetap konsisten dengan fact-nya. Kita memakai MERGE INTO untuk upsert fact dan pola hapus-lalu-isikan untuk summary. Seluruh pernyataan DDL dan agregasi disimpan sebagai berkas SQL di src/sql/ sehingga skema warehouse dapat dibaca tanpa membuka kode Python.

Skema Warehouse #

sql src/sql/schema.sql (dipangkas) 
CREATE TABLE IF NOT EXISTS fact_pesanan (
    order_id          VARCHAR PRIMARY KEY,
    order_date        DATE NOT NULL,
    customer_name     VARCHAR NOT NULL,
    city              VARCHAR NOT NULL,
    category          VARCHAR NOT NULL,
    quantity          INTEGER NOT NULL,
    unit_price        DECIMAL(12,2) NOT NULL,
    total_price       DECIMAL(14,2) NOT NULL,
    status            VARCHAR NOT NULL,
    source_partition  DATE NOT NULL
);

CREATE TABLE IF NOT EXISTS summary_harian (
    order_date        DATE NOT NULL,
    city              VARCHAR NOT NULL,
    jumlah_pesanan    BIGINT NOT NULL,
    total_pendapatan  DECIMAL(16,2) NOT NULL,
    pendapatan_bersih DECIMAL(16,2) NOT NULL,
    PRIMARY KEY (order_date, city)
);

Dua keputusan pada skema ini penting. Pertama, fact_pesanan berkunci order_id dengan PRIMARY KEY; kunci inilah yang menjadi tumpuan MERGE INTO. Kedua, kolom source_partition mencatat tanggal partisi asal baris, terpisah dari order_date yang diambil dari waktu pesanan. Pemisahan ini membuat rekonsiliasi pada Bagian 9 tetap tepat meskipun suatu hari ada data yang datang terlambat dari partisi sebelumnya, tema chapter Late Arriving Data.

Upsert fact dengan MERGE INTO #

DuckDB menyediakan MERGE INTO sejak versi 1.4, dan versi terpasang pada proyek ini adalah 1.5.6. Fungsi berikut mendaftarkan DataFrame bersih sebagai tabel sementara, lalu menggabungkannya:

python src/load.py — upsert_fact (dipangkas) 
def upsert_fact(con, fact: pd.DataFrame) -> int:
    """Upsert baris fact ke tabel fact_pesanan lewat MERGE INTO."""
    con.register("staged_fact", fact)
    sql = """
        MERGE INTO fact_pesanan AS f
        USING (
            SELECT order_id, order_date, customer_name, city, category,
                   quantity,
                   CAST(unit_price AS DECIMAL(12,2)) AS unit_price,
                   CAST(total_price AS DECIMAL(14,2)) AS total_price,
                   status, source_partition
            FROM staged_fact
        ) AS s
        ON f.order_id = s.order_id
        WHEN MATCHED THEN UPDATE SET
            order_date = s.order_date, city = s.city,
            quantity = s.quantity, total_price = s.total_price,
            status = s.status, source_partition = s.source_partition
        WHEN NOT MATCHED THEN INSERT VALUES (
            s.order_id, s.order_date, s.customer_name, s.city, s.category,
            s.quantity, s.unit_price, s.total_price, s.status, s.source_partition
        )
    """
    try:
        con.execute(sql)
    except duckdb.Error as e:
        raise LoadError(f"MERGE fact gagal: {e}") from e
    return len(fact)

Alurnya baris demi baris:

  1. con.register("staged_fact", fact): DataFrame pandas menjadi tabel SQL bernama staged_fact tanpa menulis berkas perantara.
  2. USING (...): sumber gabungan dibaca dengan CAST ke DECIMAL. Uang disimpan sebagai DECIMAL, bukan FLOAT, agar penjumlahan tidak kehilangan sen.
  3. ON f.order_id = s.order_id: baris sumber disandingkan dengan baris tabel berdasarkan kunci. Baris yang kuncinya sudah ada masuk cabang WHEN MATCHED dan diperbarui; baris baru masuk WHEN NOT MATCHED dan disisipkan.
  4. except duckdb.Error: kegagalan DuckDB dibungkus menjadi LoadError (exit code 5) agar kontrak exit code tetap utuh.

Summary Harian dari SQL #

Agregasi kita per kota per tanggal, termasuk pendapatan bersih yang mengecualikan pesanan dibatalkan lewat klausa FILTER. Karena SQL adalah bahasa yang paling alami untuk pekerjaan ini, agregasi ditulis sebagai berkas:

sql src/sql/summary_harian.sql 
INSERT INTO summary_harian
SELECT
    f.order_date,
    f.city,
    count(*) AS jumlah_pesanan,
    sum(f.total_price) AS total_pendapatan,
    sum(f.total_price) FILTER (WHERE f.status <> 'dibatalkan') AS pendapatan_bersih
FROM fact_pesanan f
WHERE f.order_date = ?
GROUP BY f.order_date, f.city

rebuild_summary(con, tanggal) menghapus summary tanggal tersebut terlebih dulu (DELETE FROM summary_harian WHERE order_date = ?), lalu menjalankan berkas SQL di atas dengan parameter tanggal. Pola hapus-lalu-isikan ini idempotent terhadap eksekusi ulang: hasil kedua kalinya menggantikan, bukan menambah. Pendekatan upsert kunci dan pendekatan ganti partisi dibandingkan pada chapter Incremental Transformation dan Pemuatan Data.

Ekspor Gold #

python src/load.py — export_gold 
def export_gold(con, tanggal, gold_dir) -> Path:
    """Ekspor summary satu tanggal ke CSV zona gold."""
    gold_dir.mkdir(parents=True, exist_ok=True)
    tujuan = gold_dir / f"summary_harian_{tanggal.isoformat()}.csv"
    con.execute(f"""
        COPY (
            SELECT * FROM summary_harian
            WHERE order_date = DATE '{tanggal.isoformat()}'
            ORDER BY city
        ) TO '{tujuan}' (FORMAT CSV, HEADER)
    """)
    return tujuan

COPY ... TO milik DuckDB menulis CSV lengkap dengan header dalam satu pernyataan. Nilai tanggal adalah objek date yang sudah berformat ISO, bukan masukan bebas dari pengguna; bila jalur berkas berasal dari luar, gunakan parameter terikat, bukan penyisipan string.

Eksekusi Nyata #

Menjalankan partisi pertama dan memeriksa isi warehouse:

bash 
python3 -m src.pipeline --date 2026-11-02

Output yang Dihasilkan:

text 
[extract] pesanan_2026-11-02.csv: 8 baris, 771 bytes
[transform] 8 baris -> 8 fact, 0 karantina, 0 duplikat
[load] MERGE fact: 8 baris; summary 2026-11-02: 4 baris
[validate] OK  reconcile kota: 8 baris di 4 kota (total sumber 8 = total target 8)
[validate] OK  agregat harian: fact: 8 baris, Rp3,542,000.00 bruto, Rp3,092,000.00 bersih; summary: 8 baris, Rp3,542,000.00 bruto, Rp3,092,000.00 bersih
[validate] OK  gold csv: 4 baris gold vs 4 baris summary
[gold] data/gold/summary_harian_2026-11-02.csv
[selesai] run-2026-11-02-075011 status=success durasi=43ms exit=0

Query berikut memeriksa warehouse setelah ketiga partisi selesai:

Output yang Dihasilkan:

text 
== fact_pesanan per kota, order_date 2026-11-02 ==
('Bandung', 2, Decimal('1550000.00'))
('Jakarta', 3, Decimal('1439000.00'))
('Surabaya', 2, Decimal('375000.00'))
('Yogyakarta', 1, Decimal('178000.00'))

== summary_harian 2026-11-02 (tabel analitik) ==
('Bandung', 2, Decimal('1550000.00'), Decimal('1550000.00'))
('Jakarta', 3, Decimal('1439000.00'), Decimal('989000.00'))
('Surabaya', 2, Decimal('375000.00'), Decimal('375000.00'))
('Yogyakarta', 1, Decimal('178000.00'), Decimal('178000.00'))

== total warehouse setelah tiga partisi ==
[(23, 12, Decimal('10537000.00'))]

Baris Jakarta memperlihatkan makna kolom pendapatan_bersih: tiga pesanan bernilai bruto Rp1.439.000, tetapi satu di antaranya berstatus dibatalkan sehingga bersihnya Rp989.000. Setelah tiga partisi, warehouse menyimpan 23 baris fact, 12 baris summary, dan total pendapatan Rp10.537.000.

Transaksi BEGIN ... COMMIT serta validasi sebelum commit dibahas pada Bagian 9, dan eksekusi ulang yang identik dibuktikan pada Bagian 5. Bacaan lanjut: dokumentasi DuckDB untuk MERGE INTO, FILTER, dan COPY.