Kendalikan Peristiwa Lewat dan Tidak Mengikut Urutan dalam Pemprosesan Strim

Masalah dan Skop

Klien mudah alih melaporkan peristiwa pembelian dengan sekurang-kurangnya event_id, event_time, user_id, dan amount yang telah ditukar kepada satu mata wang. Percubaan semula rangkaian boleh mencipta duplikasi, peranti luar talian boleh memuat naik beberapa jam kemudian, dan sekatan (partitions) yang berbeza tidak mengekalkan urutan global. Kira hasil pendapatan untuk setiap jam jam UTC mengikut event_time, dengan input kemuncak sebanyak 20,000 peristiwa sesaat.

Pihak perniagaan mahukan anggaran bagi jam semasa dalam tempoh satu minit, hasil tepat pada masanya (on-time) selepas penanda aras air melepasi penghujung tetingkap, dan pembetulan automatik selama 24 jam selepas tetingkap tersebut berakhir. Data yang tiba melebihi 24 jam tidak boleh hilang secara senyap; ia dihantar ke audit dan penyelarasan luar talian. Daya pemprosesan (throughput), ketepatan masa, dan tempoh pembetulan ialah input masalah, bukan tuntutan prestasi mengenai enjin pemprosesan strim.

Andaikan pengeluar menetapkan event_id yang stabil, dan percubaan semula yang sah dengan ID tersebut membawa kandungan perniagaan yang sama. Peristiwa mentah kekal dalam storan yang boleh dimainkan semula (replayable). Skop ini merangkumi semantik masa, penanda aras air, pencetus tetingkap, penyahduplikasian, kemas kini hasil, kapasiti keadaan (state), pemulihan, dan penyelarasan. Pemilihan broker dan penukaran mata wang adalah di luar skop. Ini ialah soalan kejuruteraan data: matlamatnya adalah untuk membina kontrak data yang boleh disahkan yang menyatakan secara jelas pertukaran antara kelengkapan, kependaman (latency), dan kos keadaan.

Perkara yang Dinilai oleh Penemu Duga

Isyarat pertama ialah sama ada calon memisahkan tiga jam (clocks). event_time menentukan jam perniagaan yang mengandungi sesuatu peristiwa. processing_time menyatakan masa enjin melihatnya dan boleh memacu penyegaran awal satu minit. Penanda aras air ialah anggaran enjin tentang kemajuan masa peristiwa. Jika masa pemprosesan menentukan tetingkap, memainkan semula sejarah yang sama pada masa yang berbeza boleh menghasilkan keputusan setiap jam yang berbeza.

Isyarat kedua ialah menganggap penanda aras air sebagai anggaran kemajuan dan bukannya janji mutlak. Memajukannya dengan cepat memberikan output yang tepat pada masanya tetapi mengklasifikasikan lebih banyak rekod sebagai lewat. Menahannya meningkatkan kelengkapan tetapi menangguhkan penutupan tetingkap dan mengekalkan lebih banyak keadaan. Jawapan yang kukuh menerbitkan dasar daripada data kelewatan ketibaan yang diukur dan SLO pembetulan, berbanding menghafal kelewatan tetap lima minit atau satu jam.

Isyarat ketiga ialah memisahkan penanda aras air, kelewatan yang dibenarkan (allowed lateness), dan pengekalan penyahduplikasian. Penanda aras air mengawal pencetusan tepat pada masanya. Kelewatan yang dibenarkan mengawal tempoh keadaan tetingkap kekal boleh dibetulkan. Keadaan penyahduplikasian mesti merangkumi tempoh di mana salinan lain bagi sesuatu peristiwa boleh tiba. Nilai-nilai ini mungkin berkaitan, tetapi ia bukan satu tetapan yang sama.

Akhir sekali, output mesti menumpu (converge). Peristiwa lewat menyebabkan beberapa hasil untuk satu tetingkap. Menambah (append) setiap snapshot menyebabkan jumlah hiliran mengira tetingkap secara berulang. Hasil memerlukan upsert idempoten yang berkuncikan tetingkap, dengan penanda semakan atau kemuktamadan. Titik semak (checkpoint) melindungi keadaan pengendali; jika sink tidak dapat melakukan komit bersama-sama titik semak, reka bentuk masih memerlukan penulisan transaksi atau penggantian versi idempoten.

