KAL
Menu

Watermark

Data terlambat, window dengan semantik waktu.

Daftar isi

Agregasi pada aliran data menghadapi soal yang tidak ada di dunia batch, yaitu data yang tiba terlambat. Sebuah perangkat yang mengirim suhu pukul 08:00 bisa saja pesannya sampai pukul 08:12 karena gangguan jaringan, dan pertanyaannya menjadi rumit, yaitu apakah window pukul 08:00 harus terus ditunggu. Watermark adalah jawaban Spark atas soal itu, yaitu garis batas waktu yang menyatakan seberapa lama data terlambat masih dilayani. Topik ini mendemonstrasikan semantiknya langsung pada data yang sengaja dibuat terlambat.

Window pada waktu kejadian#

Agregasi streaming hampir selalu dihitung di atas event time, yaitu waktu yang tertulis pada data itu sendiri, bukan waktu data tiba di mesin. Fungsi F.window("waktu", "5 minutes") mengelompokkan baris ke dalam window lima menit berdasarkan kolom waktu kejadian, dan hasilnya berupa kolom struct berisi awal dan akhir window. Dua kejadian pada menit yang sama selalu jatuh di window yang sama, berapa pun keterlambatan pengirimannya.

Perbedaan ini perlu dibedakan tegas dari processing time yang berjalan sesuai jam mesin. Bila agregasi dihitung di atas waktu pemrosesan, data yang tiba terlambat otomatis masuk window yang salah, dan hasilnya berubah setiap kali uji diulang. Agregasi berbasis waktu kejadian menjaga kebenaran hasil terhadap urutan kedatangan, dan kebenaran inilah yang menuntut adanya kebijakan tentang seberapa lama menunggu data terlambat.

Menetapkan watermark#

Fungsi withWatermark("waktu", "10 minutes") menyatakan dua hal sekaligus. Pertama, kolom waktu dipakai sebagai acuan waktu kejadian. Kedua, Spark menoleransi data terlambat sampai sepuluh menit dari kejadian terbaru yang sudah diproses. Setiap micro-batch, Spark memperbarui nilai maksimum waktu kejadian yang pernah terlihat, dan garis watermark bergerak maju menjadi maksimum itu dikurangi sepuluh menit.

Data yang waktu kejadiannya jatuh di bawah garis bergerak itu dinyatakan terlalu terlambat, window-nya sudah dianggap selesai, dan datanya dibuang tanpa pesan error. Keputusan ini adalah kompromi yang disengaja, yaitu hasil akhir dinyatakan final lebih awal dengan risiko kehilangan data yang sangat terlambat, alih-alih menahan seluruh state agregasi terbuka tanpa batas waktu. Nilai sepuluh menit dipilih sesuai kebutuhan bisnis, misalnya toleransi perangkat sensor yang memang sering terputus.

Berbeda dengan agregasi groupBy biasa, agregasi ber-window pada streaming menyimpan state per window di memori dan checkpoint. Watermark menjaga jumlah state tetap terkendali karena window yang sudah melewati garis dihapus dari state, dan inilah alasan Spark menuntut watermark sebelum mode append diizinkan pada agregasi.

Demonstrasi dua batch#

Skrip pertama menjalankan trigger availableNow atas folder berisi dua kejadian tepat waktu pada menit ke-1. Agregasi menghitung jumlah kejadian dan suhu maksimum per window lima menit, dengan mode keluaran update sehingga hanya window yang berubah yang tampil. Kueri beragregasi dengan state ini berdurasi dekat batas verifikasi otomatis lima belas detik dan waktunya bervariasi antar percobaan. Karena itu keluarannya diambil dari pengujian manual yang skripnya identik, sebagaimana pola pada topik UDF di modul transformasi.

python uji_watermark.py 
# tidak-dijalankan
import os
import shutil

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

shutil.rmtree("/tmp/sumber-wm", ignore_errors=True)
shutil.rmtree("/tmp/ckpt-wm", ignore_errors=True)
os.makedirs("/tmp/sumber-wm")
with open("/tmp/sumber-wm/batch1.txt", "w") as f:
    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-watermark")
    .config("spark.sql.shuffle.partitions", "1")
    .config("spark.sql.adaptive.enabled", "false")
    .getOrCreate()
)

df = (
    spark.readStream
    .format("csv")
    .schema("kamar string, waktu timestamp, suhu double")
    .load("/tmp/sumber-wm")
)
agg = (
    df.withWatermark("waktu", "10 minutes")
    .groupBy(F.window("waktu", "5 minutes"))
    .agg(F.count("*").alias("jumlah"), F.max("suhu").alias("suhu_max"))
)

q = (
    agg.writeStream
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "/tmp/ckpt-wm")
    .option("truncate", "false")
    .trigger(availableNow=True)
    .start()
)
q.awaitTermination()

spark.stop()
text Ilustrasi keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
-------------------------------------------
Batch: 0
-------------------------------------------
+------------------------------------------+------+--------+
|window                                    |jumlah|suhu_max|
+------------------------------------------+------+--------+
|{2026-11-04 08:00:00, 2026-11-04 08:05:00}|2     |30.0    |
+------------------------------------------+------+--------+

-------------------------------------------
Batch: 1
-------------------------------------------
+------+------+--------+
|window|jumlah|suhu_max|
+------+------+--------+
+------+------+--------+

