Windowing

Windowing adalah teknik membatasi agregasi stream pada rentang waktu tertentu, sehingga hasil berupa jumlah kejadian per periode atau per sesi dapat dinyatakan secara bermakna alih-alih satu akumulasi sepanjang log. Unit rentangnya disebut window, yaitu interval waktu berbatas yang menampung kejadian sesuai waktu kejadian masing-masing. Topik ini membahas empat tipe window pada Kafka Streams, yaitu tumbling, hopping, sliding, dan session, disertai tabel pemilihan tipe, grace period untuk record yang tiba terlambat, serta perilaku keluaran ketika window ditutup. Satu simulasi Python memutar agregasi berwindow pada kejadian dengan waktu kejadian tetap tanpa broker, sehingga isi setiap window dan efek data terlambat dapat diperiksa langsung.

Window tumbling dan hopping#

Tumbling window adalah window berukuran tetap yang tidak tumpang-tindih, dengan batas awal rapi pada kelipatan ukurannya sejak epoch. Window berukuran 10 detik menghasilkan interval [0;10), [10;20), dan [20;30), dan setiap kejadian masuk tepat satu window berdasarkan waktu kejadiannya. Bentuk ini tepat untuk ringkasan periodik, misalnya jumlah kejadian per 10 detik.

Hopping window berukuran tetap tetapi bergerak dengan langkah yang lebih kecil dari ukurannya, sehingga window berurutan saling tumpang-tindih. Window berukuran 10 detik dengan langkah 5 detik menghasilkan [0;10), [5;15), [10;20), dan seterusnya, dan satu kejadian dapat masuk ke dua window sekaligus. Agregasi setiap window dihitung terpisah, sehingga kejadian yang sama ikut terhitung pada beberapa window dalam keluaran keseluruhan. Bentuk ini digunakan ketika hasil agregasi perlu diperbarui lebih sering daripada sekali per ukuran window, misalnya ringkasan 10 detik yang bergerak tiap 5 detik.

Diagram berikut membandingkan interval kedua tipe tersebut pada sumbu waktu 0 sampai 30 detik.

Window sliding dan session#

Sliding window adalah window berukuran tetap yang batas awalnya mengikuti waktu kejadian, alih-alih kelipatan rapi sejak epoch. Setiap kejadian membentuk window yang menampung kejadian-kejadian di belakangnya dengan jarak tidak lebih dari ukuran window, sehingga hasil agregasi diperbarui pada setiap kejadian. Bentuk ini cocok untuk agregasi yang hasilnya harus tersedia segera setelah setiap kejadian tiba.

Session window mengelompokkan kejadian menurut aktivitas, dan ukurannya dinamis. Window dimulai pada kejadian pertama, diperpanjang oleh setiap kejadian berikutnya yang jaraknya kurang dari gap, yaitu jeda inaktivitas maksimum yang ditoleransi, dan berakhir ketika tidak ada kejadian baru selama gap itu. Dua sesi yang jaraknya kurang dari gap melebur menjadi satu sesi, sehingga batas sesi tidak pernah rapi pada kelipatan tertentu. Bentuk ini tepat untuk kejadian yang berkelompok secara alami, misalnya rangkaian kejadian satu pengguna yang berakhir setelah tidak ada kejadian baru selama gap.

Memilih tipe window#

Empat tipe window berbeda pada ukuran, tumpang-tindih, dan kebutuhan yang dilayani. Tabel berikut merangkum perbedaannya sebagai acuan pemilihan.

Ukuran window dan langkahnya menentukan dua biaya sekaligus, yaitu banyaknya window aktif pada state store dan banyaknya record pembaruan yang diteruskan ke tahap berikutnya. Window yang lebih kecil atau langkah yang lebih rapat menghasilkan resolusi yang lebih halus, tetapi memperbesar kedua biaya tersebut. Ukuran window lazim diturunkan dari jeda yang ingin dipantau, bukan dari keinginan memperoleh resolusi sehalus mungkin.