Soalan untuk Dijelaskan Sebelum Menjawab

• Adakah metrik perniagaan berdasarkan kejadian atau ketibaan? Masalah ini menggunakan event_time pembelian. Masa pemprosesan hanya sesuai apabila metrik tersebut secara khusus merujuk kepada “permintaan yang diproses oleh sistem sekarang.”
• Adakah hasil satu minit merupakan anggaran atau angka muktamad? Ia adalah anggaran di sini, jadi pencetus awal masa pemprosesan adalah sah. Jika setiap hasil yang dipaparkan mesti lengkap, sistem perlu menunggu lebih lama atau menggunakan pengiraan kelompok (batch).
• Bolehkah laporan berubah selepas 24 jam? 24 jam tersebut ialah tempoh pembetulan automatik kerja penstriman. Data yang tiba kemudian dihantar ke penyelarasan. Keperluan kawal selia atau penyelesaian untuk membetulkan setiap rekod lewat bermakna pembersihan keadaan tidak boleh mentakrifkan kebenaran mutlak data.
• Bolehkah duplikasi dikenal pasti melalui ID? event_id yang stabil wujud di sini. Meneka daripada pengguna, jumlah, dan masa akan memadamkan pembelian yang sah serta terlepas pandang duplikasi; betulkan kontrak pengeluar terlebih dahulu.
• Bagaimana jika satu ID membawa kandungan yang berbeza? Jangan pilih first-write-wins atau last-write-wins. Rekod konflik cap jari muatan (payload fingerprint) dan kuarantinkannya kerana pengeluar melanggar kontrak keidempotenan.
• Adakah sink menerima snapshot, delta, atau penarikan balik (retractions)? Reka bentuk ini mengeluarkan snapshot tetingkap yang lengkap dan melakukan upsert mengikut (window_start, dimensions). Sink jenis append-only memerlukan log perubahan berversi yang lapisan bacaannya memilih semakan terkini.
• Bolehkah cap masa peristiwa dipercayai? Kuarantinkan cap masa yang jauh pada masa hadapan, lebih lama daripada tempoh pengekalan data mentah, atau tidak sah untuk zon masa yang dipersetujui. Cap masa masa hadapan yang tidak sah yang mengambil bahagian dalam masa peristiwa maksimum boleh memajukan penanda aras air secara terlalu agresif.

Jawapan 30 Saat

“Saya akan menetapkan tetingkap setiap jam UTC mengikut event_time, menjana penanda aras air bagi setiap sekatan sumber yang aktif, dan memajukannya pada kemajuan selamat minimum. Pencetus masa pemprosesan mengeluarkan anggaran setiap minit; penanda aras air mengeluarkan hasil tepat pada masanya, dengan keadaan dikekalkan selama 24 jam untuk pembetulan. event_id dan cap jari muatan menyahduplikasi peristiwa, manakala sink melakukan upsert mengikut tetingkap dan semakan. Data yang terlalu lewat dihantar ke output sisi (side output) dan penyelarasan harian. Pemulihan menggabungkan sumber yang boleh dimainkan semula, titik semak, dan sink bertransaksi atau idempoten.”

Analisis Mendalam Langkah demi Langkah

Mulakan dengan kontrak hasil. Gunakan tetingkap separuh terbuka seperti [10:00, 11:00), dan sertakan sekurang-kurangnya window_start serta dimensi pelaporan dalam kunci. Setiap output ialah snapshot hasil pendapatan yang lengkap untuk tetingkap tersebut pada versi semasanya:

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

revision meningkat secara monotonik untuk sesuatu tetingkap, dan sink hanya menerima semakan yang lebih tinggi. EARLY ialah penyegaran satu minit dan tidak mendakwa kelengkapan. ON_TIME bermaksud penanda aras air telah melepasi penghujung tetingkap. FINAL bermaksud tempoh pembetulan automatik 24 jam telah tamat. FINAL ialah kontrak operasi, bukan dakwaan bahawa tiada lagi data wujud; penyelarasan masih mengendalikan rekod di luar had pemotongan.

