Bagian 6: Mengatur Partition

Kata partition pada Spark memiliki dua arti yang sering tertukar: partition task adalah potongan data yang diproses satu task di memori, sedangkan partition penyimpanan adalah direktori bulan=YYYY-MM di disk. Pada bagian ini kita mengatur keduanya pada potongan data kuartal pertama 2024 (99.645 baris, tiga bulan), mengukur perbedaan repartition lawan coalesce, lalu membandingkan tiga varian penulisan parquet. Materi pendukung ada pada chapter Spark Partitioning; biaya shuffle di balik repartition dibahas pada Shuffle dan Skew.

Dua Makna Partition #

Keduanya independen: DataFrame 1 partition boleh ditulis menjadi 24 direktori, dan DataFrame 8 partition boleh ditulis ke satu direktori. Kebingungan dua istilah ini adalah sumber galat paling umum saat orang mengharapkan repartition mengubah tata letak disk.

Membaca Ulang Jumlah Partisi #

Fungsi pembandingnya sederhana — memanggil operasi lalu menghitung partition hasilnya:

def bandingkan_repartition_coalesce(df):
    """Bandingkan jumlah partisi dan ukuran data antar operasi partisi.

    `repartition` melakukan shuffle penuh (mahal tapi seimbang), `coalesce`
    menggabungkan partisi sebelah tanpa shuffle (murah tapi tidak seimbang).
    """
    awal = jumlah_partisi(df)
    return {
        "awal": awal,
        "repartition_8": jumlah_partisi(df.repartition(8)),
        "coalesce_2": jumlah_partisi(df.coalesce(2)),
        "coalesce_1": jumlah_partisi(df.coalesce(1)),
    }

Hasilnya pada potongan Q1-2024:

Output yang Dihasilkan:

== potongan Q1-2024 untuk demo (jumlah baris: 99,645) ==
== jumlah partisi DataFrame ==
  {'awal': 1, 'repartition_8': 8, 'coalesce_2': 1, 'coalesce_1': 1}

Tiga angka pada keluaran tersebut layak dijelaskan karena semuanya perilaku resmi Spark, bukan kebetulan:

1. awal: 1 — pembacaan parquet menggabungkan berkas kecil ke dalam satu partition selama ukurannya di bawah spark.sql.files.maxPartitionBytes (bawaan 128 MB). Dataset pesanan hanya 7,6 MB, jadi 24 direktori bulan dibaca sebagai satu partition.
2. repartition_8: 8 — repartition mengacak baris lewat shuffle penuh ke delapan partition, terlihat pada plan sebagai Exchange hashpartitioning; ini satu-satunya cara menambah jumlah partition.
3. coalesce_2: 1 dan coalesce_1: 1 — coalesce hanya dapat menggabungkan partition yang sudah ada, tidak pernah menambah. Karena sumbernya satu partition, coalesce(2) tetap menghasilkan satu.

Tiga Varian Penulisan #

src/partisi.py menulis DataFrame yang sama dengan tiga varian sebelum partitionBy("bulan"):

    mulai = time.perf_counter()
    if varian == "default":
        df.write.mode("overwrite").partitionBy(kolom_partisi).parquet(path)
    elif varian == "repartition":
        df.repartition(F.col(kolom_partisi)).write.mode("overwrite").partitionBy(
            kolom_partisi
        ).parquet(path)
    elif varian == "coalesce":
        df.coalesce(1).write.mode("overwrite").partitionBy(kolom_partisi).parquet(path)

Mari kita telaah baris-baris kode di atas:

1. Baris 3 (default): ditulis apa adanya; jumlah berkas per direktori mengikuti jumlah partition task — pada data sumber yang banyak partition, tiap direktori bisa berisi banyak berkas kecil.
2. Baris 5-7 (repartition): shuffle per nilai bulan sehingga satu task memegang satu bulan; hasilnya tiap direktori berisi sedikit berkas besar. partitionBy lalu tinggal memindahkan baris task ke direktori yang tepat.
3. Baris 9-10 (coalesce): menggabungkan tanpa shuffle — paling murah, namun pada dataset besar seluruh penulisan bisa jatuh ke satu task.

Hasil ketiganya pada mesin uji:

Output yang Dihasilkan:

== tulis 3 varian ke direktori sementara ==
  default      durasi=1.598s berkas=3 rata2=312.7KB partisi_dir=3
  repartition  durasi=1.131s berkas=3 rata2=312.7KB partisi_dir=3
  coalesce     durasi=0.617s berkas=3 rata2=312.7KB partisi_dir=3
  contoh nama berkas varian default (10 pertama):
    bulan=2024-01/part-00000-2d8ab31b-ada0-46bd-9628-bad4a2e89535.c000.snappy.parquet
    bulan=2024-02/part-00000-2d8ab31b-ada0-46bd-9628-bad4a2e89535.c000.snappy.parquet
    bulan=2024-03/part-00000-2d8ab31b-ada0-46bd-9628-bad4a2e89535.c000.snappy.parquet
  jumlah berkas varian repartition (10 pertama):
    bulan=2024-01/part-00000-4257d45b-d7da-4b34-902c-df7c83b59607.c000.snappy.parquet
    bulan=2024-02/part-00000-4257d45b-d7da-4b34-902c-df7c83b59607.c000.snappy.parquet
    bulan=2024-03/part-00000-4257d45b-d7da-4b34-902c-df7c83b59607.c000.snappy.parquet

Pada skala kecil ini ketiga varian menghasilkan struktur berkas yang sama: tiga direktori, satu berkas per direktori, rata-rata 312,7 KB. Alasannya sudah dibaca di bagian sebelumnya — sumber hanya memiliki satu partition, dan task yang tidak memegang baris tidak menulis berkas. Selisih durasi antar varian (0,6 sampai 1,6 detik) tidak konsisten antar eksekusi dan bukan angka yang layak dibandingkan; yang stabil adalah perilakunya pada skala besar, tempat default pada sumber banyak partition menghasilkan banyak berkas kecil per direktori.

Layout Hasil Pipeline #

Pipeline pada Bagian 9 menulis agregat bulanan lewat simpan_partitioned yang memakai repartition per bulan. Tata letaknya di data/hasil/agregat_bulanan:

Output yang Dihasilkan:

== layout data/hasil/agregat_bulanan (pipeline) ==
  {'path': '/home/z/my-project/tutorial/content/data-engineering/proyek/pemrosesan-data-dengan-spark/data/hasil/agregat_bulanan', 'jumlah_partisi_dir': 24, 'jumlah_berkas': 24, 'ukuran_mb': 0.04, 'rata_berkas_kb': 1.5}
  daftar direktori: 24 partisi
    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, masing-masing satu berkas. Ukuran rata-rata berkas hanya 1,5 KB karena agregat 24 bulan x 8 kategori memang 192 baris; pada skala ini berkas kecil tidak menjadi beban. Namun ingat angka itu sebagai peringatan skala: data nyata yang di-partition per hari selama lima tahun menghasilkan lebih dari 1.800 direktori.

Struktur berkas kini terkendali, namun satu alat diagnosis belum kita pegang: bagaimana membaca rencana eksekusi yang memutuskan semuanya? Itu inti Bagian 7.