Grace period dan data terlambat#

Agregasi berwindow bekerja atas event time, yaitu waktu kejadian yang dibawa record, dan urutan kedatangan record pada task tidak selalu sama dengan urutan waktu kejadiannya. Window tetap menerima record yang waktu kejadiannya berada di dalam batasnya meskipun record itu tiba belakangan, selama kedatangannya masih terjadi sebelum window ditutup. Batas penutupan itu adalah batas akhir window ditambah grace period, yaitu masa tenggang yang mengakomodasi keterlambatan wajar.

Record yang tiba setelah window ditutup dibuang dan tidak masuk ke hasil mana pun. Masa tenggang yang lebih panjang menerima lebih banyak data terlambat, tetapi menunda hasil final dan memperpanjang pemeliharaan state window, sebab masa retention bawaan store berwindow mencakup masa tenggang. Besaran ini lazim diturunkan dari keterlambatan maksimum yang ditoleransi pipeline, bukan dari nilai bawaan semata.

Perilaku keluaran saat window ditutup#

Hasil agregasi berwindow berupa KTable yang entrinya ber-key gabungan antara key asli dan batas window, dan setiap kejadian baru meneruskan record pembaruan berisi nilai berjalan. Satu window yang menampung lima kejadian karenanya menghasilkan lima record pembaruan sebelum ditutup, dan nilai terakhir menjadi nilai akhir window tersebut. Tahap berikutnya yang mengharapkan satu record final per window perlu menyaring pembaruan itu.

Window ditutup setelah masa tenggang sejak batas akhir window berlalu, dan hasilnya baru final pada titik itu. DSL menyediakan operator Suppressed yang menahan record pembaruan dan meneruskan satu record final per window setelah penutupannya, sehingga tahap berikutnya tidak perlu menangani nilai berjalan sendiri. Tanpa penahanan itu, jumlah record keluaran lebih banyak daripada jumlah window, dan hal itu bukan error melainkan semantik bawaan agregasi berwindow.

Simulasi windowing pada kejadian tetap#

Simulasi berikut memutar agregasi berwindow pada delapan kejadian dengan waktu kejadian tetap, tanpa broker dan tanpa Kafka Streams sungguhan. Tiga bagian keluarannya meniru tiga tipe window, yaitu tumbling 10 detik, hopping ukuran 10 dengan langkah 5, dan session dengan gap 5 detik untuk satu key bernama dewi. Bagian session sengaja menyajikan kejadian pada t=8 paling akhir untuk meniru record yang tiba terlambat setelah kejadian pada t=21.

# simulasi agregasi berwindow pada kejadian ber-timestamp tetap, tanpa broker

# pasangan (detik, kata): timestamp kejadian dinyatakan dalam detik
kejadian = [
    (5, "log"), (7, "kejadian"), (9, "log"), (12, "dibaca"),
    (14, "log"), (18, "kejadian"), (22, "log"), (23, "kejadian"),
]

UKURAN = 10  # ukuran window dalam detik
LANGKAH = 5  # jarak antar window hopping dalam detik


def jumlahkan(isi):
    # hitung kejadian per kata di dalam satu window
    hasil = {}
    for _, kata in isi:
        hasil[kata] = hasil.get(kata, 0) + 1
    return hasil


def format(hasil):
    return ", ".join(f"{kata}={jumlah}" for kata, jumlah in hasil.items())


print("kejadian:", kejadian)

