Chapter Windowing menutup dengan pertanyaan yang belum terjawab: kapan sebuah window boleh dinyatakan selesai? Menutup terlalu cepat membuat pesanan dari ponsel yang sempat offline tertinggal; menunggu terlalu lama membuat dashboard basi dan state agregasi membengkak di memori. Watermark menjawab kedua kekhawatiran itu: penanda waktu yang terus maju mengikuti data dan menyatakan batas keterlambatan yang ditoleransi. Kita bedah mekanismenya lewat simulasi Python murni, bukti eksekusi Spark, lalu cara memilih nilai allowed lateness.
Watermark: Batas Waktu yang Bergerak #
Watermark dihitung dari data itu sendiri: watermark = event time maksimum yang sudah terpantau − allowed lateness. Sebuah late event — event yang tiba setelah event lain yang lebih baru — masih diterima selama event time-nya berada di atas watermark; begitu jatuh di bawah watermark, event dibuang. Watermark hanya bergerak maju, tidak pernah mundur, karena event time maksimum tidak pernah mengecil.
Watermark sekaligus menjadi pemicu penutupan window: saat watermark melewati akhir sebuah window, window dinyatakan final, hasilnya diteruskan ke sink, dan state-nya dibebaskan. Watermark adalah pernyataan optimis — “semua event dengan event time di bawah titik ini sudah tiba” — dan allowed lateness menentukan seberapa besar ruang aman untuk asumsi tersebut.
Simulasi: Mesin Watermark Murni Python #
Simulasi berikut meniadakan Spark agar aturannya terlihat telanjang: sepuluh checkout TokoKita diproses satu per satu, sebagian dari web yang tepat waktu dan sebagian dari aplikasi mobile yang tertinggal beberapa detik, di atas window tumbling 10 detik dengan allowed lateness 5 detik.
"""Simulasi murni-Python: aturan watermark = max event time - allowed lateness."""
TOLERANSI = 5 # allowed lateness (detik)
DURASI = 10 # durasi window tumbling (detik)
# (event_time, waktu_tiba, sumber) -- checkout TokoKita, waktu dalam detik
events = [
(12, 12, "web"),
(20, 20, "web"),
(27, 27, "web"),
(13, 28, "mobile"),
(24, 29, "web"),
(33, 30, "web"),
(22, 31, "mobile"),
(31, 32, "web"),
(25, 33, "mobile"),
(38, 34, "web"),
]
def window_mulai(et):
return et // DURASI * DURASI
maks_et = None # event time tertinggi yang sudah dilihat
status = {} # window_start -> jumlah baris terhitung
ditutup = set()
diterima = 0
dibuang = 0
for et, tiba, sumber in events:
# watermark sebelum batch ini = (max event time s.d. batch lalu) - toleransi.
# Sebelum event pertama, Spark memakai watermark awal bernilai epoch (0).
wm = (maks_et - TOLERANSI) if maks_et is not None else 0
awal = window_mulai(et)
if et < wm:
dibuang += 1
print(f"tiba={tiba:>3} et={et:>3} {sumber:>6} | wm={wm:>3} "
f"-> DIBUANG (et {wm - et} detik di bawah watermark)")
else:
diterima += 1
status[awal] = status.get(awal, 0) + 1
print(f"tiba={tiba:>3} et={et:>3} {sumber:>6} | wm={wm:>3} "
f"-> diterima, window [{awal},{awal + DURASI}) = {status[awal]}")
maks_et = et if maks_et is None else max(maks_et, et)
wm = maks_et - TOLERANSI
for w0 in sorted(status):
if w0 not in ditutup and wm >= w0 + DURASI:
ditutup.add(w0)
print(f" >> watermark {wm} menyentuh akhir window "
f"[{w0},{w0 + DURASI}) -> window ditutup, "
f"hitungan final {status[w0]}")
print()
print("ringkasan window:")
for w0 in sorted(status):
keadaan = "ditutup" if w0 in ditutup else "masih terbuka"
print(f" [{w0:>2},{w0 + DURASI:>2}) hitungan={status[w0]:>2} {keadaan}")
print(f"diterima={diterima} dibuang={dibuang} dari {len(events)} event")Output yang Dihasilkan:
tiba= 12 et= 12 web | wm= 0 -> diterima, window [10,20) = 1
tiba= 20 et= 20 web | wm= 7 -> diterima, window [20,30) = 1
tiba= 27 et= 27 web | wm= 15 -> diterima, window [20,30) = 2
>> watermark 22 menyentuh akhir window [10,20) -> window ditutup, hitungan final 1
tiba= 28 et= 13 mobile | wm= 22 -> DIBUANG (et 9 detik di bawah watermark)
tiba= 29 et= 24 web | wm= 22 -> diterima, window [20,30) = 3
tiba= 30 et= 33 web | wm= 22 -> diterima, window [30,40) = 1
tiba= 31 et= 22 mobile | wm= 28 -> DIBUANG (et 6 detik di bawah watermark)
tiba= 32 et= 31 web | wm= 28 -> diterima, window [30,40) = 2
tiba= 33 et= 25 mobile | wm= 28 -> DIBUANG (et 3 detik di bawah watermark)
tiba= 34 et= 38 web | wm= 28 -> diterima, window [30,40) = 3
>> watermark 33 menyentuh akhir window [20,30) -> window ditutup, hitungan final 3
ringkasan window:
[10,20) hitungan= 1 ditutup
[20,30) hitungan= 3 ditutup
[30,40) hitungan= 3 masih terbuka
diterima=7 dibuang=3 dari 10 eventMari kita telaah baris-baris kode di atas:
- Baris 6-17 (
events = [...]): setiap event membawa tiga fakta — event time, waktu tiba, dan sumber. Eventet=13yang tiba di detik ke-28 adalah late event 15 detik. - Baris 33 (
wm = ...): watermark yang menentukan nasib event dihitung dari event time maksimum sebelumnya, meniru engine yang hanya mengenal masa lalu saat memfilter baris baru. - Baris 35-43 (
if et < wm ... else): baris di bawah watermark langsung dibuang dan dihitung sebagaidibuang; baris yang lolos menambah hitungan window miliknya — termasuk late event yang masih di atas watermark dan memperbarui window yang belum ditutup. - Baris 44-51 (
maks_et = ...sampaiprint(f" >> ...): event time maksimum diperbarui, lalu setiap window yang akhirnya terlampaui watermark baru ditandaiditutupsekali dan tidak pernah dibuka ulang.
Output tersebut memperlihatkan tiga nasib berbeda. Event et=24 yang tiba di detik ke-29 terlambat dibanding tetangganya, tetapi event time 24 masih di atas watermark 22, sehingga hitungan window [20,30) naik dari 2 menjadi 3 — late event yang masih sempat memperbarui hasil. Sebaliknya, et=13 yang tiba di detik ke-28 sudah 9 detik di bawah watermark dan window [10,20)-nya sudah ditutup: event dibuang tanpa dihitung. Penutupan itu sendiri dipicu watermark, bukan jam dinding — window [20,30) baru ditutup saat watermark mencapai 33 karena watermark melompat mengikuti data dan tidak pernah berhenti tepat di angka 30.
Bukti di Spark: Event Mobile yang Terbuang #
Aturan yang sama berlaku di Spark Structured Streaming lewat satu baris withWatermark. Skrip berikut membelah checkout menjadi dua sumber: baris web yang tepat waktu, dan baris aplikasi mobile yang event time-nya ditahan 40 detik, meniru ponsel yang menyimpan event lokal saat offline. Watermark diset hanya 5 detik.
"""Watermark demo v2: web on-time vs mobile tertahan 40 detik, breakdown sumber."""
import time
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")
t0 = time.time()
# Baris genap = checkout web (event_time = waktu kedatangan).
# Baris ganjil = checkout aplikasi mobile yang tertahan 40 detik
# (event_time 40 detik di belakang waktu kedatangan).
checkout = (spark.readStream.format("rate")
.option("rowsPerSecond", 10).option("numPartitions", 1)
.load()
.withColumn("sumber", F.when(F.col("value") % 2 == 0, "web").otherwise("mobile"))
.withColumn("event_time",
(F.unix_timestamp("timestamp")
- F.when(F.col("value") % 2 == 1,
F.lit(40)).otherwise(F.lit(0))
).cast("timestamp")))
ringkasan = (checkout.withWatermark("event_time", "5 seconds")
.groupBy(F.window("event_time", "6 seconds"), "sumber").count())
q = (ringkasan.writeStream.format("console").outputMode("update")
.trigger(processingTime="3 seconds").option("truncate", "false").start())
q.awaitTermination(45)
q.stop()
print(f"durasi script: {time.time() - t0:.1f} detik")
spark.stop()Output yang Dihasilkan (Batch 0 kosong, Batch 2 terakhir memperbarui hitungan mobile menjadi 20, dan Batch 4 sampai 15 hanya berisi baris web berulang; ketiganya dipangkas agar ringkas):
-------------------------------------------
Batch: 1
-------------------------------------------
+------------------------------------------+------+-----+
|window |sumber|count|
+------------------------------------------+------+-----+
|{2026-10-16 14:02:30, 2026-10-16 14:02:36}|web |8 |
|{2026-10-16 14:01:54, 2026-10-16 14:02:00}|mobile|10 |
|{2026-10-16 14:02:36, 2026-10-16 14:02:42}|web |2 |
+------------------------------------------+------+-----+
-------------------------------------------
Batch: 3
-------------------------------------------
+------------------------------------------+------+-----+
|window |sumber|count|
+------------------------------------------+------+-----+
|{2026-10-16 14:02:36, 2026-10-16 14:02:42}|web |27 |
+------------------------------------------+------+-----+
durasi script: 47.6 detikTiga bagian skrip menentukan hasilnya. Bagian readStream.format("rate") menghasilkan 10 baris per detik dengan event_time mobile digeser 40 detik ke belakang. Bagian withWatermark("event_time", "5 seconds") mendeklarasikan batas keterlambatan, lalu groupBy per window dan per sumber memisahkan nasib web dan mobile; output mode update mencetak baris yang berubah setiap batch.
Baris mobile hanya muncul pada dua batch pertama, saat watermark masih bernilai awal. Begitu baris web mendorong event time maksimum maju, setiap event mobile jatuh di bawah watermark 5 detik dan dibuang tanpa pesan galat: hitungan mobile berhenti di 20 padahal event terus datang. Timestamp window memakai jam dinding eksekusi sehingga berbeda antar run; polanya stabil.
Memilih Nilai allowed lateness #
Toleransi terlalu kecil membuang event sah seperti checkout ponsel offline; terlalu besar menahan penutupan window, menumpukkan state di memori, dan menunda hasil pada output mode append. Praktik yang sehat: ukur distribusi keterlambatan sumber secara empiris, lalu set allowed lateness di sekitar persentil tinggi (misalnya p99). Rincian resmi mekanisme watermark terdokumentasi di Structured Streaming Programming Guide
.
Watermark tidak hanya mengatur penutupan window — komputasi yang menggabungkan dua stream sekaligus juga bergantung padanya untuk membatasi state. Bagaimana join antar data yang mengalir dengan kecepatan berbeda tetap bisa selesai, itu pokok bahasan Chapter Stream Joins.