Menangani Event yang Terlambat dan Tidak Berurutan dalam Stream Processing

Masalah dan Cakupan

Klien seluler melaporkan event pembelian dengan setidaknya event_id, event_time, user_id, dan amount yang sudah dikonversi ke satu mata uang. Percobaan ulang jaringan dapat membuat duplikat, perangkat offline dapat mengunggah beberapa jam kemudian, dan partisi yang berbeda tidak mempertahankan urutan global. Hitung pendapatan untuk setiap jam kalender UTC berdasarkan event_time, dengan input puncak 20.000 event per detik.

Bisnis menginginkan perkiraan untuk jam saat ini dalam waktu satu menit, hasil tepat waktu (on-time) setelah watermark melewati akhir window, dan koreksi otomatis selama 24 jam setelah window tersebut berakhir. Data yang tiba melebihi 24 jam tidak boleh hilang secara diam-diam; data tersebut masuk ke audit dan rekonsiliasi offline. Throughput, ketepatan waktu, dan periode koreksi adalah input masalah, bukan klaim performa tentang engine stream-processing.

Asumsikan produsen menetapkan event_id yang stabil, dan percobaan ulang yang valid dengan ID tersebut membawa konten bisnis yang sama. Event mentah tetap berada di penyimpanan yang dapat diputar ulang (replayable). Cakupan ini mencakup semantik waktu, watermark, pemicu window, deduplikasi, pembaruan hasil, kapasitas state, pemulihan, dan rekonsiliasi. Pemilihan broker dan konversi mata uang berada di luar cakupan. Ini adalah pertanyaan rekayasa data: tujuannya adalah kontrak data yang dapat diverifikasi yang membuat kompromi antara kelengkapan, latensi, dan biaya state menjadi eksplisit.

Hal yang Dievaluasi oleh Pewawancara

Sinyal pertama adalah apakah kandidat memisahkan tiga jenis jam (clock). event_time menentukan jam bisnis yang memuat suatu event. processing_time menyatakan kapan engine melihatnya dan dapat mendorong penyegaran awal satu menit. Watermark adalah perkiraan engine terhadap progres event-time. Jika processing time menentukan window, memutar ulang riwayat yang sama pada waktu yang berbeda dapat menghasilkan hasil per jam yang berbeda.

Sinyal kedua adalah memperlakukan watermark sebagai estimasi progres alih-alih janji mutlak. Memajukannya dengan cepat memberikan output yang tepat waktu tetapi mengklasifikasikan lebih banyak rekaman sebagai terlambat. Menahannya akan meningkatkan kelengkapan tetapi menunda penutupan window dan menahan lebih banyak state. Jawaban yang kuat menurunkan kebijakan dari data penundaan kedatangan yang terukur dan SLO koreksi, alih-alih menghafal penundaan tetap lima menit atau satu jam.

Sinyal ketiga adalah memisahkan watermark, allowed lateness (keterlambatan yang diizinkan), dan retensi deduplikasi. Watermark mengontrol pemicuan on-time. Allowed lateness mengontrol berapa lama state window tetap dapat dikoreksi. State deduplikasi harus mencakup periode di mana salinan lain dari suatu event dapat tiba. Nilai-nilai tersebut mungkin saling terkait, tetapi bukan merupakan satu pengaturan yang sama.

Terakhir, output harus konvergen. Event yang terlambat menyebabkan beberapa hasil untuk satu window. Menambahkan (append) setiap snapshot membuat penjumlahan downstream menghitung window tersebut berulang kali. Hasil memerlukan upsert idempoten yang diberi kunci berdasarkan window, dengan penanda revisi atau finalitas. Checkpoint melindungi state operator; jika sink tidak dapat melakukan commit bersamaan dengan checkpoint, desain masih memerlukan penulisan transaksional atau penggantian versi yang idempoten.

Pertanyaan untuk Diklarifikasi Sebelum Menjawab

