Spark Structured Streaming

Laporan penjualan harian TokoKita selalu basi satu hari: baru besok pagi Budi tahu bahwa kemarin flash sale Rok Tenun habis di jam kedua. Untuk keputusan yang butuh angka menit ini — stok menipis, antrean checkout melonjak, harga promo berjalan — kita butuh pemrosesan yang tidak menunggu data lengkap terkumpul. Structured Streaming menjawab kebutuhan itu dengan ide yang elegan: stream diperlakukan sebagai tabel yang terus bertambah barisnya, dan query yang sama seperti batch dijalankan ulang secara berkala atas potongan tabel terbaru. Pada chapter ini kita menjalankan demo nyata: menghitung kunjungan per window waktu dari sumber data yang mengalir.

Model Micro-Batch: Tabel yang Terus Bertambah #

Cara kerja Structured Streaming disebut micro-batch. Setiap interval trigger (misalnya setiap 4 detik), Spark mengambil baris baru sejak trigger terakhir, menjalankan query yang sama seperti query batch di atasnya, lalu menuliskan hasil yang berubah ke sink. Bagi kode DataFrame kita, tidak ada perbedaan sintaks besar — groupBy, agg, join tetap berlaku; yang berubah adalah pembacaannya memakai spark.readStream dan penulisannya memakai writeStream. Konsekuensi model ini penting dipahami: hasil tidak pernah “final” pada satu titik waktu, melainkan diperbarui sepanjang stream berjalan, sehingga kita harus eksplisit memilih output mode dan cara menangani event yang datang terlambat.

Demo: rate Source, Window, dan Watermark #

Sumber rate adalah generator baris bawaan Spark — cocok untuk belajar karena tidak butuh Kafka atau server lain. Skrip berikut membaca 5 baris per detik, menghitung jumlah baris per window 8 detik berdasarkan event_time, dan mencetak hasilnya ke console. event_time sengaja digeser ke belakang 0-2 detik untuk meniru event yang datang terlambat dari sumbernya:

"""Micro-batch Structured Streaming: rate source, window, dan watermark."""
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

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

# Sumber "rate" menghasilkan baris (timestamp, value) 5 baris per detik.
# event_time digeser ke belakang 0, 1, atau 2 detik meniru event yang
# datang terlambat dari sumbernya (misalnya jaringan lambat).
kunjungan = (spark.readStream
             .format("rate")
             .option("rowsPerSecond", 5)
             .option("numPartitions", 1)
             .load()
             .withColumn(
                 "event_time",
                 (F.unix_timestamp("timestamp") - (F.col("value") % 3))
                 .cast("timestamp")))

ringkasan = (kunjungan
             .withWatermark("event_time", "4 seconds")
             .groupBy(F.window("event_time", "8 seconds"))
             .count())

q = (ringkasan.writeStream
     .format("console")
     .outputMode("append")
     .trigger(processingTime="4 seconds")
     .option("truncate", "false")
     .start())
q.awaitTermination(48)
q.stop()
print(f"status kueri setelah stop: {q.status['message']}")

spark.stop()

Output yang Dihasilkan (Batch 1-4, 6, 8, dan 10 mencetak tabel kosong seperti Batch 0 dan 12, dipangkas agar ringkas):

-------------------------------------------
Batch: 0
-------------------------------------------
+------+-----+
|window|count|
+------+-----+
+------+-----+

-------------------------------------------
Batch: 5
-------------------------------------------
+------------------------------------------+-----+
|window                                    |count|
+------------------------------------------+-----+
|{2026-10-16 13:29:20, 2026-10-16 13:29:28}|26   |
+------------------------------------------+-----+

-------------------------------------------
Batch: 7
-------------------------------------------
+------------------------------------------+-----+
|window                                    |count|
+------------------------------------------+-----+
|{2026-10-16 13:29:28, 2026-10-16 13:29:36}|40   |
+------------------------------------------+-----+