Kedua kejadian jatuh pada window yang sama, yaitu lima menit berawal pukul 08:00, sehingga muncul satu baris dengan jumlah dua. Baris berlabel Batch: 1 yang kosong adalah micro-batch penutup pada trigger availableNow, yaitu langkah finalisasi state tanpa data baru, dan keluaran kosongnya bukan kejanggalan. Setelah batch ini, waktu kejadian terbesar yang tercatat adalah 08:01:40, sehingga garis watermark berdiri di 07:51:40, yaitu maksimum dikurangi toleransi sepuluh menit.

Skrip kedua menyimulasikan data terlambat. Skrip itu menulis dua kejadian berwaktu lampau ke folder sumber yang sama, lalu menjalankan ulang kueri yang identik dengan checkpoint yang sama, sebagaimana kueri produksi yang dihidupkan kembali. Skrip ini juga dieksekusi manual dengan checkpoint yang diwarisi dari skrip pertama, dan nomor batchnya menyusul hasil skrip pertama.

python lanjutan_watermark.py 
# tidak-dijalankan
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

with open("/tmp/sumber-wm/batch2.txt", "w") as f:
    f.write("kamar1,2026-11-04 07:58:00,35\n")
    f.write("kamar2,2026-11-04 07:40:00,36\n")

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

df = (
    spark.readStream
    .format("csv")
    .schema("kamar string, waktu timestamp, suhu double")
    .load("/tmp/sumber-wm")
)
agg = (
    df.withWatermark("waktu", "10 minutes")
    .groupBy(F.window("waktu", "5 minutes"))
    .agg(F.count("*").alias("jumlah"), F.max("suhu").alias("suhu_max"))
)

q = (
    agg.writeStream
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "/tmp/ckpt-wm")
    .option("truncate", "false")
    .trigger(availableNow=True)
    .start()
)
q.awaitTermination()

spark.stop()
text Ilustrasi keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
-------------------------------------------
Batch: 2
-------------------------------------------
+------------------------------------------+------+--------+
|window                                    |jumlah|suhu_max|
+------------------------------------------+------+--------+
|{2026-11-04 07:55:00, 2026-11-04 08:00:00}|1     |35.0    |
+------------------------------------------+------+--------+

Hasil batch ke-2 memperlihatkan kedua akibat bagi data terlambat dalam satu tabel. Kejadian pukul 07:58:00 berada di atas garis watermark 07:51:40, sehingga window 07:55 sampai 08:00 masih terbuka, diperbarui, dan tampil dengan jumlah satu serta suhu maksimum 35. Sebaliknya kejadian pukul 07:40:00 jatuh di bawah garis, window 07:35 sampai 07:40 sudah ditutup, dan baris itu dibuang tanpa jejak. Tidak ada baris baru untuk window lama di keluaran, dan tidak ada error yang muncul, itulah semantik watermark yang bekerja senyap.

Memilih toleransi dan mode keluaran#

Nilai toleransi watermark adalah keputusan bisnis yang berdampak langsung pada dua hal. Semakin besar toleransi, semakin banyak data terlambat yang tetap diperhitungkan, tetapi state agregasi bertahan lebih lama di memori dan hasil akhir tertunda lebih jauh. Semakin kecil toleransi, hasil lebih cepat final dengan risiko data nyata terbuang. Nilai ini layak dinyatakan eksplisit di kode beserta alasannya, karena pembaca skrip setahun kemudian tidak akan bisa menebak mengapa sepuluh menit dipilih.

Mode update dipakai pada demonstrasi karena menunjukkan perubahan per window secara langsung. Mode append pada agregasi ber-window akan menahan setiap window sampai watermark melewati akhir window, lalu menuliskan hasil finalnya satu kali, dan pola ini cocok ketika konsumen menuntut keluaran yang tidak berubah. Mode complete menulis seluruh state setiap potongan, praktis untuk data yang kecil dan tetap, tetapi biayanya berbanding lurus dengan ukuran state dan karenanya jarang cocok untuk aliran panjang.

Kesalahan umum#

Kesalahan pertama adalah menambahkan withWatermark setelah groupBy. Posisi pemanggilan menentukan makna, karena watermark harus dinyatakan pada kolom waktu sebelum agregasi disusun, dan penempatan yang salah memunculkan error atau semantik yang tidak diharapkan. Bentuk yang benar adalah df.withWatermark(...).groupBy(...) sebagaimana contoh di atas.

Kesalahan kedua adalah mengira watermark menyelamatkan semua data terlambat. Data yang melewati toleransi dibuang tanpa pesan, sehingga laporan bisa berbeda dari harapan tanpa alarm apa pun. Apabila data terlambat bernilai tinggi, jalankan salinan kedua kueri dengan toleransi besar atau simpan data mentah terlebih dulu agar penelusuran tetap mungkin.

Kesalahan ketiga adalah menjalankan ulang skrip kedua tanpa skrip pertama. Checkpoint menampung garis watermark dan posisi berkas, sehingga skrip kedua mengharapkan state dari skrip pertama. Pada praktik produksi, kueri yang dihidupkan ulang selalu meneruskan state melalui checkpoint, dan demonstrasi dua skrip di atas sengaja meniru pola tersebut.

Ringkasan#

  • Agregasi streaming dihitung di atas event time melalui F.window(), sehingga hasil tidak bergantung pada urutan kedatangan data.
  • withWatermark("waktu", "10 minutes") menyatakan toleransi data terlambat, dan garisnya bergerak maju mengikuti waktu kejadian terbesar yang sudah diproses.
  • Data di atas garis memperbarui window yang masih terbuka, sedangkan data di bawah garis dibuang tanpa pesan.
  • Mode update memperlihatkan perubahan per window, append menulis hasil final per window, dan pilihan toleransi adalah keputusan bisnis yang layak didokumentasikan.

Referensi#