# tumbling: window [mulai, mulai + 10), batas awal kelipatan 10
print()
print("tumbling 10 detik:")
tumbling = {}
for detik, kata in kejadian:
    mulai = (detik // UKURAN) * UKURAN
    tumbling.setdefault(mulai, []).append((detik, kata))
for mulai in sorted(tumbling):
    print(f"  window [{mulai};{mulai + UKURAN}): {format(jumlahkan(tumbling[mulai]))}")

# hopping: ukuran 10, langkah 5, batas awal kelipatan 5
print()
print("hopping ukuran 10, langkah 5:")
hopping = {}
for detik, kata in kejadian:
    # kejadian masuk ke setiap window [s; s + 10) dengan s kelipatan 5
    # yang rentangnya menampungnya, yaitu s lebih besar dari detik - 10
    terkecil = ((detik - UKURAN) // LANGKAH + 1) * LANGKAH
    for mulai in range(terkecil, detik + 1, LANGKAH):
        hopping.setdefault(mulai, []).append((detik, kata))
for mulai in sorted(hopping):
    print(f"  window [{mulai};{mulai + UKURAN}): {format(jumlahkan(hopping[mulai]))}")

# session: satu key saja, gap 5 detik, kejadian terakhir tiba terlambat
print()
print("session gap 5 detik untuk key dewi:")
GAP = 5
sesi = []  # setiap sesi: [mulai, akhir, jumlah_kejadian]
for detik in [1, 4, 12, 15, 21, 8]:
    gabung = [s for s in sesi if s[0] - GAP <= detik <= s[1] + GAP]
    if gabung:
        mulai = min([s[0] for s in gabung] + [detik])
        akhir = max([s[1] for s in gabung] + [detik])
        jumlah = sum(s[2] for s in gabung) + 1
        sesi = [s for s in sesi if s not in gabung]
        sesi.append([mulai, akhir, jumlah])
    else:
        sesi.append([detik, detik, 1])
    sesi.sort()
    teks = ", ".join(
        f"sesi [{s[0]};{s[1]}] ({s[2]} kejadian)" for s in sesi)
    label = "kejadian telat t" if detik == 8 else "kejadian t"
    print(f"  {label}={detik}: {teks}")

kejadian: [(5, 'log'), (7, 'kejadian'), (9, 'log'), (12, 'dibaca'), (14, 'log'), (18, 'kejadian'), (22, 'log'), (23, 'kejadian')]

tumbling 10 detik:
  window [0;10): log=2, kejadian=1
  window [10;20): dibaca=1, log=1, kejadian=1
  window [20;30): log=1, kejadian=1

hopping ukuran 10, langkah 5:
  window [0;10): log=2, kejadian=1
  window [5;15): log=3, kejadian=1, dibaca=1
  window [10;20): dibaca=1, log=1, kejadian=1
  window [15;25): kejadian=2, log=1
  window [20;30): log=1, kejadian=1

session gap 5 detik untuk key dewi:
  kejadian t=1: sesi [1;1] (1 kejadian)
  kejadian t=4: sesi [1;4] (2 kejadian)
  kejadian t=12: sesi [1;4] (2 kejadian), sesi [12;12] (1 kejadian)
  kejadian t=15: sesi [1;4] (2 kejadian), sesi [12;15] (2 kejadian)
  kejadian t=21: sesi [1;4] (2 kejadian), sesi [12;15] (2 kejadian), sesi [21;21] (1 kejadian)
  kejadian telat t=8: sesi [1;15] (5 kejadian), sesi [21;21] (1 kejadian)

Tiga window tumbling menampung seluruh delapan kejadian tanpa pengulangan, sedangkan lima window hopping menampung enam belas kejadian karena setiap kejadian masuk ke dua window sekaligus. Pada bagian session, kejadian t=1 dan t=4 membentuk sesi pertama, kejadian t=12 dan t=15 sesi kedua, dan kejadian t=21 berdiri sendiri karena jaraknya melebihi gap. Kejadian terlambat t=8 berjarak kurang dari gap terhadap kedua sesi yang terpisah itu, sehingga keduanya melebur menjadi sesi [1;15] berisi lima kejadian. Perilaku tersebut sesuai session window pada Kafka Streams, yaitu sesi lama dapat melebur kembali oleh record terlambat selama masih tersimpan pada store sesinya.

Agregasi berwindow pada DSL#

Pada DSL Kafka Streams, agregasi berwindow disusun dengan menyisipkan operator windowedBy di antara pengelompokan dan agregasinya. Potongan berikut menghitung kejadian per key pada window tumbling 10 detik dengan masa tenggang 5 detik. Potongan ini merupakan bagian proyek Maven atau Gradle dengan dependensi kafka-streams, sehingga tidak dieksekusi di lingkungan penulisan dan berfungsi sebagai bahan bacaan.

// tidak-dijalankan
KTable<Windowed<String>, Long> jumlahPerWindow = kata.groupBy(
        (kunci, teks) -> teks)
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofSeconds(10), Duration.ofSeconds(5)))
    .count(Materialized.as("jumlah-kata-per-window"));