Terbitkan dasar masa daripada kekangan. Setiap sekatan sumber yang aktif mengekstrak event_time yang telah disahkan. Strategi bounded-out-of-orderness yang biasa ialah:

partition_watermark = max_valid_event_time_seen - out_of_order_bound
operator_watermark = min(active_partition_watermarks)

Pilih out_of_order_bound daripada taburan kelewatan ketibaan yang diperhatikan, kadar data lewat yang boleh diterima, dan sasaran kependaman ON_TIME. Penanda aras air bagi setiap sekatan menghalang sekatan yang pantas daripada mengisytiharkan bahawa sekatan yang perlahan telah selesai. Pengendali berbilang input mengambil kemajuan minimum supaya ia tidak mendahului input yang masih boleh menghasilkan rekod yang lebih lama. Tandakan sekatan sebagai melahu (idle) hanya selepas ia tidak menghasilkan data untuk tempoh selang yang dipersetujui; jika tidak, ia boleh menghentikan penanda aras air global selama-lamanya. Ambang kelalaian yang terlalu singkat menyebabkan rekod lama daripada sekatan yang disambung semula dianggap lewat, jadi output sisi tetap diperlukan.

Tetingkap mempunyai dua keluarga pencetus. Pencetus awal masa pemprosesan mengeluarkan snapshot lengkap semasa sekali setiap minit. Apabila penanda aras air melepasi window_end, keluarkan semakan ON_TIME. Kekalkan keadaan tetingkap sehingga penanda aras air melepasi window_end + 24h; setiap peristiwa lewat yang sah akan mengemas kini agregat dan menghasilkan semakan yang lebih tinggi. Semasa pembersihan, keluarkan FINAL dan lepaskan keadaan tersebut. Model pencetus dan pengumpulan Beam menggambarkan mengapa “bila perlu mengeluarkan” dan “sama ada pane ialah delta atau snapshot terkumpul” merupakan pilihan yang berasingan. Reka bentuk ini menggunakan snapshot terkumpul dan upsert, supaya sistem hiliran tidak perlu memasang semula pane.

Nyahduplikasi mengikut event_id sebelum pengagregatan. Simpan event_id → (event_time, payload_fingerprint). Benarkan peristiwa pertama melaluinya, buang salinan lain dengan cap jari yang sama, dan hantar ID yang sama dengan cap jari berbeza ke strim konflik. Pengekalan bergantung pada selang masa terpanjang di mana hulu boleh memainkan semula salinan lain, bukan hanya pada saiz tetingkap. Semantik penyahduplikasian berasaskan penanda aras air Spark juga memerlukan ambang kelewatan yang lebih panjang daripada jurang cap masa antara duplikasi paling awal dan paling akhir. Pembersihan keadaan yang berlaku terlalu awal membolehkan duplikasi lewat dikira semula.

Anggarkan keadaan secara eksplisit. Jika kemuncak 20,000 peristiwa sesaat berterusan selama 24 jam dan setiap peristiwa adalah unik, kerja tersebut akan mengingati 20,000 × 86,400 = 1,728,000,000 ID. Malah muatan logik ilustrasi sebanyak 40 bait setiap entri menjadikan had atas ini sekitar 69.12 GB. Objek enjin, indeks, bahagian belakang keadaan (state backend), titik semak, dan replika meningkatkan jejak fizikal. Oleh itu, keadaan memerlukan pembahagian berasaskan kunci dan titik semak berperingkat (incremental checkpoints), manakala pengekalan mematuhi kontrak main semula. Jika 99% duplikasi tiba dalam tempoh yang lebih singkat, tempoh penstriman yang lebih pendek boleh digabungkan dengan kunci perniagaan sink dan penyelarasan luar talian, tetapi risiko sisa duplikasi mesti dikira secara kuantitatif dan bukannya disembunyikan di sebalik TTL.

