Bagian 7: Menganalisis Query Plan

Setiap transformasi yang kita tulis di Bagian 4 sampai 6 diterjemahkan Catalyst menjadi rencana eksekusi, dan rencana itulah yang memutuskan join mana yang dipakai, filter mana yang didorong ke berkas, dan shuffle mana yang tidak bisa dihindari. Pada bagian ini kita membaca rencana agregasi bulanan dari Bagian 5 pada tiga mode explain, lalu menambang penanda PartitionFilters dan PushedFilters sebagai bukti pemangkasan. Materi pendukung ada pada chapter Spark SQL dan Spark Performance; setiap Exchange yang muncul adalah shuffle yang dibahas pada Shuffle dan Skew.

Menangkap Rencana menjadi String #

explain pada umumnya mencetak ke layar; src/plan.py mengalihkan stdout sebentar sehingga rencana dapat dikembalikan sebagai string dan disertakan pada laporan:

python
def tangkap_rencana(df, mode="formatted"):
    """Kembalikan teks query plan pada `mode` (formatted/simple/extended/cost).

    `explain` bawaan mencetak ke stdout; fungsi ini mengalihkan stdout
    sementara agar teks rencana dapat dikembalikan sebagai string.
    """
    import io
    from contextlib import redirect_stdout

    penyangga = io.StringIO()
    with redirect_stdout(penyangga):
        df.explain(mode)
    return penyangga.getvalue().strip()

Mari kita telaah baris-baris kode di atas:

Baris 10-11 (import lokal): io dan redirect_stdout hanya dibutuhkan sekejap, jadi dideklarasikan di dalam fungsi.
Baris 13-15 (redirect): seluruh print dari df.explain(mode) dialihkan ke penyangga StringIO; tidak ada keluaran yang bocor ke layar pengguna.
Baris 16 (nilai balik): teks rencana dikembalikan tanpa spasi awal-akhir; tanda_partisi dan tabel_plan kemudian memindai teks ini dengan pencocokan baris.

Pohon Rencana Mode formatted #

Kita menjalankan tangkap_rencana(kueri, "formatted") atas agregasi bulanan (gabungan tiga tabel, countDistinct). Bagian atas keluarannya berupa pohon yang dibaca dari bawah (Scan) ke atas (Sort), dan setiap operator memperoleh nomor yang dirinci kemudian:

Output yang Dihasilkan:

text
== explain('formatted'): agregasi bulanan ==
== Physical Plan ==
AdaptiveSparkPlan (23)
+- Sort (22)
   +- Exchange (21)
      +- HashAggregate (20)
         +- Exchange (19)
            +- HashAggregate (18)
               +- HashAggregate (17)
                  +- Exchange (16)
                     +- HashAggregate (15)
                        +- Project (14)
                           +- BroadcastHashJoin Inner BuildRight (13)
                              :- Project (9)
                              :  +- BroadcastHashJoin Inner BuildRight (8)
                              :     :- Project (3)
                              :     :  +- Filter (2)
                              :     :     +- Scan parquet  (1)
                              :     +- BroadcastExchange (7)
                              :        +- Project (6)
                              :           +- Filter (5)
                              :              +- Scan parquet  (4)
                              +- BroadcastExchange (12)
                                 +- Filter (11)
                                    +- Scan parquet  (10)

Empat baris terbaca langsung dari pohon tersebut. Pertama, kedua join adalah BroadcastHashJoin — bukti Bagian 5 tercetak kembali oleh optimizer tanpa kita minta. Kedua, tiga Exchange adalah tiga shuffle: dua BroadcastExchange (sisi kecil dikirim ke task) dan satu Exchange agregasi. Ketiga, ada tiga HashAggregate bertumpuk — akibat countDistinct yang dijelaskan di bawah. Keempat, Sort paling atas berasal dari orderBy("bulan") dan menghabiskan satu Exchange rangepartitioning.

Detail Operator: Scan dan PushedFilters #

Mode formatted merinci setiap operator bernomor. Operator (4) adalah pembacaan header pesanan yang difilter status:

Output yang Dihasilkan:

text
(4) Scan parquet 
Output [3]: [pesanan_id#0, status#4, bulan#6]
Batched: true
Location: InMemoryFileIndex [file:/home/z/my-project/tutorial/content/data-engineering/proyek/pemrosesan-data-dengan-spark/data/pesanan]
PushedFilters: [IsNotNull(status), EqualTo(status,selesai), IsNotNull(pesanan_id)]
ReadSchema: struct<pesanan_id:string,status:string>

Baris Output memuat tiga kolom padahal tabel aslinya tujuh — Spark hanya membaca kolom yang dipakai rencana (column pruning). PushedFilters menunjukkan predikat yang didorong sampai ke pembaca parquet: EqualTo(status,selesai) dievaluasi pada row group sebelum baris dibangunkan. Location berupa InMemoryFileIndex atas satu direktori, dan ReadSchema hanya menyisakan dua kolom pada berkas.

Tiga Tingkat Agregasi untuk countDistinct #

countDistinct("pesanan_id") memaksa Catalyst bekerja dua fase. Operator (15) dan (16) memperlihatkan fase pertama:

Output yang Dihasilkan:

text
(15) HashAggregate
Input [4]: [pesanan_id#7, baris#8, total#17L, bulan#6]
Keys [2]: [bulan#6, pesanan_id#7]
Functions [3]: [partial_sum(total#17L), partial_count(baris#8), partial_avg(total#17L)]
Aggregate Attributes [3]: [sum(total#17L)#38L, count(baris#8)#40L, avg(total#17L)#41]
Results [6]: [bulan#6, pesanan_id#7, sum#43L, count#45L, sum#48, count#49L]

(16) Exchange
Input [6]: [bulan#6, pesanan_id#7, sum#43L, count#45L, sum#48, count#49L]
Arguments: hashpartitioning(bulan#6, pesanan_id#7, 8), ENSURE_REQUIREMENTS, [plan_id=73]

Fase pertama mengagregasi per pasangan bulan x pesanan_id (kunci memuat kolom distinct), hasilnya diacak dengan hashpartitioning(bulan, pesanan_id, 8), lalu operator (18) dan (20) menggabungkan per bulan menjadi count(distinct pesanan_id). Tanpa countDistinct, rencana hanya membutuhkan satu tingkat agregasi — inilah harga yang dibayar untuk ketepatan “jumlah pesanan unik”, dan alasan agregasi_bulanan lebih mahal daripada agregasi_kategori_bulanan pada pipeline.

Mode simple untuk Perbandingan Ringkas #

Mode yang sama dalam satu baris per operator — cukup untuk memastikan struktur tanpa merinci argumen:

Output yang Dihasilkan:

text
== explain('simple') untuk perbandingan ringkas ==
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Sort [bulan#6 ASC NULLS FIRST], true, 0
   +- Exchange rangepartitioning(bulan#6 ASC NULLS FIRST, 8), ENSURE_REQUIREMENTS, [plan_id=80]
      +- HashAggregate(keys=[bulan#6], functions=[sum(total#17L), count(baris#8), avg(total#17L), count(distinct pesanan_id#7)])
         +- Exchange hashpartitioning(bulan#6, 8), ENSURE_REQUIREMENTS, [plan_id=77]
            +- HashAggregate(keys=[bulan#6], functions=[merge_sum(total#17L), merge_count(baris#8), merge_avg(total#17L), partial_count(distinct pesanan_id#7)])
               +- HashAggregate(keys=[bulan#6, pesanan_id#7], functions=[merge_sum(total#17L), merge_count(baris#8), merge_avg(total#17L)])
                  +- Exchange hashpartitioning(bulan#6, pesanan_id#7, 8), ENSURE_REQUIREMENTS, [plan_id=73]
                     +- HashAggregate(keys=[bulan#6, pesanan_id#7], functions=[partial_sum(total#17L), partial_count(baris#8), partial_avg(total#17L)])
                        +- Project [pesanan_id#7, baris#8, total#17L, bulan#6]
                           +- BroadcastHashJoin [cast(produk_id#9 as bigint)], [produk_id#12L], Inner, BuildRight, false, false
                              :- Project [pesanan_id#7, baris#8, produk_id#9, total#17L, bulan#6]
                              :  +- BroadcastHashJoin [pesanan_id#7], [pesanan_id#0], Inner, BuildRight, false, false
                              :     :- Project [pesanan_id#7, baris#8, produk_id#9, (cast(jumlah#10 as bigint) * harga_satuan#11L) AS total#17L]
                              :     :  +- Filter (isnotnull(pesanan_id#7) AND isnotnull(produk_id#9))
                              :     :     +- FileScan parquet [pesanan_id#7,baris#8,produk_id#9,jumlah#10,harga_satuan#11L] Batched: true, DataFilters: [isnotnull(pesanan_id#7), isnotnull(produk_id#9)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/home/z/my-project/tutorial/content/data-engineering/proyek/pemro..., PartitionFilters: [], PushedFilters: [IsNotNull(pesanan_id), IsNotNull(produk_id)], ReadSchema: struct<pesanan_id:string,baris:int,produk_id:int,jumlah:int,harga_satuan:bigint>
                              :     +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, true]),false), [plan_id=64]
                              :        +- Project [pesanan_id#0, bulan#6]
                              :           +- Filter ((isnotnull(status#4) AND (status#4 = selesai)) AND isnotnull(pesanan_id#0))
                              :              +- FileScan parquet [pesanan_id#0,status#4,bulan#6] Batched: true, DataFilters: [isnotnull(status#4), (status#4 = selesai), isnotnull(pesanan_id#0)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/home/z/my-project/tutorial/content/data-engineering/proyek/pemro..., PartitionFilters: [], PushedFilters: [IsNotNull(status), EqualTo(status,selesai), IsNotNull(pesanan_id)], ReadSchema: struct<pesanan_id:string,status:string>
                              +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]),false), [plan_id=68]
                                 +- Filter isnotnull(produk_id#12L)
                                    +- FileScan parquet [produk_id#12L] Batched: true, DataFilters: [isnotnull(produk_id#12L)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/home/z/my-project/tutorial/content/data-engineering/proyek/pemro..., PartitionFilters: [], PushedFilters: [IsNotNull(produk_id)], ReadSchema: struct<produk_id:bigint>

Perhatikan cast(produk_id#9 as bigint) pada BroadcastHashJoin — tanda kurung cast pada kunci join yang kita singgung di Bagian 5 tercetak apa adanya di sini.

PartitionFilters dan PushedFilters #

Filter pada kolom partition dan kolom biasa meninggalkan jejak berbeda. Potongan Desember 2025 berstatus selesai:

Output yang Dihasilkan:

text
== potongan terfilter: PartitionFilters dan PushedFilters ==
  PartitionFilters: [isnotnull(bulan#6), (bulan#6 = 2025-12)]
  PushedFilters: [IsNotNull(status), EqualTo(status,selesai)]
  Scan: +- Scan parquet  (1)

Predikat bulan = 2025-12 masuk PartitionFilters — 23 direktori lain tidak dibuka sama sekali. Predikat status = selesai masuk PushedFilters — dievaluasi pada metadata row group parquet. Kedua penanda inilah yang kita cari setiap kali bertanya “mengapa query ini cepat atau lambat”.

Mode extended: Empat Tahap Catalyst #

Mode extended mencetak keempat tahap sekaligus: parsed (apa yang kita tulis), analyzed (resolusi kolom), optimized (aturan Catalyst diterapkan), physical (operator siap eksekusi):

Output yang Dihasilkan:

text
== extended: potongan parsed/optimized/physical ==
== Parsed Logical Plan ==
'Project ['pesanan_id, 'status]
+- Filter ((bulan#6 = 2025-12) AND (status#4 = selesai))
   +- Relation [pesanan_id#0,tanggal#1,pelanggan_id#2L,kanal#3,status#4,diskon_rp#5L,bulan#6] parquet

== Analyzed Logical Plan ==
pesanan_id: string, status: string
Project [pesanan_id#0, status#4]
+- Filter ((bulan#6 = 2025-12) AND (status#4 = selesai))
   +- Relation [pesanan_id#0,tanggal#1,pelanggan_id#2L,kanal#3,status#4,diskon_rp#5L,bulan#6] parquet

== Optimized Logical Plan ==
Project [pesanan_id#0, status#4]
+- Filter ((isnotnull(bulan#6) AND isnotnull(status#4)) AND ((bulan#6 = 2025-12) AND (status#4 = selesai)))
   +- Relation [pesanan_id#0,tanggal#1,pelanggan_id#2L,kanal#3,status#4,diskon_rp#5L,bulan#6] parquet

== Physical Plan ==
*(1) Project [pesanan_id#0, status#4]
+- *(1) Filter (isnotnull(status#4) AND (status#4 = selesai))
   +- *(1) ColumnarToRow
      +- FileScan parquet [pesanan_id#0,status#4,bulan#6] Batched: true, DataFilters: [isnotnull(status#4), (status#4 = selesai)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/home/z/my-project/tutorial/content/data-engineering/proyek/pemro..., PartitionFilters: [isnotnull(bulan#6), (bulan#6 = 2025-12)], PushedFilters: [IsNotNull(status), EqualTo(status,selesai)], ReadSchema: struct<pesanan_id:string,status:string>

Perjalanan filter pada contoh ini terbaca lengkap: parsed menyimpan (bulan = 2025-12) AND (status = selesai) apa adanya; analyzed memberi setiap kolom nomor resolusi; optimized menyisipkan isnotnull yang diperlukan join dan menata ulang predikat; physical menambah ColumnarToRow (perpindahan dari kolom ke baris) serta memindahkan predikat bulan ke PartitionFilters. Awalan *(1) menandai whole-stage codegen: operator dalam tanda kurung yang sama dikompile menjadi satu fungsi JVM.

Dengan kemampuan membaca rencana, optimasi tidak lagi dugaan: kita tahu persis operator mana yang bekerja dan filter mana yang didorong. Bagian 8 mengubah pemahaman ini menjadi pengukuran — cache, pemangkasan kolom, dan gabungan teknik — lalu membandingkannya dengan angka nyata. Rujukan API explain ada di https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html dan dokumentasi utama Spark di https://spark.apache.org/docs/latest/ .