Bagian 1: Memahami Dataset Besar

Pada Proyek: Streaming Data dengan Kafka kita mengalirkan event saat data tiba; sekarang datanya telah beristirahat sebagai berkas historis dan tugas kita berubah menjadi batch processing. Dataset yang akan diolah adalah hasil ekspor basis data transaksional TokoKita: 800 ribu pesanan dan 1,6 juta baris pesanan selama dua tahun (2024-2025), ditambah dua dimensi kecil berupa 60 ribu pelanggan dan 400 produk. Seluruhnya tersimpan sebagai parquet terpartition sekitar 22,6 MB. Pada bagian ini kita belum menjalankan satu transformasi pun; kita hanya membedah bentuk datanya lewat skema, sampel, distribusi, dan struktur direktori, persis seperti seorang data engineer memeriksa dataset baru sebelum menulis pipeline. Materi pendukung ada pada chapter Apache Spark dan Parquet.

Empat Tabel dan Skemanya #

Dataset terdiri atas dua tabel transaksional besar dan dua tabel dimensi kecil. Tabel pesanan adalah header: satu baris mewakili satu pesanan, lengkap dengan tanggal, pelanggan, kanal pembelian, status, dan diskon. Pembacaan pertama memakai spark.read.parquet kemudian printSchema dan show:

Output yang Dihasilkan:

text
== skema pesanan ==
root
 |-- pesanan_id: string (nullable = true)
 |-- tanggal: date (nullable = true)
 |-- pelanggan_id: long (nullable = true)
 |-- kanal: string (nullable = true)
 |-- status: string (nullable = true)
 |-- diskon_rp: long (nullable = true)
 |-- bulan: string (nullable = true)

+-----------+----------+------------+--------+------------+---------+-------+
|pesanan_id |tanggal   |pelanggan_id|kanal   |status      |diskon_rp|bulan  |
+-----------+----------+------------+--------+------------+---------+-------+
|ORD-0000214|2024-08-01|2770        |aplikasi|selesai     |0        |2024-08|
|ORD-0000215|2024-08-02|2783        |aplikasi|selesai     |0        |2024-08|
|ORD-0000216|2024-08-03|2796        |aplikasi|selesai     |6000     |2024-08|
|ORD-0000217|2024-08-04|2809        |web     |selesai     |0        |2024-08|
|ORD-0000218|2024-08-05|2822        |web     |dikembalikan|0        |2024-08|
+-----------+----------+------------+--------+------------+---------+-------+
only showing top 5 rows

Tabel kedua, baris_pesanan, adalah detail: satu pesanan berisi satu sampai tiga baris produk sehingga jumlah barisnya hampir dua kali jumlah header. Perhatikan kolom harga_satuan yang disalin dari tabel produk pada saat transaksi dicatat — harga historis, bukan harga saat ini:

Output yang Dihasilkan:

text
== skema baris_pesanan ==
root
 |-- pesanan_id: string (nullable = true)
 |-- baris: integer (nullable = true)
 |-- produk_id: integer (nullable = true)
 |-- jumlah: integer (nullable = true)
 |-- harga_satuan: long (nullable = true)

+-----------+-----+---------+------+------------+
|pesanan_id |baris|produk_id|jumlah|harga_satuan|
+-----------+-----+---------+------+------------+
|ORD-0246168|3    |60       |2     |78000       |
|ORD-0012114|1    |352      |1     |77000       |
|ORD-0410562|1    |240      |4     |73000       |
|ORD-0528399|2    |4        |2     |91000       |
|ORD-0098062|1    |340      |4     |38000       |
+-----------+-----+---------+------+------------+
only showing top 5 rows

Dua tabel dimensi jauh lebih kecil. produk memuat 400 item pada delapan kategori; pelanggan memuat 60 ribu orang pada 20 kota dengan tiga segmen. Sepuluh baris pertama keduanya:

Output yang Dihasilkan:

text
+---------+--------------------------+----------+-----+
|produk_id|nama_produk               |kategori  |harga|
+---------+--------------------------+----------+-----+
|1        |KopiKita Signature kecil  |kopi bubuk|85000|
|2        |Gayo Gold Signature kecil |kopi bubuk|87000|
|3        |Rumah Kopi Signature kecil|kopi bubuk|89000|
|4        |Senja Signature kecil     |kopi bubuk|91000|
|5        |Awan Signature kecil      |kopi bubuk|93000|
+---------+--------------------------+----------+-----+
only showing top 5 rows

text
+------------+------------+--------------+-------+----------+
|pelanggan_id|nama        |kota          |segmen |bergabung |
+------------+------------+--------------+-------+----------+
|1           |Budi Santoso|Jakarta       |reguler|2022-01-01|
|2           |Siti Santoso|Bandung       |reguler|2022-01-02|
|3           |Dewi Santoso|Surabaya      |reguler|2022-01-03|
|4           |Andi Santoso|Medan         |reguler|2022-01-04|
|5           |Rina Santoso|Bandar Lampung|reguler|2022-01-05|
+------------+------------+--------------+-------+----------+
only showing top 5 rows

Satu hal penting dari skema di atas: pesanan_id bertipe string (kunci bisnis, bukan angka), pelanggan_id pada parquet dibaca long sementara produk_id pada baris_pesanan bertipe integer. Perbedaan tipe pada kolom kunci join ini akan kita tangani secara sadar pada Bagian 5, dan nullable = true pada seluruh kolom berarti penanganan null pada Bagian 4 bukan opsional.

Distribusi Status dan Nilai Transaksi #

Dua agregasi cepat memberi gambaran isi dataset. Pertama, distribusi status: 85 persen pesanan selesai, 10 persen dikembalikan, 5 persen dibatalkan. Kedua, GMV per kategori setelah baris dijoin ke header dan difilter status selesai:

Output yang Dihasilkan:

text
== distribusi status pesanan ==
+------------+------+
|      status| count|
+------------+------+
|     selesai|680000|
|dikembalikan| 80000|
|  dibatalkan| 40000|
+------------+------+

== GMV per kategori produk (seluruh dataset) ==
+---------------+-----------+------------+
|kategori       |jumlah_item|gmv         |
+---------------+-----------+------------+
|alat seduh     |531984     |137606198000|
|kopi bubuk     |532008     |49831266000 |
|perlengkapan   |517989     |43324979000 |
|kopi saset     |518004     |35556594000 |
|sirup          |532002     |33870938000 |
|susu dan krimer|518013     |27786829000 |
|teh            |517995     |20015516000 |
|cemilan        |532002     |15251048000 |
+---------------+-----------+------------+

Angka alat seduh yang jauh di atas kategori lain bukan kebetulan: harga dasar alat seduh (Rp250.000) memang tiga sampai empat kali harga dasar kategori lain pada daftar HARGA_DASAR di src/generate_data.py. Membaca distribusi seperti ini di awal menyelamatkan kita dari kejutan saat hasil agregasi pada Bagian 5 ditampilkan.

Struktur Fisik: Partition Bulan pada Disk #

Tabel pesanan ditulis dengan partitionBy("bulan") sehingga di disk muncul 24 direktori bulan=YYYY-MM. Konsekuensinya: kolom bulan tidak disimpan di dalam berkas parquet, melainkan menjadi nama direktori — topik yang dibahas pada chapter Spark Partitioning. Tata letaknya:

Output yang Dihasilkan:

text
== direktori partition data/pesanan ==
  bulan=2024-01  (1 berkas parquet)
  bulan=2024-02  (1 berkas parquet)
  bulan=2024-03  (1 berkas parquet)
  bulan=2024-04  (1 berkas parquet)
  bulan=2024-05  (1 berkas parquet)
  bulan=2024-06  (1 berkas parquet)
  bulan=2024-07  (1 berkas parquet)
  bulan=2024-08  (1 berkas parquet)

Daftar berlanjut teratur hingga bulan=2025-12 — total 24 direktori, masing-masing berisi tepat satu berkas parquet. Rekam jejak ukuran disimpan pada data/_meta.json yang ditulis src/generate_data.py setiap kali dataset dibangkitkan:

Output yang Dihasilkan:

text
== ringkasan _meta.json ==
  baris_pesanan  1,599,999 baris    14.7 MB
  pelanggan         60,000 baris     0.3 MB
  pesanan          800,000 baris     7.6 MB
  produk               400 baris     0.0 MB
  total: 22.6 MB

Rumus Pembangkit yang Deterministik #

Seluruh kolom diturunkan dari nomor urut pesanan p (0 sampai 799.999) dengan operasi modular, sehingga dataset bersifat deterministik: pembangkitan ulang menghasilkan jumlah baris dan nilai yang persis sama, dan uji end-to-end pada tests/test_pipeline_e2e.py dapat menghitung ulang agregat di Python murni sebagai pembanding. Inti rumusnya pada src/generate_data.py:

python
    p = F.col("id")
    tanggal = F.date_add(
        F.lit(MULAI_TANGGAL).cast("date"), (p % HARI_DATASET).cast("int")
    )
    return (
        spark.range(jumlah, numPartitions=1)
        .select(
            F.format_string("ORD-%07d", p + 1).alias("pesanan_id"),
            tanggal.alias("tanggal"),
            ((p * 13) % jumlah_pelanggan + 1).alias("pelanggan_id"),
            F.when(p % 10 < 6, "aplikasi")
            .when(p % 10 < 8, "web")
            .otherwise("toko")
            .alias("kanal"),
            F.when(p % 20 < 17, "selesai")
            .when(p % 20 < 19, "dikembalikan")
            .otherwise("dibatalkan")
            .alias("status"),
            F.date_format(tanggal, "yyyy-MM").alias("bulan"),
        )
    )

Mari kita telaah baris-baris kode di atas:

Baris 2-4 (tanggal): pesanan ke-p jatuh pada 2024-01-01 ditambah (p % 731) hari; 731 adalah banyak hari dua tahun. Ini menjelaskan kenapa Desember 2025 dihuni pesanan dengan (nomor - 1) % 731 antara 700 sampai 730 — fakta yang kita pakai untuk menguji konsistensi pada Bagian 5.
Baris 8 (pesanan_id): format_string("ORD-%07d", ...) menghasilkan pad tujuh digit, misalnya ORD-0246168.
Baris 10 (pelanggan_id): (p * 13) % jumlah_pelanggan + 1 menyebarkan pelanggan; tanpa * 13, pesanan berurutan akan menumpuk pada pelanggan berurutan.
Baris 11-17 (kanal dan status): when bertingkat membagi 10 pesanan menjadi 6 aplikasi, 2 web, 2 toko; dan 20 pesanan menjadi 17 selesai, 2 dikembalikan, 1 dibatalkan.
Baris 18 (bulan): kolom partition diturunkan dari tanggal, bukan kolom terpisah, sehingga tidak mungkin berbeda dengan tanggal transaksinya. Kolom diskon_rp dibangkitkan dengan pola serupa (satu dari lima pesanan menerima diskon Rp2.000 sampai Rp6.000).

Dataset kini dikenal dari sisi skema, distribusi, dan tata letak fisiknya. Langkah berikutnya bukan menulis transformasi, melainkan memastikan mesin pengolahnya siap: satu SparkSession local[1] dengan konfigurasi yang dapat diandalkan, tugas Bagian 2.