Laluan output mesti bertolak ansur dengan main semula. Sebaik-baiknya, sink mengambil bahagian dalam transaksi titik semak supaya kedudukan sumber, keadaan pengendali, dan output tetingkap dikomit bersama-sama. Jika tidak, jadikan (window_key, revision) sebagai upsert idempoten: memainkan semula semakan yang sama atau lebih lama selepas ranap tidak boleh menulis ganti nilai baharu. Label exactly-once bagi broker tidak mencukupi apabila penulisan pangkalan data luaran berada di luar pengesahan titik semak. Semantik hujung ke hujung bergantung pada sempadan komit yang dikongsi antara sumber, keadaan, dan sink.

Peristiwa yang melebihi 24 jam dihantar ke output sisi too-late dan kekal dalam log mentah. Kerja kelompok harian mengira semula tetingkap yang terjejas mengikut event_time dengan peraturan ID dan jumlah yang sama, kemudian membandingkannya dengan snapshot penstriman FINAL. Perbezaan ketara akan menghasilkan semakan yang lebih tinggi atau memasuki kelulusan kewangan. Kerja kelompok menggantikan tetingkap secara atomik atau melakukan upsert berversi. Menambahkan hasil kelompok pada jumlah sedia ada akan mengira data yang sama dua kali apabila kelompok dimainkan semula.

Gunakan jujukan minimum untuk mengesahkan semantik. Tetingkap [10:00, 11:00) menerima A=100, B=50, dan kemudian salinan A yang serupa. Hasil ON_TIME yang telah dinyahduplikasi ialah 150. Selepas penanda aras air melepasi 11:00, C=20 tiba semasa tempoh kelewatan yang dibenarkan, dan semakan yang lebih tinggi menukar tetingkap kepada 170. D tiba selepas penanda aras air melepasi titik pembersihan dan dihantar ke penyelarasan dan bukannya menyentuh keadaan yang telah dibersihkan secara langsung. Jika muatan kedua A mempunyai jumlah 120, ia dihantar ke strim konflik; hasilnya tidak boleh menjadi 170 atau 190.

Matriks ujian juga merangkumi gangguan urutan merentasi sekatan, sekatan melahu yang menghentikan penanda aras air, kuarantin cap masa masa hadapan, sempadan sejurus sebelum dan selepas penanda aras air, ranap sistem sebelum dan selepas penulisan sink, pemulihan titik semak dan main semula, semakan tidak teratur yang sampai ke sink, serta pelaksanaan penyelarasan berulang. Metrik pengeluaran merangkumi p50/p95/p99 dan kelewatan ketibaan ekor (tail), penanda aras air semasa dan lag bagi setiap pengendali, kiraan dan jumlah early/on-time/late/too-late, padanan penyahduplikasian dan konflik cap jari, bait keadaan, tempoh titik semak, semakan sink yang ditolak, dan perbezaan kelompok-lawan-strim.

Contoh Jawapan yang Kukuh

“Saya akan mentakrifkan angka satu minit sebagai anggaran dan 24 jam sebagai tempoh pembetulan automatik. event_time menetapkan pembelian pada jam UTC; masa pemprosesan hanya memacu pencetus awal satu minit. Setiap sekatan input aktif menjana penanda aras air daripada masa peristiwa sah terbesarnya tolak elaun ketidakteraturan, dan hiliran menggunakan input aktif yang paling perlahan. Sekatan melahu ditandakan secara eksplisit, dan cap masa masa hadapan yang tidak sah dikuarantinkan.

Sebelum pengagregatan, saya menyahduplikasi mengikut event_id, menyimpan masa peristiwa dan cap jari muatan. Percubaan semula yang serupa dikira sekali; ID yang sama dengan kandungan berbeza memasuki strim konflik. Tetingkap jam semula jadi mengeluarkan snapshot EARLY yang lengkap setiap minit, snapshot ON_TIME selepas penghujungnya berada di belakang penanda aras air, dan mengekalkan keadaan selama 24 jam. Peristiwa lewat dalam tempoh tersebut mengemas kini jumlah keseluruhan dan meningkatkan semakan; pembersihan mengeluarkan FINAL.

