KAL
Menu

Structured Streaming

readStream, transformasi, writeStream, trigger.

Daftar isi

Topik ini merakit pipeline streaming pertama yang utuh, dari source berkas sampai sink konsol. Polanya terdiri dari empat langkah yang berulang di hampir semua pekerjaan streaming, yaitu mendefinisikan sumber, mentransformasikan, menentukan sink beserta mode keluaran, lalu menetapkan trigger. Setelah topik ini, sebuah kueri streaming dapat dijalankan dan dihentikan secara terkendali, dan hasilnya dapat dibaca kembali.

Anatomi kueri streaming#

Kueri streaming disusun dengan rantai panggilan yang terlihat panjang tetapi jelas urutannya. spark.readStream membuka pembaca aliran, .format() dan .schema() menetapkan sumber, transformasi mengalir seperti pada DataFrame batch, lalu writeStream menyiapkan penulisan. Ketika start() dipanggil, kueri benar-benar berjalan di latar belakang, dan awaitTermination() membuat skrip menunggu sampai kueri selesai atau dihentikan.

Mode keluaran menentukan apa yang ditulis ke sink setiap potongan selesai. Mode append menulis hanya baris baru yang masuk, cocok untuk transformasi tanpa agregasi. Mode update menulis hanya baris agregasi yang berubah, sedangkan mode complete menulis seluruh hasil agregasi setiap kali ada perubahan. Untuk pipeline penyaringan sederhana seperti contoh berikut, append adalah pilihan yang alami.

Skrip berikut memantau folder berisi data suhu kamar, menyaring baris dengan suhu minimal 30, lalu menampilkan hasilnya di konsol. Berkas sumber dituliskan lebih dulu oleh skrip itu sendiri agar eksperimen dapat diulang dari kondisi yang bersih.

python uji_stream.py 
import os
import shutil

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

shutil.rmtree("/tmp/sumber-suhu", ignore_errors=True)
shutil.rmtree("/tmp/ckpt-suhu", ignore_errors=True)
os.makedirs("/tmp/sumber-suhu")
with open("/tmp/sumber-suhu/batch1.txt", "w") as f:
    f.write("kamar1,2026-11-04 08:00:10,28\n")
    f.write("kamar2,2026-11-04 08:00:40,31\n")
    f.write("kamar1,2026-11-04 08:01:10,30\n")
    f.write("kamar2,2026-11-04 08:01:40,29\n")

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("uji-stream")
    .config("spark.sql.shuffle.partitions", "2")
    .config("spark.sql.adaptive.enabled", "false")
    .getOrCreate()
)

raw = (
    spark.readStream
    .format("csv")
    .schema("kamar string, waktu timestamp, suhu double")
    .load("/tmp/sumber-suhu")
)
panas = raw.filter(col("suhu") >= 30)

q = (
    panas.writeStream
    .outputMode("append")
    .format("console")
    .option("checkpointLocation", "/tmp/ckpt-suhu")
    .option("truncate", "false")
    .trigger(availableNow=True)
    .start()
)
q.awaitTermination()
print("STATUS =", q.status["message"])

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
-------------------------------------------
Batch: 0
-------------------------------------------
+------+-------------------+----+
|kamar |waktu              |suhu|
+------+-------------------+----+
|kamar2|2026-11-04 08:00:40|31.0|
|kamar1|2026-11-04 08:01:10|30.0|
+------+-------------------+----+

STATUS = Stopped

Keluaran memperlihatkan satu micro-batch bernomor 0 berisi dua baris yang lolos penyaringan, yaitu suhu 31 dan 30, sementara baris 28 dan 29 tertahan. Tabel konsol dicetak oleh sink setiap micro-batch selesai, dan nomor batch bertambah untuk setiap potongan berikutnya. Baris terakhir berasal dari q.status["message"] yang dibaca setelah awaitTermination() kembali, dan status “Stopped” menandakan kueri berakhir sesuai rencana karena trigger availableNow memang selesai setelah data habis. Baris yang tampil bukan seluruh isi folder, melainkan hanya keluaran transformasi filter yang dinyatakan di atas.

