Spark DataFrames

Peta besar Spark sudah kita pegang: driver, executor, dan lazy evaluation. API kerja sehari-hari di atas mesin tersebut adalah DataFrame — tabel terdistribusi dengan kolom bernama dan schema yang jelas, konsep yang memang diadopsi dari pandas namun dirancang untuk berjalan paralel di banyak executor. Pada chapter ini kita membangun DataFrame pesanan TokoKita dari data mentah, menjalankan operasi inti select, filter, withColumn, dan groupBy, lalu membandingkan dua cara menentukan schema. Di akhir chapter, kita juga memetakan perbedaan mendasar antara pandas dan Spark DataFrame supaya tahu kapan memakai yang mana.

Data Pesanan dan Dua Cara Menentukan Schema #

DataFrame Spark bisa dibuat dari list Python, berkas CSV dan Parquet, tabel katalog, hingga database eksternal. Dari sisi schema ada dua jalan: biarkan Spark menebak tipe tiap kolom (inferensi), atau tentukan sendiri lewat StructType. Kedua jalan menghasilkan DataFrame yang sama-sama bisa dipakai, tetapi seperti akan kita lihat pada output, hasilnya tidak identik — dan pada data ratusan juta baris, perbedaan tipe ikut menentukan pemakaian memori.

Skrip Lengkap dan Outputnya #

Skrip berikut memuat enam pesanan TokoKita, membandingkan kedua cara menentukan schema, lalu berjalan dari operasi dasar sampai agregasi omzet per kota:

"""Spark DataFrame: membuat data, operasi dasar, agregasi, dan schema eksplisit."""
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as f_sum
from pyspark.sql.types import IntegerType, LongType, StringType, StructField, StructType

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

# Data pesanan TokoKita: (id, pelanggan, kota, produk, qty, harga)
pesanan = [
    (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),
]

print("== 1. Schema hasil inferensi ==")
df_infer = spark.createDataFrame(pesanan, ["id", "pelanggan", "kota", "produk", "qty", "harga"])
df_infer.printSchema()

print("== 2. Schema yang kita definisikan sendiri ==")
skema = StructType([
    StructField("id", IntegerType(), False),
    StructField("pelanggan", StringType(), True),
    StructField("kota", StringType(), True),
    StructField("produk", StringType(), True),
    StructField("qty", IntegerType(), True),
    StructField("harga", LongType(), True),
])
df = spark.createDataFrame(pesanan, skema)
df.printSchema()
df.show()

print("== 3. select dan filter berantai ==")
df.select("pelanggan", "produk", "qty").filter(col("qty") > 1).show()

print("== 4. withColumn: kolom baru hasil hitung ==")
df_total = df.withColumn("total", col("qty") * col("harga"))
df_total.show()

print("== 5. groupBy: agregasi per kota ==")
df_total.groupBy("kota").agg(
    f_sum("qty").alias("total_qty"),
    f_sum("total").alias("omzet"),
).orderBy("kota").show()

Output yang Dihasilkan:

== 1. Schema hasil inferensi ==
root
 |-- id: long (nullable = true)
 |-- pelanggan: string (nullable = true)
 |-- kota: string (nullable = true)
 |-- produk: string (nullable = true)
 |-- qty: long (nullable = true)
 |-- harga: long (nullable = true)

== 2. Schema yang kita definisikan sendiri ==
root
 |-- id: integer (nullable = false)
 |-- pelanggan: string (nullable = true)
 |-- kota: string (nullable = true)
 |-- produk: string (nullable = true)
 |-- qty: integer (nullable = true)
 |-- harga: long (nullable = true)

+---+---------+--------+------------+---+------+
| id|pelanggan|    kota|      produk|qty| harga|
+---+---------+--------+------------+---+------+
|  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|
+---+---------+--------+------------+---+------+

== 3. select dan filter berantai ==
+---------+------------+---+
|pelanggan|      produk|qty|
+---------+------------+---+
|     Budi|Kemeja Batik|  2|
|     Dewi|   Rok Tenun|  3|
|     Rina|Kemeja Batik|  2|
+---------+------------+---+

== 4. withColumn: kolom baru hasil hitung ==
+---+---------+--------+------------+---+------+------+
| id|pelanggan|    kota|      produk|qty| harga| total|
+---+---------+--------+------------+---+------+------+
|  1|     Budi| Bandung|Kemeja Batik|  2|240000|480000|
|  2|     Siti| Jakarta|Kain Songket|  1|850000|850000|
|  3|     Dewi|Surabaya|   Rok Tenun|  3|310000|930000|
|  4|     Budi| Bandung|Blouse Katun|  1|120000|120000|
|  5|     Rina| Bandung|Kemeja Batik|  2|240000|480000|
|  6|     Siti| Jakarta|   Rok Tenun|  1|310000|310000|
+---+---------+--------+------------+---+------+------+

