Scoring fraud tidak bisa menunggu job malam. Ketika transaksi tiba, model membutuhkan fitur seperti “berapa transaksi pengguna ini dalam 60 detik terakhir” pada detik itu juga, sehingga fitur harus dihitung sambil event mengalir — bukan dibaca dari snapshot kemarin. Kecepatan ini datang dengan tagihan: event datang tidak berurutan, state harus dirawat di memori, dan pada suatu titik sistem harus memutuskan berhenti menunggu event terlambat. Chapter ini membangun rolling counter sungguhan dengan state dan watermark, lalu membandingkan hasilnya terhadap hitungan batch untuk mengukur harga kecepatan tersebut secara jujur.
State, Event Time, dan Watermark #
Tiga konsep dari section Streaming Data menjadi fondasi di sini. Pertama, state: fitur rolling window menuntut sistem mengingat event yang sudah lewat — untuk setiap pengguna, kita menyimpan daftar event yang diterima. Kedua, event time versus processing time — dibedah tuntas di chapter Event Time vs Processing Time — setiap event membawa dua jam: kapan kejadian terjadi (et) dan kapan event tiba di sistem (tiba). Ketiga, watermark: sistem mengumumkan “saya tidak akan menunggu event dengan event time lebih lama dari ini lagi”, dan event yang tiba setelah ambang tersebut dibuang — mekanisme yang sama dengan Watermarks dan Late Events.
Simulasi berikut memakai 14 event checkout dengan waktu dalam detik sejak pukul 12:00; dua event dari aplikasi mobile datang terlambat. Blok memotong daftar EVENTS dan menampilkan inti perhitungannya:
LEBAR = 60 # detik: lebar rolling window
TOLERANSI = 5 # detik: allowed lateness -> watermark = max event time - 5
state = {} # user -> daftar (event_time, nilai) yang diterima
maks_et = None # event time maksimum yang sudah terpantau
diterima = dibuang = 0
for et, tiba, user, nilai in EVENTS:
wm = (maks_et - TOLERANSI) if maks_et is not None else 0
if et < wm:
dibuang += 1
print(f"tiba={tiba:>3} et={et:>3} {user:<5} nilai={nilai:>7,} | wm={wm:>3}"
f" -> DIBUANG (terlalu terlambat untuk window {LEBAR}s)")
else:
diterima += 1
state.setdefault(user, []).append((et, nilai))
# fitur online: hitung ulang rolling window user ini setelah event diterima
isi = [(t, v) for t, v in state[user] if et - LEBAR < t <= et]
fitur_count = len(isi)
fitur_sum = sum(v for _, v in isi)
print(f"tiba={tiba:>3} et={et:>3} {user:<5} nilai={nilai:>7,} | wm={wm:>3}"
f" -> fitur {user}: count_60s={fitur_count}, sum_60s={fitur_sum:>9,}")
maks_et = et if maks_et is None else max(maks_et, et)Mari kita telaah baris-baris kode di atas:
- Baris 3-5 (
state,maks_et): state adalah kamus per pengguna;maks_etmelacak event time tertinggi yang terpantau dan menjadi dasar watermark. - Baris 8 (
wm = ...): watermark dihitung sebagaimaks_et - TOLERANSI— sistem bersedia menunggu event terlambat paling lama 5 detik. - Baris 9-14 (
if et < wm): event dengan event time di bawah watermark dianggap terlambat dan dibuang, karena window-nya sudah ditutup. - Baris 19-21 (
isi = ...): setiap event diterima, rolling window 60 detik pengguna tersebut dihitung ulang dari state — inilah nilai fitur yang disajikan ke model saat itu juga.
Output yang Dihasilkan:
tiba= 10 et= 10 Budi nilai=120,000 | wm= 0 -> fitur Budi: count_60s=1, sum_60s= 120,000
tiba= 18 et= 18 Siti nilai= 75,000 | wm= 5 -> fitur Siti: count_60s=1, sum_60s= 75,000
tiba= 25 et= 25 Budi nilai=210,000 | wm= 13 -> fitur Budi: count_60s=2, sum_60s= 330,000
tiba= 33 et= 33 Dewi nilai= 45,000 | wm= 20 -> fitur Dewi: count_60s=1, sum_60s= 45,000
tiba= 41 et= 40 Siti nilai=120,000 | wm= 28 -> fitur Siti: count_60s=2, sum_60s= 195,000
tiba= 48 et= 48 Budi nilai= 75,000 | wm= 35 -> fitur Budi: count_60s=3, sum_60s= 405,000
tiba= 56 et= 55 Dewi nilai=350,000 | wm= 43 -> fitur Dewi: count_60s=2, sum_60s= 395,000
tiba= 62 et= 60 Siti nilai= 45,000 | wm= 50 -> fitur Siti: count_60s=3, sum_60s= 240,000
tiba= 70 et= 70 Budi nilai=120,000 | wm= 55 -> fitur Budi: count_60s=3, sum_60s= 405,000
tiba= 80 et= 78 Dewi nilai=210,000 | wm= 65 -> fitur Dewi: count_60s=3, sum_60s= 605,000
tiba= 86 et= 52 Budi nilai=350,000 | wm= 73 -> DIBUANG (terlalu terlambat untuk window 60s)
tiba= 90 et= 90 Siti nilai=210,000 | wm= 73 -> fitur Siti: count_60s=3, sum_60s= 375,000
tiba= 96 et= 95 Dewi nilai= 75,000 | wm= 85 -> fitur Dewi: count_60s=3, sum_60s= 635,000
tiba= 99 et= 99 Budi nilai= 45,000 | wm= 90 -> fitur Budi: count_60s=3, sum_60s= 240,000Perhatikan dua event terlambat: event Dewi (et 78, tiba 80) masih lolos karena watermark baru 65, sedangkan event Budi (et 52, tiba 86) tiba saat watermark sudah 73 dan langsung dibuang.
Stream versus Batch: Harga Sebuah Dibuang #
Pertanyaan yang menentukan desain: seberapa jauh fitur stream menyimpang dari kebenaran? Skrip menutup dengan menghitung ulang rolling window yang sama dari seluruh event yang benar terjadi — sebagaimana batch feature computation di lake membacanya:
# --- Kebenaran batch: rolling window dihitung dari SEMUA event yang benar terjadi --
print()
print("== Fitur terakhir per pengguna: stream vs hitung ulang penuh (batch) ==")
for user in sorted(state):
et_terakhir = max(t for t, _ in state[user])
# batch membaca seluruh event di lake, termasuk yang dibuang stream
semua_user = [(et, v) for et, _, u, v in EVENTS if u == user]
batch_isi = [v for t, v in semua_user if et_terakhir - LEBAR < t <= et_terakhir]
stream_isi = [v for t, v in state[user] if et_terakhir - LEBAR < t <= et_terakhir]
cocok = sorted(batch_isi) == sorted(stream_isi)
print(f" {user:<5} stream: count={len(stream_isi):>2}, sum={sum(stream_isi):>9,}"
f" | batch: count={len(batch_isi):>2}, sum={sum(batch_isi):>9,}"
f" | cocok: {cocok}")Output yang Dihasilkan:
== Fitur terakhir per pengguna: stream vs hitung ulang penuh (batch) ==
Budi stream: count= 3, sum= 240,000 | batch: count= 4, sum= 590,000 | cocok: False
Dewi stream: count= 3, sum= 635,000 | batch: count= 3, sum= 635,000 | cocok: True
Siti stream: count= 3, sum= 375,000 | batch: count= 3, sum= 375,000 | cocok: True
diterima=13, dibuang=1 dari 14 eventRekap satu baris tersebut dicetak oleh penutup skrip yang menghitung nasib seluruh event:
print()
print(f"diterima={diterima}, dibuang={dibuang} dari {len(EVENTS)} event")| Aspek | Streaming | Batch |
|---|---|---|
| Latensi fitur | milidetik setelah event tiba | jam, sesuai jadwal |
| Kebenaran | bisa menyimpang (event dibuang) | kebenaran penuh dari seluruh data |
| Biaya | state di memori, selalu menyala | dihitung sekali lalu tidur |
| Peran | keputusan saat itu juga | pelatihan dan koreksi |
Satu event yang dibuang membuat fitur Budi melenceng besar: count_60s seharusnya 4 dengan total Rp590.000, stream hanya melihat 3 dan Rp240.000. Arsitektur praktis menyatukan keduanya — stream melayani keputusan real-time, batch mengoreksi dan melatih ulang model — pola yang di produksi berjalan di atas engine seperti Apache Kafka
dan Spark Structured Streaming dari chapter Stream Processing dengan Spark. Koreksi batch biasanya berjalan terjadwal setiap hari, sehingga selisih seperti milik Budi hanya hidup beberapa jam sebelum state stream disegarkan dari lake.
Fitur kini bisa dihitung di dua kecepatan. Satu persoalan terakhir menyatukan seluruh babak ini: bagaimana semua pipeline yang kita bangun terhubung ke siklus hidup model di produksi? Chapter Integrasi dengan MLOps menutupnya.