Trigger availableNow dan siklus micro-batch#

Trigger menentukan kapan potongan berikutnya dijalankan. Tanpa trigger, kueri berjalan terus dengan potongan berulang setiap sekian detik. Trigger availableNow berbeda, yaitu memproses seluruh data yang tersedia saat kueri dimulai sampai habis, lalu kueri berhenti dengan sendirinya. Pola ini dipakai pada seluruh contoh modul ini karena hasilnya dapat direkam dan diulang persis, sekaligus meniru pola produksi yang menjadwalkan pemrosesan berkala alih-alih menjaga kueri hidup sepanjang waktu.

Pada keluaran di atas, perhatikan kolom waktu yang berisi 08:00:40 dan 08:01:10. Schema yang dideklarasikan sudah menyebut tipe timestamp, sehingga teks berkas diurai menjadi nilai waktu sejak awal dan transformasi berbasis waktu siap dipakai tanpa konversi tambahan. Manfaat menyatakan schema di awal terasa justru di sini, karena pemilihan tipe yang benar membuat data selanjutnya bebas dari pembersihan tipe.

Selain availableNow, trigger lain yang sering dipakai adalah processingTime seperti processingTime="30 seconds" untuk pemrosesan berkala, dan once yang kini digantikan oleh availableNow. Pemilihan trigger tidak mengubah kebenaran hasil, melainkan hanya mengatur ritme potongan, karena kebenaran dijaga oleh checkpoint yang dibahas pada topik terakhir modul ini.

Transformasi streaming sama dengan batch#

Seluruh transformasi DataFrame yang dipelajari sebelumnya bekerja pada DataFrame streaming. Penyaringan, pemilihan kolom, pemetaan, bahkan agregasi groupBy tersedia dengan makna yang sama. Perbedaan muncul ketika agregasi terlibat, karena hasil agregasi harus dipertahankan antar potongan, dan di situlah konsep watermark pada topik berikutnya berperan.

Skrip di atas juga memperlihatkan kelas StreamingQuery yang dikembalikan oleh start(). Objek q mengizinkan pemeriksaan status melalui q.status, penghentian dengan q.stop(), dan menunggu selesai dengan q.awaitTermination(). Pada pola availableNow, awaitTermination() akan kembali setelah seluruh data diproses, sehingga skrip bisa melanjutkan ke langkah berikutnya secara terkendali.

Kesalahan umum#

Kesalahan pertama adalah memanggil start() tanpa checkpointLocation. Kueri streaming menolak berjalan karena titik pemulihan tidak diberikan, dan ini disengaja oleh Spark untuk mencegah kueri kehilangan posisi data. Perbaikannya adalah menyediakan folder checkpoint kosong pada kueri baru dan memakainya secara konsisten untuk kueri yang sama.

Kesalahan kedua adalah menjalankan ulang skrip dengan checkpoint lama tanpa membersihkan folder sumber. Berkas yang sudah dicatat pada checkpoint tidak diproses ulang, sehingga keluaran tampak kosong padahal skrip dijalankan tanpa error. Ketika eksperimen ingin diulang dari awal, bersihkan source, sink, dan checkpoint bersamaan, seperti yang dilakukan di awal skrip contoh.

Kesalahan ketiga adalah memakai mode append pada agregasi tanpa watermark. Spark menolak karena keluaran append menuntut kepastian bahwa sebuah baris tidak akan berubah lagi, dan kepastian itu hanya bisa diberikan oleh watermark. Perbaikannya adalah menambahkan withWatermark atau sementara memakai mode update saat eksplorasi.

Ringkasan#

  • Kueri streaming tersusun dari readStream, transformasi, writeStream dengan mode keluaran, lalu trigger, dan start() menjalankannya di latar belakang.
  • Mode append menulis baris baru, update menulis baris agregasi yang berubah, dan complete menulis seluruh hasil.
  • Trigger availableNow memproses seluruh data yang tersedia lalu berhenti, dan polanya nyaman untuk belajar serta penjadwalan berkala.
  • checkpointLocation wajib pada setiap kueri streaming dan menjadi fondasi jaminan pengiriman.

Referensi#