• Apakah metrik bisnis didasarkan pada waktu kejadian atau waktu kedatangan? Masalah ini menggunakan event_time pembelian. Processing time hanya sesuai jika metriknya secara spesifik adalah “permintaan yang diproses oleh sistem sekarang.”
• Apakah hasil satu menit berupa perkiraan atau angka final? Di sini berupa perkiraan, sehingga pemicu awal berbasis processing-time valid. Jika setiap hasil yang ditampilkan harus lengkap, sistem harus menunggu lebih lama atau menggunakan komputasi batch.
• Bisakah laporan berubah setelah 24 jam? 24 jam tersebut adalah periode koreksi otomatis dari streaming job. Data yang lebih lambat masuk ke rekonsiliasi. Persyaratan regulasi atau penyelesaian transaksi (settlement) untuk mengoreksi setiap rekaman yang terlambat berarti pembersihan state tidak dapat menentukan kebenaran akhir dari data tersebut.
• Bisakah duplikat diidentifikasi berdasarkan ID? event_id yang stabil tersedia di sini. Menebak dari pengguna, jumlah, dan waktu akan menghapus pembelian yang sah sekaligus melewatkan duplikat; perbaiki kontrak produsen terlebih dahulu.
• Bagaimana jika satu ID membawa konten yang berbeda? Jangan memilih first-write-wins atau last-write-wins. Catat konflik payload-fingerprint dan karantina karena produsen melanggar kontrak idempotensi.
• Apakah sink menerima snapshot, delta, atau retraksi? Desain ini menghasilkan snapshot window lengkap dan melakukan upsert berdasarkan (window_start, dimensions). Sink append-only memerlukan changelog berversi yang lapisan bacanya memilih revisi terbaru.
• Bisakah timestamp event dipercaya? Karantina timestamp yang jauh di masa depan, lebih lama dari retensi data mentah, atau tidak valid untuk zona waktu yang disepakati. Timestamp masa depan yang buruk yang ikut serta dalam event time maksimum dapat memajukan watermark secara terlalu agresif.

Jawaban 30 Detik

“Saya akan menetapkan window per jam UTC berdasarkan event_time, menghasilkan watermark per partisi sumber aktif, dan memajukannya pada progres aman minimum. Pemicu processing-time memancarkan perkiraan setiap menit; watermark memancarkan hasil tepat waktu (on-time), dengan state dipertahankan selama 24 jam untuk koreksi. event_id dan fingerprint payload mendeduplikasi event, sementara sink melakukan upsert berdasarkan window dan revisi. Data yang terlalu terlambat dikirim ke side output dan rekonsiliasi harian. Pemulihan menggabungkan sumber yang dapat diputar ulang, checkpoint, dan sink transaksional atau idempoten.”

Pembahasan Mendalam Langkah demi Langkah

Mulailah dengan kontrak hasil. Gunakan window setengah terbuka seperti [10:00, 11:00), dan sertakan setidaknya window_start serta dimensi pelaporan dalam kuncinya. Setiap output adalah snapshot pendapatan lengkap untuk window tersebut pada versinya saat ini:

HourlyRevenue {
  window_start
  window_end
  revenue
  revision
  result_state  // EARLY | ON_TIME | FINAL
}

revision meningkat secara monoton untuk suatu window, dan sink hanya menerima revisi yang lebih besar. EARLY adalah penyegaran satu menit dan tidak mengklaim kelengkapan. ON_TIME berarti watermark telah melewati akhir window. FINAL berarti periode koreksi otomatis 24 jam telah berakhir. FINAL adalah kontrak operasional, bukan klaim bahwa tidak ada lagi data yang ada; rekonsiliasi tetap menangani rekaman di luar batas waktu tersebut.

Turunkan kebijakan waktu dari batasan yang ada. Setiap partisi sumber yang aktif mengekstrak event_time yang telah divalidasi. Strategi bounded-out-of-orderness yang umum adalah:

partition_watermark = max_valid_event_time_seen - out_of_order_bound
operator_watermark = min(active_partition_watermarks)

Pilih out_of_order_bound dari distribusi penundaan kedatangan yang diamati, tingkat keterlambatan data yang dapat diterima, dan target latensi ON_TIME. Watermark per partisi mencegah partisi yang cepat menyatakan partisi yang lambat telah selesai. Operator multi-input mengambil progres minimum sehingga tidak mendahului input yang masih dapat menghasilkan rekaman yang lebih lama. Tandai partisi sebagai idle hanya setelah partisi tersebut tidak menghasilkan data selama interval yang disepakati; jika tidak, partisi tersebut dapat menghentikan watermark global selamanya. Ambang batas idleness yang terlalu pendek membuat rekaman lama dari partisi yang aktif kembali menjadi terlambat, sehingga side output tetap diperlukan.

