Stream processing

Fəsil 10-dakı batch job gecə saat üçdə işləyir və səhər hesabat hazır olur. Amma fraud komandası soruşur: "Şübhəli ödənişi 24 saat sonra yox, 2 saniyə sonra görə bilərikmi?" Analitiklər isə dashboard-un dünənki yox, indiki vəziyyəti göstərməsini istəyir. Batch processing-in gizli fərziyyəsi input-un sonlu olmasıdır: job bütün input-u oxuyur, sonra output yazır. Real data isə heç vaxt bitmir, istifadəçilər dünən də əməliyyat edirdi, bu gün də edir. Batch bu sonsuz axını süni şəkildə günlük parçalara bölür.

Stream processing həmin fərziyyədən imtina edir. Data sonsuz, artımlı şəkildə gələn event-lər axınıdır: kiçik, self-contained, immutable record-lar, hər biri müəyyən bir anda baş verən şeyi təsvir edir və adətən timestamp daşıyır. Event producer tərəfindən bir dəfə yaradılır və bir neçə consumer tərəfindən emal oluna bilər. Əlaqəli event-lər topic və ya stream adlanan qrupda toplanır. Bu fəsil event-lərin necə ötürüldüyünü, database-lərlə necə bağlandığını və sonsuz axının üzərində necə hesablama aparıldığını izah edir.

Event-lərin ötürülməsi: messaging sistemləri

Ən sadə yanaşma consumer-in yeni event olub-olmadığını database-dən periodik soruşmasıdır. Polling bahalıdır: tez-tez soruşanda boş sorğular çoxalır, gec soruşanda gecikmə artır. Daha yaxşısı consumer-in xəbər tutmasıdır, yəni messaging system.

İstənilən messaging sistemini iki sualla xarakterizə etmək olar:

1. Producer consumer-dən sürətli olsa nə olur? Üç variant var: mesajları atmaq, queue-da buffer-ləmək, və ya backpressure tətbiq edib producer-i ləngitmək. Buffer-lənirsə, queue böyüyəndə nə olur, diskə yazılır, yoxsa yaddaş bitəndə sistem çökür?
2. Node çökəndə və ya müvəqqəti offline olanda mesajlar itirmi? Durability diskə yazma və replication ilə alınır və qiyməti var.

Birbaşa ötürmə və broker

Producer-dən consumer-ə birbaşa şəbəkə bağlantısı qurmaq olar: UDP multicast, webhook-lar, broker-siz messaging kitabxanaları. Sadədir, lakin consumer offline olanda event itir; fault tolerance tətbiq kodunun üzərinə düşür.

Message broker bu problemi həll edən xüsusi database növüdür: producer mesajı broker-ə yazır, consumer oradan oxuyur. Broker-in olması ilə durability sualı mərkəzləşir, consumer-lər gəlib-gedə bilir, mesajlar onları gözləyir. Lakin broker adi database-dən fərqlidir:

Çoxlu consumer və acknowledgement

Bir topic-i bir neçə consumer oxuyanda iki pattern var. Load balancing-də hər mesaj consumer-lərdən yalnız birinə çatır; emal paralelləşir. Fan-out-da hər mesaj hər consumer-ə çatır; bir event-i bir-birindən xəbərsiz bir neçə sistem emal edir. İkisi birləşə də bilər: iki ayrı consumer qrupu hər mesajı alır, hər qrupun daxilində isə yük bölünür.

Consumer mesajı emal etdikdən sonra acknowledgement göndərir, broker ondan sonra mesajı silir. Ack gəlməsə, broker mesajı başqa consumer-ə yenidən çatdırır və bu, load balancing ilə birləşəndə mesajların sırasını poza bilər. Emal bitib, lakin ack şəbəkədə itibsə, mesaj iki dəfə emal olunur; bu, Fəsil 8-dəki qeyri-müəyyənliyin birbaşa nəticəsidir.

Partition olunmuş log-lar

Ənənəvi broker-də mesaj oxunan kimi silinir, yəni onu ikinci dəfə oxumaq və ya yeni consumer-ə tarixi vermək mümkün deyil. Batch-in gücü isə input-un immutable olmasındadır. İkisini birləşdirən log-based message broker ideyasıdır: Fəsil 3-dəki append-only log, Fəsil 5-dəki replication log-u ilə eyni struktur, lakin messaging üçün.

Producer mesajı log-un sonuna yazır, consumer log-u ardıcıl oxuyur, sona çatanda yeni mesajı gözləyir. Throughput üçün log partition-lara bölünür, hər partition fərqli maşında ola bilər. Partition daxilində hər mesajın monoton artan offset-i var və sıra zəmanətlidir; partition-lar arasında sıra zəmanəti yoxdur.

Load balancing burada fərqli görünür: hər partition consumer group-dakı bir consumer-ə verilir. Nəticədə topic-i oxuyan consumer-lərin sayı partition sayından çox ola bilməz və bir yavaş mesaj öz partition-ının bütün mesajlarını saxlayır. Əvəzində offset modeli ack-ı çox ucuzlaşdırır: broker hər consumer group üçün yalnız bir rəqəm, son emal olunmuş offset-i saxlayır. Consumer çökəndə yenisi həmin offset-dən davam edir; offset yazılmazdan əvvəl çökübsə, bir neçə mesaj təkrar emal olunur, Fəsil 5-dəki log sequence number-in eynisidir.

