Spark Partitioning

Setiap DataFrame yang kita proses sebenarnya adalah kumpulan partition — potongan data yang menjadi unit kerja terkecil Spark: satu task memproses satu partition di satu executor. Jumlah dan bentuk partition menentukan seberapa paralel pekerjaan berjalan serta berapa banyak data berpindah saat shuffle. Pada chapter ini kita membandingkan repartition dan coalesce, menulis Parquet terpartisi dengan partitionBy, membuktikan partition pruning saat membaca dengan filter, lalu menutup dengan masalah berkas kecil yang kerap melambatkan pipeline produksi. Seluruh bukti, seperti biasa, kita baca dari eksekusi nyata Spark.

Partition adalah Unit Kerja Spark #

Parallelisme Spark sederhana aritmetikanya: dengan 100 partition dan 10 executor, hanya 10 task yang berjalan serentak dan totalnya 10 gelombang eksekusi. Partition terlalu sedikit membuat executor menganggur; partition terlalu banyak menimbulkan biaya penjadwalan task dan metadata. Dua operasi utama untuk mengubah jumlah partition — repartition() dan coalesce() — memilih kompromi yang berbeda, dan tabel berikut merangkum keduanya sebelum kita buktikan lewat eksekusi:

Skrip Lengkap dan Outputnya #

"""Spark partitioning: repartition vs coalesce, parquet terpartisi, dan pruning."""
import shutil
from pathlib import Path

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, input_file_name

