Belajar Apache Spark - Spark Streaming & Structured Streaming

#Pendahuluan

Sejauh ini semua episode berfokus pada batch processing — data statis yang dibaca, diolah, lalu ditulis. Episode 10 ini membuka dimensi baru: streaming, memproses data yang terus mengalir masuk secara real-time. Dari log pengguna, telemetry IoT, hingga event transaksi, streaming adalah tulang punggung banyak sistem modern.

Spark menangani streaming dengan dua pendekatan: DStream yang lebih tua dan Structured Streaming yang modern. Keduanya berbeda mendasar, dan memahami perbedaannya mencegah kalian menulis pipeline dengan API yang sudah tidak disarankan.

Di episode ini kita akan membangun pipeline streaming dari nol, memahami cara mengendalikan kapan data diproses dan berapa lama data yang belum lengkap ditunggu, serta mengarahkan hasil ke berbagai sink termasuk Kafka.

#DStream vs Structured Streaming

#DStream: Pendekatan Klasik

DStream (Discretized Stream) adalah abstraksi streaming paling awal di Spark, dibangun di atas RDD. Data dibagi menjadi micro-batch berupa RDD yang dieksekusi secara terpisah. API-nya berbasis transformasi RDD dan diakses lewat streamingContext.

DStream masih berfungsi, tapi sudah berada dalam mode maintenance. Tidak ada optimasi Catalyst, integrasi dengan DataFrame terbatas, dan tidak ada dukungan event-time natively. Untuk proyek baru, komunitas merekomendasikan Structured Streaming.

#Structured Streaming: Modern dan Declarative

Structured Streaming memperlakukan data streaming sebagai tabel tak berujung (unbounded table). Setiap batch data baru adalah baris baru yang ditambahkan ke tabel tersebut. Kalian menulis query yang sama seperti batch processing, dan Spark mengulanginya secara inkremental.

DStream            → aliran RDD, API RDD
Structured Streaming → tabel tak berujung, API DataFrame/SQL

Keunggulan utamanya: satu kode untuk batch dan streaming, optimasi Catalyst, event-time windowing, dan exactly-once semantics dengan sink yang mendukung. Inilah alasan Structured Streaming menjadi pilihan utama untuk proyek baru.

#Membangun Pipeline Streaming

#Menginisialisasi Session dan ReadStream

Pipeline streaming dimulai dengan SparkSession biasa, lalu membaca sumber lewat readStream:

Membaca aliran dari Kafka

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
 
spark = SparkSession.builder \
    .appName("streaming-kafka") \
    .config("spark.sql.shuffle.partitions", "8") \
    .getOrCreate()
 
kafka = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "klik-pengguna") \
    .load()

spark.readStream.format("kafka") membaca record Kafka. Data mentah berada di kolom value sebagai biner, jadi perlu di-decode menjadi string terlebih dahulu.

#Transformasi dan WriteStream

Decode data lalu jalankan agregasi window:

Agregasi window dan writeStream

from pyspark.sql import functions as F
 
klik = kafka.select(
    F.from_json(F.col("value").cast("string"),
                F.schema_of_json('{"user_id":"1","waktu":"2026-08-10T10:00:00Z","halaman":"/beranda"}'))
    .alias("data")
).select("data.*")
 
per_menit = klik \
    .groupBy(F.window("waktu", "1 minute"), "halaman") \
    .count()
 
query = per_menit.writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

per_menit.writeStream memulai query streaming yang berjalan terus-menerus. outputMode("update") hanya mengeluarkan baris yang berubah. Query berjalan di background; kalian bisa memanggil query.awaitTermination() untuk memblokir sampai selesai.

#Trigger Modes, Watermarking, dan Output Modes

#Trigger Modes

Trigger menentukan kapan micro-batch dijalankan:

• Default: secepat mungkin, setiap batch selesai langsung batch berikutnya.
• Processing time: interval tetap, misalnya 5 seconds.
• Once: satu batch saja, lalu berhenti — berguna untuk one-shot job.
• Continuous: latensi rendah dengan sink tertentu, eksperimental.

Trigger processing time 5 detik

per_menit.writeStream \
    .trigger(processingTime="5 seconds") \
    .format("console") \
    .start()

#Watermarking

Watermark adalah ambang keterlambatan data. Karena data bisa datang terlambat (misalnya telemetry yang tertunda di jaringan), watermark memberi tahu Spark berapa lama menunggu sebelum menganggap event tertentu sudah selesai:

Watermark 10 menit

per_menit = klik \
    .withWatermark("waktu", "10 minutes") \
    .groupBy(F.window("waktu", "1 minute"), "halaman") \
    .count()

withWatermark("waktu", "10 minutes") membuat agregasi hanya memperhitungkan event hingga 10 menit setelah jendela berakhir. Data yang terlambat melewati watermark dibuang — sehingga state yang disimpan tidak membesar tanpa batas.

#Output Modes

Tiga output mode yang perlu dipahami:

• append: hanya baris baru yang ditambahkan — cocok untuk windowing dengan watermark.
• update: baris yang berubah dikeluarkan kembali — cocok untuk agregasi.
• complete: seluruh hasil ditulis ulang setiap batch — cocok untuk agregasi tanpa watermark.

append   → hasil akhir (dengan watermark), sink file
update   → hasil yang berubah, sink console
complete → seluruh hasil agregasi, ukuran state kecil

#Sink ke Kafka, Files, Console, dan Storage

#Sink Console untuk Debug

format("console") adalah cara tercepat melihat hasil streaming di terminal. Cocok untuk pengembangan, bukan produksi.

#Sink File untuk Data Lake

Menulis ke Parquet memakai sink file dengan mode append dan path:

Sink ke Parquet

per_menit.writeStream \
    .outputMode("append") \
    .format("parquet") \
    .option("path", "data/hasil_streaming") \
    .option("checkpointLocation", "data/checkpoint") \
    .start()

Parameter checkpointLocation wajib — di sinilah Spark menyimpan metadata posisi pemrosesan agar bisa melanjutkan (recovery) saat gagal. Detail checkpoint dibahas mendalam di episode 16.

#Sink Kafka

Untuk meneruskan hasil ke sistem lain, tulis kembali ke Kafka:

Sink kembali ke Kafka

hasil.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "hasil-analitik") \
    .option("checkpointLocation", "data/checkpoint-kafka") \
    .start()

Kolom value harus berupa biner, jadi jangan lupa mengkonversi hasil ke string lalu encode sebelum menulis.

#Penutup

Episode 10 membuka pintu ke real-time processing: kalian memahami mengapa Structured Streaming menggantikan DStream, membangun pipeline dengan readStream dan writeStream, mengendalikan eksekusi lewat trigger modes, membatasi data terlambat dengan watermarking, dan mengarahkan hasil ke console, file, maupun Kafka.

Inti yang harus dibawa pulang:

• Structured Streaming memperlakukan stream sebagai tabel tak berujung.
• readStream membaca sumber; writeStream menulis hasil dan menjalankan query.
• Watermark mencegah state membesar dengan data yang terlambat.
• Pilih output mode sesuai jenis agregasi: append, update, atau complete.
• checkpointLocation wajib untuk recovery dan fault tolerance.

Di episode 11 selanjutnya kita akan membahas machine learning dengan MLlib — pipeline API dengan transformers dan estimators, feature engineering, training dan tuning model, serta use cases klasifikasi, regresi, clustering, dan recommendation.