Log sonsuz böyüyə bilməz. Broker onu segment-lərə bölür və köhnə segment-ləri silir və ya arxivləşdirir. Consumer o qədər geri qalıb ki, oxumadığı segment silinibsə, həmin mesajları itirir. Bu modelin böyük üstünlüyü təkrar oxumaqdır: offset-i geri qaytarıb emalı yenidən işlətmək olur, yəni kod səhvindən sonra bərpa batch-dəki kimi sadədir.

Database-lər və stream-lər

Fəsil 5-də replication log-u gördük: leader-in yazdığı hər dəyişiklik follower-lərə stream kimi gedir. Bu, database-in özündə artıq stream olduğunu göstərir. Real sistemlərdə isə eyni data bir neçə yerdə yaşayır: OLTP database, search index, cache, data warehouse. Onları sinxron saxlamaq lazımdır.

Tətbiqin hər yerə ayrı-ayrı yazması, yəni dual write, sakit bir bəladır. İki client eyni anda database-ə və index-ə fərqli sırada yazsa, iki sistem fərqli vəziyyətdə qalır və heç kim xəbər tutmur. Bir yazma uğurlu, digəri uğursuz olsa da, eyni şey. Fəsil 7-dəki atomic commit bunu həll edərdi, amma heterogen sistemlər arasında bahalıdır.

Change data capture

Daha yaxşı həll bir sistemi system of record seçmək, qalanlarını isə onun dəyişiklik log-undan törətməkdir. Change data capture, yəni CDC, database-in replication log-unu oxuyub hər dəyişikliyi event kimi message broker-ə verir. Search index, cache və warehouse həmin log-u eyni sırada tətbiq edir; onlar artıq database-in follower-ləridir.

Yeni consumer tam tarixi istəyəndə iki yol var: database-in initial snapshot-ını götürüb log-un uyğun offset-indən davam etmək, və ya log compaction tətbiq etmək. Compaction Fəsil 3-dəki kimi hər key üçün yalnız son value-nu saxlayır; nəticədə log-un ölçüsü database-in ölçüsünə yaxınlaşır və onu başdan oxumaq database-in tam kopyasını verir.

Event sourcing

CDC-də event-lər database-in aşağı səviyyəli dəyişiklikləridir: "cədvəl X, sətir Y, yeni dəyər Z". Event sourcing-də isə event tətbiq səviyyəsində mənalıdır: "istifadəçi səbətə məhsul əlavə etdi". Tətbiq state-i dəyişdirmir, baş verənləri yazır; cari state event-lərin üzərindən keçməklə hesablanır.

Burada command ilə event fərqi var. Command istifadəçinin istəyidir və rədd edilə bilər: "bu yeri rezerv et" yerin boş olub-olmadığını yoxlamaq tələb edir. Yoxlama keçəndən sonra fakt kimi event yazılır: "yer rezerv olundu". Event artıq həqiqətdir, consumer-lər onu rədd edə bilməz.

State, stream və immutability

Bütün bunların arxasındakı prinsip sadədir: state dəyişikliklərin inteqrasiyasıdır, dəyişiklik isə state-in törəməsidir.

state(now) = integrate(stream of events)
stream     = derivative(state over time)

Mutable state-i immutable event log-dan törətmək mühasibatlıqdan gəlir: səhv yazılış silinmir, əks yazılışla düzəldilir. Üstünlükləri: audit tarixi, səhv kodu düzəldib state-i yenidən qurmaq, eyni log-dan bir neçə fərqli read view yaratmaq, Fəsil 12-də bu "unbundling" adlanacaq. Yazma və oxuma modelləri ayrılanda, Fəsil 7-dəki bəzi concurrency problemləri də yox olur: tək event atomik yazılır, view-lar ondan törəyir.

Məhdudiyyətləri də var. Tez-tez dəyişən data-da log çox böyüyür, compaction isə səhv edilsə tarixi itirir. Bəzən data-nı həqiqətən silmək lazımdır, məsələn, hüquqi tələblə; immutable log-da silmə "silmək istəyini yazmaq"dan çox, bütün törəmə sistemlərdə həmin data-nı tapıb yenidən yazmaq deməkdir.

Stream-lərin emalı

Stream-i haraya göndərmək olar? Database, cache və ya index-ə yazıb orada sorğulamaq; istifadəçiyə push bildiriş kimi göndərmək; və ya başqa stream-ə çevirmək. Sonuncu Fəsil 10-dakı job-un stream versiyasıdır: operator input stream-i oxuyur, output stream yazır. Fərq ondadır ki, operator heç vaxt bitmir, ona görə sort-merge join mümkün deyil və fault tolerance fərqli qurulmalıdır.