== 5. groupBy: agregasi per kota ==
+--------+---------+-------+
|    kota|total_qty|  omzet|
+--------+---------+-------+
| Bandung|        5|1080000|
| Jakarta|        2|1160000|
|Surabaya|        3| 930000|
+--------+---------+-------+

Mari kita telaah baris-baris kode di atas:

1. Baris 25 (df_infer = ...): createDataFrame hanya diberi daftar nama kolom, sehingga Spark menebak tipe data — dan seperti terlihat pada output bagian 1, seluruh kolom angka menjadi long.
2. Baris 29-36 (skema = StructType([...])): kita menulis schema sendiri: id dan qty sebagai IntegerType (4 byte), harga sebagai LongType karena nilainya besar, dan argumen ketiga False pada id menandakan kolom tersebut tidak boleh NULL.
3. Baris 37-39 (df = ...): DataFrame dibuat dengan schema eksplisit; printSchema() pada output bagian 2 memperlihatkan id: integer (nullable = false) — persis seperti yang kita tulis.
4. Baris 42 (df.select(...).filter(...)): kedua operasi dirantai dalam satu ekspresi; hasilnya hanya tiga baris dengan qty > 1.
5. Baris 45 (df_total = df.withColumn(...)): withColumn menambah kolom total hasil qty * harga. Perhatikan bahwa hasilnya disimpan ke variabel baru — DataFrame Spark bersifat immutable, operasi apa pun mengembalikan DataFrame baru dan df yang lama tidak berubah.
6. Baris 49-52 (df_total.groupBy(...)): agregasi sum diberi nama yang jelas lewat alias(), dan orderBy("kota") membuat urutan baris output menjadi deterministik — tanpa orderBy, Spark tidak menjamin urutan hasil agregasi.

Schema: Inferensi versus Definisi #

Output bagian 1 dan 2 menunjukkan perbedaan nyata kedua pendekatan. Inferensi mempermudah eksplorasi cepat, tetapi Spark memilih tipe paling aman: seluruh angka Python menjadi long (8 byte) dan semua kolom dianggap bisa NULL. Schema eksplisit lebih awal ditulis tetapi memberi kendali penuh: kolom id cukup integer (4 byte) dan ditandai tidak nullable. Pada 500 juta baris event TokoKita, selisih 4 byte pada satu kolom berarti 2 GB — dikali sepuluh kolom numerik saja, selisihnya sudah 20 GB memori yang dihemat atau terbuang.

Operasi Inti: select, filter, dan withColumn #

Tiga operasi paling sering dipakai terlihat pada bagian 3 dan 4 output. select() memilih kolom yang dibawa ke hasil, filter() menyaring baris sesuai kondisi, dan withColumn() menambah satu kolom berisi hasil ekspresi — pada skrip kita, col("qty") * col("harga"). Fungsi col() mereferensikan kolom sebagai operan ekspresi, bukan sekadar mengambil isinya, sehingga perkalian dihitung di dalam Spark tanpa memindahkan data ke Python. Ketiganya adalah transformation: menyusun rencana tanpa memproses data, dan karena DataFrame bersifat immutable, hasilnya selalu DataFrame baru.

Agregasi dengan groupBy #

Pola groupBy().agg() menghasilkan satu baris per nilai unik kolom kunci. Di balik layar agregasi berjalan dua tahap: partial aggregate di tiap partition, lalu penggabungan hasil antar partition lewat shuffle — perpindahan data antar executor yang menjadi topik utama chapter Spark Partitioning. Dua kebiasaan pada baris 50-51 patut ditiru: alias() memberi nama kolom hasil agar mudah dibaca, dan orderBy() menjamin urutan baris konsisten antar eksekusi.

Spark DataFrame versus pandas #

Keduanya bukan saingan: untuk data sampai beberapa gigabyte, pandas di laptop sering kali lebih cepat dan lebih sederhana, dan keduanya bisa dipertemukan lewat spark_df.toPandas() setelah data diperkecil lewat filter atau agregasi. Panduan resmi lengkap operasi DataFrame ada di Getting Started Spark SQL . Sejauh ini kita menulis transformasi lewat metode berantai; chapter berikutnya memperkenalkan gaya kedua yang sama kuatnya — menulis query SQL langsung di atas DataFrame yang sama pada Spark SQL.