Window memiliki dua kelompok pemicu (trigger). Pemicu awal processing-time memancarkan snapshot lengkap saat ini sekali semenit. Ketika watermark melewati window_end, pancarkan revisi ON_TIME. Pertahankan state window sampai watermark melewati window_end + 24h; setiap event terlambat yang valid akan memperbarui agregat dan menghasilkan revisi yang lebih tinggi. Saat pembersihan (cleanup), pancarkan FINAL dan lepaskan state. Model trigger dan akumulasi Beam mengilustrasikan mengapa “kapan harus memancarkan” dan “apakah sebuah pane merupakan delta atau snapshot terakumulasi” adalah pilihan yang terpisah. Desain ini menggunakan snapshot terakumulasi dan upsert, sehingga sistem downstream tidak perlu menyusun kembali pane-pane tersebut.

Lakukan deduplikasi berdasarkan event_id sebelum agregasi. Simpan event_id → (event_time, payload_fingerprint). Izinkan event pertama lewat, buang salinan lain dengan fingerprint yang sama, dan kirim ID yang sama dengan fingerprint berbeda ke stream konflik. Retensi bergantung pada interval terpanjang di mana upstream dapat memutar ulang salinan lain, bukan hanya pada ukuran window. Semantik deduplikasi berbasis watermark milik Spark juga memerlukan ambang penundaan yang lebih panjang daripada rentang timestamp antara duplikat paling awal dan paling akhir. Pembersihan state yang terjadi terlalu cepat memungkinkan duplikat yang terlambat dihitung kembali.

Perkirakan state secara eksplisit. Jika puncak 20.000 event per detik berlangsung selama 24 jam dan setiap event unik, job akan mengingat 20,000 × 86,400 = 1,728,000,000 ID. Bahkan payload logis ilustratif yang hanya sebesar 40 byte per entri membuat batas atas ini menjadi sekitar 69.12 GB. Objek engine, indeks, backend state, checkpoint, dan replika meningkatkan ukuran fisik. Oleh karena itu, state memerlukan partisi berbasis kunci dan checkpoint inkremental, sementara retensi mengikuti kontrak replay. Jika 99% duplikat tiba dalam periode yang lebih singkat, horison streaming yang lebih pendek dapat dikombinasikan dengan business key pada sink dan rekonsiliasi offline, tetapi risiko residu duplikat harus dikuantifikasi alih-alih disembunyikan di balik TTL.

Jalur output harus menoleransi replay. Idealnya, sink berpartisipasi dalam transaksi checkpoint sehingga posisi sumber, state operator, dan output window di-commit secara bersamaan. Jika tidak, jadikan (window_key, revision) sebagai upsert idempoten: memutar ulang revisi yang sama atau lebih lama setelah crash tidak boleh menimpa nilai baru. Label exactly-once dari sebuah broker tidak mencukupi jika penulisan database eksternal berada di luar konfirmasi checkpoint. Semantik end-to-end bergantung pada batas commit bersama antara sumber, state, dan sink.

Event di luar 24 jam dikirim ke side output too-late dan tetap berada di log mentah. Batch job harian menghitung ulang window yang terpengaruh berdasarkan event_time dengan aturan ID dan jumlah yang sama, lalu membandingkannya dengan snapshot streaming FINAL. Perbedaan material akan membuat revisi yang lebih tinggi atau masuk ke persetujuan keuangan. Batch job secara atomik menggantikan window atau melakukan upsert berversi. Menambahkan hasil batch ke total yang sudah ada akan menghitung data yang sama dua kali ketika batch diputar ulang.

Gunakan urutan minimal untuk memverifikasi semantik. Window [10:00, 11:00) menerima A=100, B=50, lalu salinan A yang identik. Hasil ON_TIME yang telah dideduplikasi adalah 150. Setelah watermark melewati 11:00, C=20 tiba selama allowed lateness, dan revisi yang lebih tinggi mengubah window menjadi 170. D tiba setelah watermark melewati titik pembersihan dan dialihkan ke rekonsiliasi alih-alih langsung menyentuh state yang telah dibersihkan. Jika payload kedua dari A memiliki jumlah 120, data tersebut masuk ke stream konflik; hasilnya tidak boleh menjadi 170 atau 190.