-------------------------------------------
Batch: 9
-------------------------------------------
+------------------------------------------+-----+
|window                                    |count|
+------------------------------------------+-----+
|{2026-10-16 13:29:36, 2026-10-16 13:29:44}|40   |
+------------------------------------------+-----+

-------------------------------------------
Batch: 11
-------------------------------------------
+------------------------------------------+-----+
|window                                    |count|
+------------------------------------------+-----+
|{2026-10-16 13:29:44, 2026-10-16 13:29:52}|40   |
+------------------------------------------+-----+

-------------------------------------------
Batch: 12
-------------------------------------------
+------+-----+
|window|count|
+------+-----+
+------+-----+

status kueri setelah stop: Stopped

Mari kita telaah baris-baris kode di atas:

1. Baris 15-19 (spark.readStream.format("rate")): membuka sumber streaming; opsi rowsPerSecond=5 mengatur laju data, numPartitions=1 menjaga sumber tetap satu partition agar output mudah dibaca.
2. Baris 20-23 (withColumn("event_time", ...)): kolom timestamp bawaan rate source adalah waktu pemrosesan; kita menurunkan event_time darinya dengan geser mundur (value % 3) detik agar demo memiliki event terlambat.
3. Baris 26 (withWatermark("event_time", "4 seconds")): mendeklarasikan batas keterlambatan — event yang event_time-nya lebih tua dari watermark akan dibuang, dan window yang sudah lewat dinyatakan selesai.
4. Baris 27 (F.window("event_time", "8 seconds")): agregasi per window waktu 8 detik; setiap baris masuk ke window [t, t+8) sesuai event_time-nya, bukan waktu kedatangannya.
5. Baris 30-35 (writeStream): sink console dengan output mode append, trigger setiap 4 detik, dan truncate=false agar kolom window tidak dipotong.
6. Baris 36 (awaitTermination(48)): membiarkan kueri berjalan 48 detik sebelum q.stop() menghentikannya dengan rapi.

Membaca Output: Kenapa Banyak Batch Kosong #

Pola output di atas adalah pelajaran terpenting demo ini. Trigger menyala setiap 4 detik, tetapi window berdurasi 8 detik baru boleh dilaporkan pada output mode append setelah watermark melewati akhir window — yaitu setelah Spark melihat event dengan event_time 12 detik melewati awal window (8 detik durasi + 4 detik watermark). Akibatnya window muncul kira-kira setiap dua batch, dan batch di antaranya mencetak tabel kosong. Window pertama berisi 26 baris karena kueri dimulai di tengah window (window berikutnya penuh: 40 baris = 5 baris/detik × 8 detik). Dua hal perlu diungkap jujur: timestamp window adalah jam dinding saat eksekusi sehingga berbeda antar run, dan jumlah baris window pertama bergeser sedikit tergantung kapan kueri dimulai. Window penuh stabil di angka 40.

Output Mode dan Sink #

Output mode menentukan baris apa yang ditulis ke sink pada setiap batch, dan pilihan kita dibatasi oleh jenis agregasi:

Sink yang sering dipakai: console untuk debugging (seperti demo ini), file untuk menulis Parquet bertahap ke direktori, kafka untuk meneruskan hasil ke topic, dan foreachBatch untuk logika tulis kustom apa pun di atas micro-batch. Satu praktik wajib produksi adalah checkpointLocation: direktori yang menyimpan offset dan state agregasi sehingga kueri bisa lanjut dari titik terakhir saat aplikasi direstart. Panduan lengkap seluruh operasi yang didukung terdokumentasi di Structured Streaming Programming Guide .

Demo ini berjalan pada satu mesin dengan satu core. Stream sungguhan dari ribuan event per detik menuntut hal yang sama seperti batch besar: pekerjaan yang dipecah dan disebar ke banyak executor — mekanisme yang kita bedah pada chapter Distributed Data Processing.