Belajar Apache Spark - DataFrame & Spark SQL

#Pendahuluan

DataFrame dan Spark SQL adalah jantung dari sebagian besar pekerjaan Spark modern. Keduanya memakai abstraksi yang sama — data berkolom dengan schema — sehingga kalian bisa menulis logika sekali dan mengeksekusinya lewat API Python, Scala, maupun SQL murni.

Episode 5 ini membahas: membuat DataFrame dari berbagai format file, menjalankan query SQL dengan temporary views, memahami schema inference versus explicit schema, dan bagaimana Catalyst optimizer menghasilkan eksekusi yang efisien. Ini adalah keterampilan yang paling sering dipakai sehari-hari di dunia data engineering.

#Membuat DataFrame dari Berbagai Sumber

#Membaca CSV

CSV adalah format paling umum untuk data tabular. PySpark menyediakan API spark.read yang ringkas:

Membaca CSV dengan header

df = spark.read.format("csv") \
    .option("header", True) \
    .option("inferSchema", True) \
    .load("data/penjualan.csv")
df.printSchema()

df.printSchema() menampilkan struktur kolom. Opsi header dan inferSchema adalah pasangan yang hampir selalu kalian set — tanpa inferSchema, semua kolom akan dibaca sebagai string.

#Membaca JSON dan Parquet

JSON cocok untuk data semi-terstruktur, sementara Parquet adalah format kolom terkompresi yang paling direkomendasikan untuk data lake:

Membaca JSON dan Parquet

json_df = spark.read.format("json").load("data/event.json")
parquet_df = spark.read.format("parquet").load("data/tabel.parquet")
parquet_df.show(5)

Karena Parquet menyimpan schema di dalam file-nya, tidak perlu konfigurasi tambahan — spark.read.format("parquet") langsung memahami strukturnya. Ini alasan mengapa Parquet menjadi standar de facto untuk penyimpanan data lake.

#Query SQL dengan Spark SQL dan Temporary Views

#Membuat Temporary View

Untuk memakai SQL, daftarkan DataFrame sebagai view:

Registrasi view dan query SQL

df.createOrReplaceTempView("penjualan")
 
hasil = spark.sql("""
    SELECT kategori, SUM(jumlah) AS total
    FROM penjualan
    GROUP BY kategori
    ORDER BY total DESC
""")
hasil.show()

df.createOrReplaceTempView("penjualan") membuat view yang hanya hidup selama session berlangsung. Kalian bisa menulis query SQL yang sama persis dengan yang kalian tulis di database — kalian bahkan bisa menggabungkan view dengan table lain di join.

#Spark SQL Sebagai Bahasa Universal

Spark SQL bukan sekadar ekstensi — seluruh DataFrame API diterjemahkan ke rencana logis yang sama. Artinya kalian bisa memilih antara menulis df.filter(df.jumlah > 100) atau spark.sql("SELECT * FROM penjualan WHERE jumlah > 100"). Keduanya menghasilkan rencana eksekusi yang identik.

df.filter(df.jumlah > 100)      ← API
spark.sql("SELECT ... WHERE ...") ← SQL

#Schema Inference dan Explicit Schema

#Bahaya Schema Inference

inferSchema memang praktis, tapi ada risiko: Spark menebak tipe dari data yang dibaca. Kolom angka dengan satu nilai yang terbaca salah bisa berubah menjadi string, atau nilai kosong mengubah tipe kolom. Untuk pipeline produksi, tebakan otomatis adalah sumber bug yang halus.

#Menyatakan Schema Secara Eksplisit

Untuk kontrol penuh, definisikan schema dengan tipe eksplisit:

Schema eksplisit dengan StructType

from pyspark.sql.types import StructType, StructField, StringType, IntegerType
 
schema = StructType([
    StructField("produk", StringType(), True),
    StructField("stok", IntegerType(), True),
])
 
df = spark.read.format("csv") \
    .option("header", True) \
    .schema(schema) \
    .load("data/stok.csv")

Argumen ketiga StructField adalah nullable — untuk kolom yang wajib ada, set False agar Spark memvalidasi data masuk. Schema eksplisit juga mempercepat loading karena Spark tidak perlu membaca seluruh data hanya untuk menebak tipe.

#Catalyst Optimizer

#Bagaimana Catalyst Bekerja

Catalyst adalah optimizer berbasis aturan yang mengubah query menjadi rencana eksekusi efisien. Ia bekerja dalam beberapa tahap: analysis, logical optimization, physical planning, dan code generation.

SQL/API → logical plan → optimized plan → physical plan → RDD → hasil

#Optimasi yang Kalian Rasakan

Beberapa optimasi Catalyst yang paling berdampak:

• Predicate pushdown: filter dipindahkan sedekat mungkin ke sumber data, sehingga file yang dibaca lebih sedikit.
• Projection pruning: kolom yang tidak dipakai tidak dibaca sama sekali.
• Constant folding: ekspresi konstanta dihitung sekali di awal.
• Join reordering: urutan join dioptimalkan berdasarkan statistik.

Cara terbaik merasakan kerja Catalyst adalah melihat rencana eksekusi:

Melihat physical plan

df.explain("extended")

df.explain("extended") menampilkan logical plan dan physical plan. Membiasakan diri membaca output ini adalah keterampilan kunci untuk episode 9 tentang performance tuning.

#Transformation dan Expression API

#Transformation Utama

Vokabulari DataFrame mirip SQL namun dalam bentuk metode:

Rantai transformasi DataFrame

hasil = df \
    .select("produk", "harga") \
    .filter(df.harga > 50000) \
    .withColumn("harga_ppn", df.harga * 1.11) \
    .groupBy("produk") \
    .agg({"harga_ppn": "avg"})

withColumn menambah atau mengganti kolom, agg menerima kamus agregasi. Rantai ini lazy — eksekusi baru terjadi saat show(), collect(), atau action lain dipanggil.

#Expression API dan Spark Functions

Spark menyediakan fungsi-fungsi bawaan di modul functions:

Fungsi bawaan Spark

from pyspark.sql import functions as F
 
df2 = df.withColumn("bulan", F.month(df.tanggal)) \
        .withColumn("label", F.when(df.stok == 0, "habis").otherwise("tersedia"))

F.when(...).otherwise(...) adalah if-else untuk kolom. Ratusan fungsi lain seperti F.coalesce, F.window, F.row_number akan muncul berulang kali di episode 7 dan 10.

#Penutup

Episode 5 membekali kalian keterampilan utama Spark: membuat DataFrame dari CSV, JSON, dan Parquet, mengquery dengan SQL melalui temporary views, mengendalikan tipe data dengan explicit schema, memahami Catalyst optimizer, dan merangkai transformation dengan expression API.

Inti yang harus dibawa pulang:

• spark.read.format(...) membaca CSV, JSON, Parquet, dan lainnya.
• Temporary view menjembatani DataFrame API dan SQL murni.
• Schema eksplisit mencegah bug tipe data di produksi.
• Catalyst melakukan pushdown, pruning, dan reordering secara otomatis.
• df.explain("extended") adalah alat diagnosis utama.

Di episode 6 selanjutnya kita akan membahas Dataset API dan strong typing — typed Dataset di Scala dan Java, peran encoders dan compile-time type safety, perbedaan struktur Dataset versus DataFrame, serta konversi dua arah antara keduanya.