Hasil pipeline baru bernilai setelah tersimpan dalam bentuk yang bisa dibaca konsumen berikutnya — dashboard, warehouse, atau tim analisis. Pada bagian penutup ini kita menulis agregat bulanan sebagai parquet terpartition bulan, membacanya kembali, lalu memverifikasi isinya terhadap agregat segar yang dihitung ulang dari sumber. Ujian terakhirnya idempotensi: menjalankan pipeline dua kali harus menghasilkan berkas yang identik, bukan berkas yang bertumpuk. Materi pendukung ada pada chapter Spark Partitioning; pertimbangan kapan langkah seperti ini layak dijalankan Spark dibahas pada Kapan Menggunakan Spark.

Seluruh penulisan hasil melewati satu fungsi pada src/simpan.py:

python 
def simpan_partitioned(df, path, kolom_partisi=("bulan",), mode="overwrite", satu_berkas=True):
    """Tulis DataFrame parquet partition by kolom_partisi.

    `satu_berkas=True` memakai `repartition(kolom_partisi)` sehingga tiap
    direktori partition berisi sedikit berkas besar (hindari berkas kecil).
    """
    path = str(path)
    if satu_berkas and kolom_partisi:
        df = df.repartition(*[F.col(k) for k in kolom_partisi])
    df.write.mode(mode).partitionBy(*kolom_partisi).parquet(path)

Mari kita telaah baris-baris kode di atas:

  1. Baris 8-9 (repartition): data diacak per nilai bulan sebelum ditulis — lanjutan langsung Bagian 6 — sehingga setiap direktori partition diisi oleh satu task dengan satu berkas besar.
  2. Baris 10 (mode="overwrite"): penulisan ulang mengganti isi direktori hasil; inilah yang membuat pipeline idempoten karena run kedua tidak menumpuk di samping run pertama.
  3. Baris 10 (partitionBy): kolom bulan keluar dari isi berkas dan menjadi nama direktori; saat dibaca kembali, kolom itu muncul dari nama direktori.
  4. Baris 3 (satu_berkas=True): tombol ini bisa dimatikan bila penerima hasil justru menginginkan banyak berkas kecil per direktori.

Membaca Kembali Hasil Pipeline #

Pengujian penulisan tidak selesai di write; kita membuka kembali data/hasil/agregat_bulanan dan memeriksa isinya:

Output yang Dihasilkan:

text 
== baca kembali hasil pipeline ==
root
 |-- kategori: string (nullable = true)
 |-- gmv: long (nullable = true)
 |-- jumlah_pesanan: long (nullable = true)
 |-- jumlah_item: long (nullable = true)
 |-- bulan: string (nullable = true)

+---------------+----------+--------------+-----------+-------+
|kategori       |gmv       |jumlah_pesanan|jumlah_item|bulan  |
+---------------+----------+--------------+-----------+-------+
|kopi saset     |1504998000|4771          |21930      |2025-12|
|perlengkapan   |1841052000|4790          |22008      |2025-12|
|susu dan krimer|1176959000|4769          |21937      |2025-12|
|kopi bubuk     |2114081000|4836          |22569      |2025-12|
|alat seduh     |5829784000|4829          |22539      |2025-12|
|teh            |848624000 |4778          |21967      |2025-12|
|cemilan        |646710000 |4840          |22561      |2025-12|
|sirup          |1435175000|4829          |22539      |2025-12|
|kopi bubuk     |2114905000|4842          |22581      |2025-01|
|susu dan krimer|1176537000|4770          |21943      |2025-01|
+---------------+----------+--------------+-----------+-------+
only showing top 10 rows
jumlah baris: 192
jumlah bulan : 24
GMV total    : 363,243,368,000

Kolom bulan berada paling akhir pada skema — tanda bahwa kolom tersebut datang dari nama direktori, bukan dari isi berkas. Angka show(10) menampilkan Desember 2025 paling atas karena show tanpa orderBy mengikuti urutan task; nilai Desember persis sama dengan GMV per kategori pada Bagian 8 karena keduanya dihitung dari sumber yang sama dengan rumus yang sama. GMV total seluruh 24 bulan: Rp363,2 miliar.