Tipik istifadə halları: complex event processing, yəni stream-də pattern axtarmaq və tapılanda kompleks event yaratmaq; stream analytics, yəni müəyyən pəncərədə statistika hesablamaq; materialised view-ların saxlanması, yəni CDC-dən gələn dəyişiklikləri derived store-a tətbiq etmək; stream üzərində axtarış, yəni əvvəlcədən qeydə alınmış sorğuların hər event-ə tətbiqi.

Vaxt haqqında düşünmək

Batch-də "son beş dəqiqə" event-lərin timestamp-inə görə ölçülür; job dünənki data-nı işlədəndə dünənki beş dəqiqəni hesablayır. Stream-də isə iki saat var: event-in baş verdiyi event time və onun emal olunduğu processing time. Processing time ilə pəncərə hesablamaq aldadıcıdır: consumer bir dəqiqə ləngisə, bir dəqiqəlik event-lər növbəti pəncərəyə düşür və qrafikdə olmayan bir spike görünür.

Event time-ın öz çətinliyi var: pəncərənin bağlandığını necə bilirsiniz? Mobil cihaz offline olub bir saat sonra event-lərini göndərə bilər. Praktik yanaşmalar: straggler-ləri ignor etmək və sayını izləmək, və ya artıq publish olunmuş pəncərə üçün düzəliş yayımlamaq. Hansı saat? Cihaz saatı səhv ola bilər, Fəsil 8-də bunu görmüşdük. Üç timestamp saxlamaq kömək edir: cihaza görə event vaxtı, cihaza görə göndərmə vaxtı, server-ə görə alınma vaxtı; sonuncu ikisinin fərqi cihaz saatının offset-ini təxmin edir.

Pəncərə növləri də fərqlidir:

Stream join-ləri

Üç növ join var. Stream-stream join: axtarış event-ləri ilə klik event-lərini session id-yə görə birləşdirmək; operator hər iki tərəfdən son bir saatın event-lərini index-də saxlamalıdır. Stream-table join: aktivlik event-lərini user profile ilə zənginləşdirmək; cədvəlin lokal kopyası CDC stream-i ilə yenilənir. Table-table join: iki CDC stream-dən materialised view qurmaq, məsələn, timeline cache.

Hamısının ortaq problemi vaxtdan asılılıqdır. Join state-ə baxır, state isə zamanla dəyişir. Satış event-i vergi dərəcəsi cədvəli ilə join olunursa, hansı vergi dərəcəsi: event anındakı, yoxsa emal anındakı? Determinizm lazımdırsa, cədvəlin versiyasını da event-in timestamp-inə görə seçmək lazımdır, data warehouse-larda bu slowly changing dimension adlanır.

Fault tolerance

Batch-də fault tolerance asandır: task uğursuz olanda yenidən işlədirsiniz, output yalnız sonda görünür, kənardan exactly-once kimi görünür. Stream bitmir, ona görə "sonda" yoxdur.

Microbatching stream-i kiçik, məsələn bir saniyəlik, blok-lara bölür və hər birini batch kimi emal edir. Checkpointing daha ümumi yanaşmadır: operator periodik olaraq state-ini durable storage-a yazır; çöküşdən sonra son checkpoint-dən və ona uyğun offset-dən davam edir. Hər iki halda framework daxilində exactly-once alınır, amma output xarici sistemə gedibsə, email göndərilibsə, təkrar emal onu iki dəfə edəcək.

Bunun həlli Fəsil 7-dən tanışdır: output-u və offset-i bir atomic commit-də yazmaq, lakin heterogen sistemlərlə yox, stream framework-ünün daxilində. Alternativ idempotence-dir: əməliyyatı bir neçə dəfə etmək bir dəfə etməklə eyni nəticə verir. Hər event-ə unikal id və offset qoşulsa, yazılan sistem "bu offset artıq tətbiq olunub" deyib təkrarı atır.

State-in bərpası üçün seçimlər: state-i remote store-da saxlamaq və replicate etmək; lokal saxlayıb periodik replicate etmək; və ya əgər input log-u saxlanılırsa, state-i yenidən hesablamaq. Pəncərə kiçikdirsə, sonuncu ən ucuzdur. Fəsil 12 bu parçaları bir yerə yığıb sistem dizaynının gələcəyinə baxır.

Tapşırıqlar

1. Ödəniş event-ləri üçün fraud detection sistemi dizayn edin: eyni kart 10 dəqiqə ərzində üç fərqli ölkədən istifadə olunarsa alert yaranmalıdır. Hansı pəncərə növü lazımdır, event time ilə processing time fərqi nəyə təsir edir, 2 saat gecikən event gələndə nə edirsiniz?
2. Sistemdə OLTP database, search index və cache var və tətbiq hər üçünə ayrıca yazır. Dual write-ın yaratdığı konkret bir uyğunsuzluq ssenarisini addım-addım göstərin, sonra eyni sistemi CDC ilə yenidən qurun. Yeni search index-i sıfırdan qurmaq üçün hansı yol seçərdiniz?
3. Log-based broker-də topic-in 8 partition-ı, consumer group-da isə 3 consumer var. Yük necə paylanır? Bir partition-a hot key düşəndə nə olur? Partition sayını 8-dən 16-ya artırmaq hansı sıra zəmanətlərini pozur və bunu necə idarə edərdiniz?