Matriks pengujian juga mencakup ketidakteraturan urutan di seluruh partisi, partisi idle yang menghentikan watermark, karantina timestamp masa depan, batas tepat sebelum dan sesudah watermark, crash sebelum dan sesudah penulisan sink, pemulihan checkpoint dan replay, revisi tidak berurutan yang mencapai sink, serta eksekusi rekonsiliasi berulang. Metrik produksi meliputi p50/p95/p99 dan penundaan kedatangan ekor (tail), watermark saat ini dan lag per operator, jumlah dan nilai early/on-time/late/too-late, hit deduplikasi dan konflik fingerprint, byte state, durasi checkpoint, revisi sink yang ditolak, serta delta batch-versus-stream.

Contoh Jawaban yang Kuat

“Saya akan mendefinisikan angka satu menit sebagai perkiraan dan 24 jam sebagai periode koreksi otomatis. event_time menetapkan pembelian ke jam UTC; processing time hanya mendorong pemicu awal satu menit. Setiap partisi input aktif menghasilkan watermark dari event time tervalidasi terbesarnya dikurangi toleransi ketidakteraturan (disorder allowance), dan downstream menggunakan input aktif yang paling lambat. Partisi idle ditandai secara eksplisit, dan timestamp masa depan yang tidak valid dikarantina.

Sebelum agregasi, saya mendeduplikasi berdasarkan event_id, menyimpan event time dan fingerprint payload. Percobaan ulang yang identik dihitung sekali; ID yang sama dengan konten berbeda masuk ke stream konflik. Window jam alami memancarkan snapshot EARLY lengkap setiap menit, snapshot ON_TIME setelah akhir window berada di belakang watermark, dan mempertahankan state selama 24 jam. Event terlambat dalam periode tersebut memperbarui total dan meningkatkan revisi; pembersihan memancarkan FINAL.

Sink melakukan upsert berdasarkan window key dan revisi alih-alih menambahkan setiap snapshot sebagai delta. Sumber data dapat diputar ulang dan state operator dipulihkan dari checkpoint. Jika sink tidak dapat bergabung dalam transaksi checkpoint, penulisan berversi mencegah replay pemulihan menggantikan hasil yang lebih baru. Event yang lebih lambat dari 24 jam dikirim ke side output, dan batch job harian menghitung ulang log mentah serta membandingkannya dengan FINAL.

Jika puncak 20.000 event per detik berlangsung selama 24 jam dan semua ID unik, batas atasnya adalah 1,728 miliar entri deduplikasi. Bahkan 40 byte logis per entri menghasilkan 69,12 GB, sehingga distribusi keterlambatan yang terukur harus menjustifikasi retensi, serta biaya state dan checkpoint perlu dipantau. Saya akan menyelesaikannya dengan injeksi kegagalan untuk duplikat, ketidakteraturan urutan, keterlambatan, timestamp masa depan, partisi idle, dan pemulihan, sambil memastikan bahwa window final sama dengan hasil komputasi ulang offline.”

Kesalahan Umum

• Menetapkan window bisnis berdasarkan processing time → event yang sama dapat mendarat di jam yang berbeda selama pemutaran ulang riwayat → gunakan event time untuk keanggotaan window dan processing time hanya untuk output awal.
• Menggambarkan watermark sebagai “tidak ada data lebih lama yang dapat tiba” → ini biasanya merupakan perkiraan heuristik progres, sehingga event yang lebih lama masih dapat muncul → definisikan allowed lateness, side output too-late, dan rekonsiliasi.
• Menggunakan satu nilai tanpa penjelasan untuk penundaan watermark, allowed lateness, dan TTL deduplikasi → masing-masing mengatur pemicuan, state window, dan pengenalan duplikat → turunkan nilainya dari SLO latensi, periode koreksi, dan kontrak replay upstream.
• Menambahkan (append) total pada setiap pemicuan → pemicuan EARLY, ON_TIME, dan late dijumlahkan berulang kali → pancarkan snapshot lengkap dan upsert berdasarkan window dan revisi, atau definisikan protokol retractable-delta.
• Memajukan watermark dari event time maksimum global → partisi yang cepat atau timestamp masa depan yang buruk membuat data partisi lambat menjadi terlambat terlalu dini → buat progres per partisi, ambil nilai minimum dari input aktif, dan validasi timestamp.
• Menjaga partisi yang senyap secara permanen di dalam perhitungan minimum → watermark berhenti, mencegah window dan state deduplikasi dibersihkan → gunakan penanganan idleness yang dapat diobservasi dan rute data terlambat yang kembali aktif dengan benar.
• Hanya menyimpan ID event → penggunaan ulang ID oleh produsen tertelan secara diam-diam → simpan juga fingerprint payload dan karantina konflik.
• Mengatakan “exactly once diaktifkan” → sink eksternal mungkin tidak berbagi batas checkpoint → telusuri jalur commit end-to-end dan gunakan sink transaksional atau penulisan idempoten berversi.
• Membuang data setelah pembersihan state → stream terlihat stabil sementara hasil keuangan menjadi tidak dapat dijelaskan → pertahankan event mentah dan terapkan side output, komputasi ulang batch, serta peninjauan diskrepansi.