Sink melakukan upsert mengikut kunci tetingkap dan semakan berbanding menambah setiap snapshot sebagai delta. Sumber boleh dimainkan semula dan keadaan pengendali dipulihkan daripada titik semak. Jika sink tidak boleh menyertai transaksi titik semak, penulisan berversi menghalang main semula pemulihan daripada menggantikan hasil yang lebih baharu. Peristiwa yang lewat melebihi 24 jam dihantar ke output sisi, dan kerja kelompok harian mengira semula log mentah dan membandingkannya dengan FINAL.

Jika kemuncak 20,000 peristiwa sesaat berterusan selama 24 jam dan semua ID adalah unik, had atasnya ialah 1.728 bilion entri penyahduplikasian. Walaupun 40 bait logik setiap entri menghasilkan 69.12 GB, jadi taburan kelewatan yang diukur mesti mewajarkan pengekalan, dan kos keadaan serta titik semak perlu dipantau. Saya akan menyelesaikannya dengan suntikan kegagalan untuk duplikasi, ketidakteraturan urutan, kelewatan, cap masa masa hadapan, sekatan melahu, dan pemulihan, dengan menegaskan bahawa tetingkap akhir adalah sama dengan pengiraan semula luar talian.”

Kesilapan Biasa

• Menetapkan tetingkap perniagaan mengikut masa pemprosesan → peristiwa yang sama boleh berada dalam jam yang berbeza semasa main semula sejarah → gunakan masa peristiwa untuk keahlian tetingkap dan masa pemprosesan hanya untuk output awal.
• Menerangkan penanda aras air sebagai “tiada data lebih lama boleh tiba” → ia biasanya merupakan anggaran kemajuan heuristik, jadi peristiwa yang lebih lama masih boleh muncul → takrifkan kelewatan yang dibenarkan, output sisi too-late, dan penyelarasan.
• Menggunakan satu nilai tanpa penjelasan untuk kelewatan penanda aras air, kelewatan yang dibenarkan, dan TTL penyahduplikasian → ia masing-masing mengawal pencetusan, keadaan tetingkap, dan pengecaman duplikasi → terbitkan nilai tersebut daripada SLO kependaman, tempoh pembetulan, dan kontrak main semula hulu.
• Menambah (append) jumlah pada setiap pencetusan → pencetusan EARLY, ON_TIME, dan late dijumlahkan berulang kali → keluarkan snapshot lengkap dan lakukan upsert mengikut tetingkap dan semakan, atau takrifkan protokol retractable-delta.
• Memajukan penanda aras air daripada masa peristiwa maksimum global → sekatan yang pantas atau cap masa masa hadapan yang salah menyebabkan data sekatan perlahan menjadi lewat terlalu awal → jana kemajuan sekatan, ambil nilai minimum bagi input aktif, dan sahkan cap masa.
• Mengekalkan sekatan yang senyap secara kekal dalam pengiraan minimum → penanda aras air terhenti, menghalang tetingkap dan keadaan penyahduplikasian daripada dibersihkan → gunakan pengendalian kelalaian yang boleh diperhatikan dan halakan data lewat yang disambung semula dengan betul.
• Hanya menyimpan ID peristiwa → penggunaan semula ID oleh pengeluar ditelan secara senyap → simpan juga cap jari muatan dan kuarantinkan konflik.
• Menyatakan “exactly once didayakan” → sink luaran mungkin tidak berkongsi sempadan titik semak → jejaki laluan komit hujung ke hujung dan gunakan sink bertransaksi atau penulisan idempoten berversi.
• Menggugurkan data selepas pembersihan keadaan → strim kelihatan stabil manakala hasil kewangan menjadi tidak dapat dijelaskan → kekalkan peristiwa mentah dan laksanakan output sisi, pengiraan semula kelompok, serta semakan percanggahan.

Soalan Susulan dan Jawapan

Soalan Susulan 1: Berapakah saiz elaun ketidakteraturan (disorder allowance) bagi penanda aras air?