spark = (
    SparkSession.builder
    .master("local[1]")
    .config("spark.ui.enabled", "false")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")

print("== 1. repartition vs coalesce ==")
df = spark.range(0, 100000, numPartitions=8)
print("Partition awal                    :", df.rdd.getNumPartitions())

sebar = df.repartition(3)
print("Setelah repartition(3)            :", sebar.rdd.getNumPartitions())
print("Isi per partition                 :", sebar.rdd.glom().map(len).collect())

gabung = df.coalesce(2)
print("Setelah coalesce(2)               :", gabung.rdd.getNumPartitions())
print("Isi per partition                 :", gabung.rdd.glom().map(len).collect())

print("== 2. Partition hasil shuffle ==")
spark.conf.set("spark.sql.adaptive.enabled", "false")
print("AQE dimatikan, shuffle default :",
      df.groupBy(col("id") % 3).count().rdd.getNumPartitions())
spark.conf.set("spark.sql.adaptive.enabled", "true")
print("AQE aktif, partition digabung  :",
      df.groupBy(col("id") % 3).count().rdd.getNumPartitions())

print("== 3. Menulis parquet terpartisi ==")
pesanan = spark.createDataFrame(
    [
        (1, "Budi", "Bandung", "Kemeja Batik", 2, 240000),
        (2, "Siti", "Jakarta", "Kain Songket", 1, 850000),
        (3, "Dewi", "Surabaya", "Rok Tenun", 3, 310000),
        (4, "Budi", "Bandung", "Blouse Katun", 1, 120000),
        (5, "Rina", "Bandung", "Kemeja Batik", 2, 240000),
        (6, "Siti", "Jakarta", "Rok Tenun", 1, 310000),
    ],
    "id INT, pelanggan STRING, kota STRING, produk STRING, qty INT, harga LONG",
)
direktori = Path("parquet_pesanan")
if direktori.exists():
    shutil.rmtree(direktori)
pesanan.write.partitionBy("kota").parquet(str(direktori))
print("Struktur direktori hasil tulis:")
for berkas in sorted(direktori.rglob("*.parquet")):
    print("  ", berkas.parent.relative_to(direktori).as_posix())

print("== 4. Partition pruning saat baca ==")
df_baca = spark.read.parquet(str(direktori))
df_baca.show()

nama_berkas = df_baca.filter(col("kota") == "Bandung").select(input_file_name())
terbaca_bandung = [f[0].split("parquet_pesanan/")[-1].rsplit("/", 1)[0]
                   for f in nama_berkas.distinct().collect()]
print("Filter kota='Bandung' membaca direktori:", sorted(set(terbaca_bandung)))

semua = df_baca.select(input_file_name()).distinct().collect()
print("Tanpa filter membaca", len(semua), "berkas parquet")

Output yang Dihasilkan:

== 1. repartition vs coalesce ==
Partition awal                    : 8
Setelah repartition(3)            : 3
Isi per partition                 : [33335, 33332, 33333]
Setelah coalesce(2)               : 2
Isi per partition                 : [50000, 50000]
== 2. Partition hasil shuffle ==
AQE dimatikan, shuffle default : 200
AQE aktif, partition digabung  : 1
== 3. Menulis parquet terpartisi ==
Struktur direktori hasil tulis:
   kota=Bandung
   kota=Jakarta
   kota=Surabaya
== 4. Partition pruning saat baca ==
+---+---------+------------+---+------+--------+
| id|pelanggan|      produk|qty| harga|    kota|
+---+---------+------------+---+------+--------+
|  1|     Budi|Kemeja Batik|  2|240000| Bandung|
|  4|     Budi|Blouse Katun|  1|120000| Bandung|
|  5|     Rina|Kemeja Batik|  2|240000| Bandung|
|  2|     Siti|Kain Songket|  1|850000| Jakarta|
|  6|     Siti|   Rok Tenun|  1|310000| Jakarta|
|  3|     Dewi|   Rok Tenun|  3|310000|Surabaya|
+---+---------+------------+---+------+--------+

Filter kota='Bandung' membaca direktori: ['kota=Bandung']
Tanpa filter membaca 3 berkas parquet

Mari kita telaah baris-baris kode di atas:

1. Baris 17 (spark.range(0, 100000, numPartitions=8)): DataFrame 100 ribu baris dibuat langsung dalam 8 partition — 12.500 baris per partition.
2. Baris 20-22 (repartition(3)): data disebar ulang lewat shuffle menjadi 3 partition; hasil glom() membuktikan sebarannya nyaris merata: [33335, 33332, 33333].
3. Baris 24-26 (coalesce(2)): 8 partition digabung menjadi 2 tanpa shuffle — partition bersebelahan dilebur menjadi [50000, 50000].
4. Baris 29-34 (AQE): dengan AQE dimatikan, agregasi groupBy menghasilkan shuffle 200 partition (nilai default spark.sql.shuffle.partitions); setelah AQE diaktifkan, partition kecil otomatis digabung menjadi 1 pada mesin satu core ini.
5. Baris 51 (partitionBy("kota")): penulisan Parquet dipecah per nilai kolom kota; baris 53-54 mencetak struktur direktorinya — satu direktori kota=... per kota, dan kolom kota tidak lagi tersimpan di dalam berkas melainkan pada nama direktori.
6. Baris 60-63 (input_file_name()): fungsi ini melaporkan berkas asal tiap baris; setelah filter kota = 'Bandung', hanya berkas dari direktori kota=Bandung yang muncul.
7. Baris 65-66 (tanpa filter): tanpa filter, seluruh 3 berkas parquet dibaca — pembanding bagi angka sebelumnya.

Partition Pruning: Filter yang Menyentuh Nama Direktori #

Bagian 4 output adalah bukti partition pruning yang jarang diperlihatkan secara eksplisit. Filter pada kolom kota diterjemahkan Spark menjadi seleksi direktori: hanya kota=Bandung yang dibuka, dua direktori lain tidak tersentuh sama sekali. Pada data latihan ini penghematannya hanya 2 dari 3 berkas, tetapi pada tabel 500 juta baris yang terpartisi per bulan, query satu bulan berarti membaca satu direktori, bukan seluruh tabel — inilah alasan tabel lakehouse hampir selalu terpartisi menurut tanggal. Panduan resmi pola direktori kota=nilai ini ada di dokumentasi Parquet Spark .

Masalah Berkas Kecil #

Output bagian 2 menyimpan pelajaran produksi. Tanpa AQE, agregasi kecil apa pun menghasilkan 200 partition; jika hasilnya langsung ditulis ke disk, kita mendapat 200 berkas kecil yang sebagian besar nyaris kosong. Berkas kecil berlipat biayanya: lebih banyak metadata yang harus dilacak sistem file atau object store, lebih banyak task hanya untuk membuka berkas, dan operasi daftar direktori yang melambat jauh sebelum datanya sendiri besar. Penanganannya ada dua arah — coalesce() sebelum menulis agar jumlah berkas masuk akal, atau kompak ulang berkala dengan proses terpisah. Konfigurasi AQE yang menggabungkan partition otomatis (dan seluruh pengaturan spark.sql.* lain) terdokumentasi di halaman konfigurasi Spark .

Partition yang tepat membuat pekerjaan selesai lebih cepat tanpa mengubah satu baris logika bisnis. Namun mengukur mana yang “tepat” membutuhkan alat baca performa — pekerjaan yang kita kerjakan pada chapter Spark Performance.