Pertanyaan Lanjutan dan Jawabannya

Pertanyaan Lanjutan 1: Seberapa besar toleransi ketidakteraturan (disorder allowance) watermark yang harus ditetapkan?

Ukur processing_time - event_time di produksi dan lakukan segmentasi berdasarkan sumber, versi klien, dan wilayah. Pertama, pilih hasil ON_TIME terbaru yang dapat diterima dan porsi event yang diizinkan masuk ke jalur koreksi; lalu pilih persentil yang memenuhi keduanya. Lanjutkan pemantauan p50, p95, p99, dan ekor distribusinya. Perubahan distribusi harus melalui rilis konfigurasi daripada membiarkan satu pencilan (outlier) mengubah kebijakan secara otomatis. Watermark mengatur pemicuan tepat waktu, sedangkan periode koreksi 24 jam tetap mencakup ekor distribusi yang lebih panjang.

Pertanyaan Lanjutan 2: Mengapa satu partisi Kafka yang hening dapat menghentikan window?

Progres yang aman dari operator multi-input adalah watermark input minimum. Partisi yang tidak maju membuat nilai minimum tersebut tidak berubah. Setelah mengonfirmasi bahwa partisi tersebut tidak menghasilkan event selama ambang batas idleness, tandai partisi tersebut sebagai idle sehingga untuk sementara keluar dari perhitungan minimum. Jangan menyetel ambang batas terlalu pendek: rekaman dari partisi yang aktif kembali bisa lebih lama dari watermark baru. Rekaman tersebut harus masuk ke pemrosesan allowed-late atau side output, dan transisi status idle memerlukan metriknya sendiri.

Pertanyaan Lanjutan 3: Mengapa state deduplikasi 24 jam tidak menjamin bahwa duplikasi tidak mungkin terjadi selamanya?

Dua puluh empat jam hanya mencakup periode koreksi otomatis untuk masalah ini. Jika upstream memutar ulang sebuah ID pada jam ke-25, state streaming sudah hilang dan event tersebut dapat masuk kembali ke agregat. Deduplikasi permanen memerlukan indeks unik bisnis yang berumur lebih panjang, registri event yang dapat di-query, atau deduplikasi offline penuh; masing-masing menambah biaya penyimpanan dan penulisan. Nyatakan jaminan sebagai “diduduplikasi dalam horison replay yang dideklarasikan,” lalu rekonsiliasi rekaman yang lebih lambat.

Pertanyaan Lanjutan 4: Bagaimana jika database downstream mendukung INSERT tetapi tidak mendukung upsert?

Tulis setiap hasil ke changelog yang immutable yang diberi kunci berdasarkan window dan revisi. Lapisan baca membangun tampilan saat ini dari revisi maksimum untuk setiap window. Konsumen harus memahami bahwa setiap rekaman adalah snapshot lengkap, bukan delta, dan tidak boleh menjumlahkan semua revisi. Jika biaya query terlalu tinggi, lakukan pemadatan (compact) secara asinkron ke dalam tabel penyaji yang mendukung penggantian atomik sambil tetap mempertahankan log versi untuk audit dan pemulihan.

Pertanyaan Lanjutan 5: Bagaimana Anda membuktikan bahwa pemulihan tidak menghitung lebih (overcount) maupun menghitung kurang (undercount)?

Untuk input yang tetap, catat set deduplikasi yang diharapkan dan revisi window. Hentikan job pada tiga batasan: setelah pembacaan sumber tetapi sebelum checkpoint state, setelah checkpointing state tetapi sebelum konfirmasi sink, dan setelah penulisan sink tetapi sebelum konfirmasi checkpoint. Putar ulang setelah pemulihan dan pastikan bahwa setiap ID event berkontribusi sekali, sink hanya mempertahankan revisi terbesar, dan window final sama dengan komputasi ulang offline. Status RUNNING tanpa injeksi kegagalan pada batas-batas commit ini bukanlah bukti kebenaran end-to-end.