Layout dan Verifikasi Baca Ulang #

Tata letak direktori hasil:

Output yang Dihasilkan:

text 
== layout partisi data/hasil ==
    bulan=2024-01: 1 berkas
    bulan=2024-02: 1 berkas
    bulan=2024-03: 1 berkas
    bulan=2024-04: 1 berkas
    bulan=2024-05: 1 berkas
    bulan=2024-06: 1 berkas
    bulan=2024-07: 1 berkas
    bulan=2024-08: 1 berkas

Daftar berlanjut hingga bulan=2025-12 — 24 direktori, satu berkas per direktori. Verifikasi kemudian membandingkan berkas yang disimpan terhadap agregat segar yang dihitung ulang dari data/ — keduanya dikelompokkan per bulan lalu dibandingkan pada jumlah baris dan jumlah gmv:

Output yang Dihasilkan:

text 
== verifikasi baca ulang terhadap agregat segar ==
  cocok: True | baris sumber: 192 | baris baca: 192
  contoh per kunci: [('2024-01', 8, 15411002000), ('2024-02', 8, 14419678000), ('2024-03', 8, 15416477000)]

cocok: True berarti seluruh 24 kunci bulan cocok pada kedua ukuran; contoh per kunci menampilkan bulan, banyak kategori (8), dan GMV. Pengujian tests/test_simpan.py mengunci perilaku ini dari dua arah: test_verifikasi_mendeteksi_selisih memastikan verifikasi gagal bila sumber diganti, dan test_overwrite_mengganti_isi_lama memastikan penulisan kedua mengganti isi lama.

Pipeline End-to-End dan Idempotensi #

Perintah python3 -m src.pipeline menjalankan kelima tahap dan mencatat waktunya:

Output yang Dihasilkan:

text 
Pipeline selesai.
  1_siapkan_dataset          0.0 s  {'sumber': 'data/ sudah ada'}
  2_baca                    3.94 s  {'pesanan': 800000, 'baris': 1599999}
  3_transformasi            1.84 s  {'pesanan_selesai': 680000}
  4_join_agregasi           4.23 s  {'baris_hasil': 192}
  5_simpan_verifikasi      14.29 s  {'cocok': True, 'partisi': 24, 'baris_hasil': 192}
ringkasan: {'pesanan': 800000, 'baris_pesanan': 1599999, 'pesanan_selesai': 680000, 'baris_hasil': 192, 'verifikasi_cocok': True}
total: 24.76 s

Tahap 5 menjadi yang paling mahal (14,29 detik) karena mencakup repartition, penulisan 24 direktori, dan verifikasi yang membaca ulang hasil. Idempotensi diuji oleh tests/test_pipeline_e2e.py::test_idempoten_dan_deterministik: pipeline dijalankan tiga kali (pakai dataset, pakai ulang, bangun ulang) dan ringkasan ketiganya harus persis sama — dataset deterministik Bagian 1 bertemu mode("overwrite") Bagian ini untuk menghasilkan jaminan tersebut.

Rekam jejak terakhir dilihat dari _meta.json:

Output yang Dihasilkan:

text 
== ringkasan _meta.json (rekam jejak dataset) ==
  dibuat_pada: 2026-10-17T08:21:24.039167+00:00
  ukuran_total_mb: 22.6

Proyek lengkap: dataset dipahami, session dikonfigurasi, data dibaca, ditransformasi, dijoin, diagregasi, partition diatur, rencana dibaca, job dioptimasi, dan hasil disimpan dengan verifikasi. Hasil akhirnya — agregat bulanan per kategori berformat parquet terpartition — persis bentuk data yang dikonsumsi lapisan berikutnya. Pada Proyek: Pipeline Data dan Dashboard Analitik kita menyusun pipeline dari ujung ke ujung dan menyajikan hasilnya pada dashboard, tempat berkas seperti data/hasil/agregat_bulanan menjadi bahan baku visualisasi.