Tipe key hasilnya adalah Windowed<String> yang membawa key asli beserta batas window-nya, sehingga hasil setiap window tidak saling menimpa. Window hopping dinyatakan pada TimeWindows yang sama dengan tambahan advanceBy(Duration.ofSeconds(5)), sedangkan window session dinyatakan lewat SessionWindows.with(Duration.ofSeconds(5)). Metode agregasinya tetap count, reduce, atau aggregate, karena window hanya mengubah lingkup waktu agregasi dan bukan cara penghitungannya.

Kesalahan umum#

1. Menyetel masa tenggang terlalu pendek atau membiarkannya nol. Record terlambat yang tiba setelah window ditutup dibuang tanpa pesan, sehingga hasil window kehilangan kejadian yang sudah tertulis pada topic input. Turunkan besaran masa tenggang dari keterlambatan maksimum yang ditoleransi, dan cocokkan jumlah kejadian input dengan hasil agregasi ketika pipeline diuji.
2. Menganggap setiap window menghasilkan tepat satu record keluaran. Agregasi berwindow meneruskan nilai berjalan pada setiap kejadian, sehingga tahap berikutnya menerima banyak pembaruan per window. Gunakan operator Suppressed apabila yang dibutuhkan adalah satu record final per window.
3. Memilih ukuran window tanpa memperhitungkan state. Ukuran window yang besar dan langkah hopping yang rapat memperbanyak window aktif, dan state store berwindow tumbuh mengikuti jumlah key dikalikan jumlah window yang dipelihara. Ukur kebutuhan state-nya sebelum menaikkan resolusi agregasi.
4. Menetapkan gap session yang tidak sesuai ritme kejadian. Gap yang terlalu kecil memecah satu aktivitas menjadi banyak sesi pendek, sedangkan gap yang terlalu besar melebur kejadian yang seharusnya terpisah. Amati jeda antar kejadian pada data nyata sebelum gap ditetapkan.

Ringkasan#

• Tumbling window berukuran tetap tanpa tumpang-tindih, hopping window tumpang-tindih dengan langkah lebih kecil dari ukurannya, sliding window bergerak mengikuti kejadian, dan session window dinamis menurut gap.
• Grace period menunda penutupan window untuk menerima data terlambat, dan record yang tiba setelah window ditutup dibuang dari hasil.
• Agregasi berwindow meneruskan nilai berjalan pada setiap kejadian, sedangkan operator Suppressed meneruskan satu record final per window setelah window ditutup.
• Ukuran window, langkah, dan masa tenggang menentukan jumlah state, banyaknya record pembaruan, serta ketahanan hasil terhadap keterlambatan.

Referensi#

• Apache Kafka Streams: core concepts — Apache Software Foundation — ikhtisar tipe window pada Kafka Streams beserta semantik waktunya.
• Apache Kafka Streams DSL: stateful transformations — Apache Software Foundation — agregasi windowedBy dengan TimeWindows dan SessionWindows, grace period, serta operator Suppressed.
• Apache Kafka Streams: overview — Apache Software Foundation — posisi windowing sebagai bagian kemampuan pustaka Kafka Streams.