Ukur processing_time - event_time dalam pengeluaran dan bahagikannya mengikut sumber, versi klien, dan rantau. Mula-mula pilih hasil ON_TIME terkini yang boleh diterima dan bahagian peristiwa yang dibenarkan masuk ke dalam laluan pembetulan; kemudian pilih persentil yang memenuhi kedua-duanya. Teruskan memantau p50, p95, p99, dan ekor taburan. Perubahan taburan harus melalui pelepasan konfigurasi dan bukannya membiarkan satu pencilan (outlier) mengubah dasar secara automatik. Penanda aras air mengawal pencetusan tepat pada masanya, manakala tempoh pembetulan 24 jam masih merangkumi ekor taburan yang lebih panjang.

Soalan Susulan 2: Mengapakah satu sekatan Kafka yang senyap boleh menghentikan tetingkap?

Kemajuan selamat bagi pengendali berbilang input ialah penanda aras air input minimum. Sekatan yang tidak maju mengekalkan nilai minimum tersebut tanpa perubahan. Selepas mengesahkan bahawa ia tidak menghasilkan sebarang peristiwa sepanjang ambang kelalaian, tandakannya sebagai melahu supaya ia keluar sementara daripada pengiraan minimum. Jangan tetapkan ambang terlalu singkat: rekod daripada sekatan yang disambung semula boleh menjadi lebih lama daripada penanda aras air baharu. Rekod tersebut mesti memasuki pemprosesan dibenarkan-lewat atau output sisi, dan peralihan status melahu memerlukan metrik tersendiri.

Soalan Susulan 3: Mengapakah keadaan penyahduplikasian 24 jam tidak menjamin bahawa duplikasi adalah mustahil selama-lamanya?

Dua puluh empat jam hanya merangkumi tempoh pembetulan automatik bagi masalah ini. Jika bahagian hulu memainkan semula ID pada jam ke-25, keadaan penstriman telah tiada dan peristiwa tersebut boleh memasuki semula agregat. Penyahduplikasian kekal memerlukan indeks unik perniagaan yang bertahan lebih lama, pendaftar peristiwa yang boleh ditanya, atau penyahduplikasian luar talian penuh; setiap satunya menambah kos storan dan penulisan. Nyatakan jaminan tersebut sebagai “dinyahduplikasi dalam lingkungan ufuk main semula yang diisytiharkan,” kemudian selaras rekod yang tiba kemudian.

Soalan Susulan 4: Bagaimana jika pangkalan data hiliran menyokong INSERT tetapi bukan upsert?

Tulis setiap hasil ke log perubahan yang tidak boleh diubah (immutable changelog) yang berkuncikan tetingkap dan semakan. Lapisan bacaan membina paparan semasa daripada semakan maksimum bagi setiap tetingkap. Pengguna mesti memahami bahawa setiap rekod ialah snapshot lengkap, bukan delta, dan tidak boleh menjumlahkan semua semakan. Jika kos pertanyaan terlalu tinggi, padatkan secara tak segerak (asynchronously) ke dalam jadual perkhidmatan yang menyokong penggantian atomik sambil mengekalkan log versi untuk audit dan pemulihan.

Soalan Susulan 5: Bagaimanakah anda membuktikan bahawa pemulihan tidak terlebih kira atau terkurang kira?

Bagi input yang tetap, rekod set penyahduplikasian yang dijangkakan dan semakan tetingkap. Hentikan kerja pada tiga sempadan: selepas bacaan sumber tetapi sebelum titik semak keadaan, selepas titik semak keadaan tetapi sebelum pengesahan sink, dan selepas penulisan sink tetapi sebelum pengesahan titik semak. Mainkan semula selepas pemulihan dan sahkan bahawa setiap ID peristiwa menyumbang sekali, sink hanya mengekalkan semakan terbesar, dan tetingkap akhir adalah sama dengan pengiraan semula luar talian. Status RUNNING tanpa suntikan kegagalan pada sempadan komit ini bukanlah bukti ketepatan hujung ke hujung.