Stream Processing Pattern

M?i h? th?ng stream processing ??u có s? khác bi?t, t? m? hình ??n gi?n ch? g?m consumer, logic x? ly và producer, ??n các c?m ph?c t?p nh? Spark, Flink Streaming và nhi?u bi?n th? khác ? gi?a. Tuy nhiên, có m?t s? m?u thi?t k? c? b?n, là nh?ng gi?i pháp ?? ???c bi?t ??n cho các yêu c?u ph? bi?n trong ki?n trúc stream processing. Chúng ta s? xem xét m?t s? pattern ph? bi?n này và minh h?a cách chúng ???c s? d?ng th?ng qua m?t vài ví d?.

Single-Event Processing (X? ly t?ng s? ki?n riêng l?)

Pattern c? b?n nh?t trong stream processing là x? ly t?ng s? ki?n m?t cách ??c l?p. ?i?u này còn ???c g?i là map/filter pattern, b?i vì nó th??ng ???c s? d?ng ?? l?c b? nh?ng s? ki?n kh?ng c?n thi?t ho?c bi?n ??i t?ng s? ki?n. (Thu?t ng? map xu?t phát t? m? hình map/reduce, trong ?ó giai ?o?n map chuy?n ??i d? li?u, còn giai ?o?n reduce th?c hi?n t?ng h?p.)

Trong pattern này, ?ng d?ng stream processing s? consume s? ki?n t? m?t stream, ch?nh s?a t?ng s? ki?n, r?i produce k?t qu? ra m?t stream khác. M?t ví d? là m?t ?ng d?ng ??c log t? stream và ghi các s? ki?n ERROR vào m?t stream ?u tiên cao, trong khi các s? ki?n còn l?i ???c ??a vào m?t stream ?u tiên th?p. M?t ví d? khác là m?t ?ng d?ng ??c d? li?u t? stream và chuy?n ??i ??nh d?ng t? JSON sang Avro.

Nh?ng ?ng d?ng này kh?ng c?n duy trì tr?ng thái bên trong ?ng d?ng, vì m?i s? ki?n có th? ???c x? ly ??c l?p. ?i?u này có ngh?a là vi?c kh?i ph?c sau l?i ?ng d?ng ho?c can b?ng t?i tr? nên c?c k? ??n gi?n, b?i kh?ng có tr?ng thái nào c?n ph?c h?i, chúng ta ch? c?n chuy?n s? ki?n sang m?t instance khác c?a ?ng d?ng ?? ti?p t?c x? ly.

Pattern này có th? ???c tri?n khai d? dàng b?ng m?t producer và consumer ??n gi?n, nh? minh h?a trong hình sau.

Processing with Local State (X? ly v?i tr?ng thái c?c b?)

H?u h?t các ?ng d?ng stream processing ??u liên quan ??n vi?c t?ng h?p th?ng tin, ??c bi?t là t?ng h?p theo c?a s? th?i gian (window aggregation). M?t ví d? ?i?n hình là tính giá t?i thi?u và t?i ?a c?a m?t c? phi?u trong m?i ngày giao d?ch và tính trung bình ??ng (MA).

Nh?ng phép t?ng h?p nh? v?y yêu c?u c?n ph?i duy trì tr?ng thái. Trong ví d? trên, ?? tính giá t?i thi?u và giá trung bình trong ngày, chúng ta c?n l?u tr? giá tr? t?i thi?u, t?ng giá tr?, và s? l??ng b?n ghi ?? x? ly tính ??n th?i ?i?m hi?n t?i.

T?t c? ?i?u này có th? ???c th?c hi?n b?ng local state (tr?ng thái c?c b?) thay vì shared state (tr?ng thái chia s?), b?i vì m?i thao tác trong ví d? trên là m?t phép t?ng h?p theo nhóm (group by aggregate). C? th?, chúng ta th?c hi?n t?ng h?p theo t?ng m? c? phi?u, thay vì t?ng h?p trên toàn b? th? tr??ng ch?ng khoán. M?t Kafka partitioner ???c s? d?ng ?? ??m b?o r?ng t?t c? s? ki?n liên quan ??n cùng m?t m? c? phi?u s? ???c ghi vào cùng m?t partition. Khi ?ó, m?i instance c?a ?ng d?ng s? nh?n toàn b? s? ki?n t? các partition ???c gán cho nó (?ay là m?t ??m b?o c?a Kafka consumer). ?i?u này có ngh?a là m?i instance c?a ?ng d?ng có th? duy trì tr?ng thái cho t?p con c?a các m? c? phi?u mà nó ?ang x? ly.

Nh?ng thách th?c khi s? d?ng tr?ng thái c?c b?

Vi?c s? d?ng tr?ng thái c?c b? khi?n ?ng d?ng stream processing tr? nên ph?c t?p h?n ?áng k?. Có m?t s? v?n ?? quan tr?ng c?n gi?i quy?t:

• S? d?ng b? nh?: Ly t??ng nh?t là tr?ng thái c?c b? nên v?a v?i b? nh? c?a instance ?ng d?ng. M?t s? h? th?ng l?u tr? c?c b? có th? ghi d? li?u ra ? ??a khi b? nh? kh?ng ??, nh?ng ?i?u này s? ?nh h??ng ?áng k? ??n hi?u su?t ?ng d?ng.
• Tính b?n v?ng c?a d? li?u (Persistence): C?n ??m b?o r?ng tr?ng thái kh?ng b? m?t khi m?t instance c?a ?ng d?ng b? t?t và có th? kh?i ph?c khi instance ???c kh?i ??ng l?i ho?c thay th?. Kafka Streams gi?i quy?t v?n ?? này khá t?t b?ng cách l?u tr?ng thái c?c b? trong b? nh? th?ng qua RocksDB tích h?p s?n, ??ng th?i ghi d? li?u xu?ng ? ??a ?? ph?c h?i nhanh sau khi restart. Tuy nhiên, t?t c? các thay ??i c?a tr?ng thái c?c b? c?ng ???c ghi vào m?t Kafka topic. N?u m?t node stream b? down, tr?ng thái c?c b? kh?ng b? m?t—nó có th? ???c kh?i ph?c d? dàng b?ng cách ??c l?i các s? ki?n t? Kafka. Ví d?, n?u tr?ng thái c?c b? ?ang l?u tr? “giá t?i thi?u hi?n t?i c?a HPG = 28.5”, thì th?ng tin này c?ng s? ???c ghi vào Kafka ?? sau này có th? tái t?o l?i cache c?c b? t? d? li?u ?ó. Kafka s? d?ng log compaction cho các topic này ?? ??m b?o r?ng chúng kh?ng phát tri?n v? h?n và lu?n có th? tái t?o l?i tr?ng thái m?t cách hi?u qu?.
• Rebalancing: ??i khi, các partition s? ???c phan b? l?i cho m?t consumer khác. Khi ?i?u này x?y ra, instance m?t partition ph?i l?u tr?ng thái cu?i cùng c?a nó, còn instance nh?n partition m?i ph?i bi?t cách kh?i ph?c l?i tr?ng thái chính xác.

Các framework stream processing có m?c ?? h? tr? khác nhau trong vi?c qu?n ly tr?ng thái c?c b?. N?u ?ng d?ng c?a chúng ta c?n duy trì tr?ng thái c?c b?, chúng ta c?n ki?m tra k? các ??m b?o c?a framework.

Multiphase Processing/Repartitioning (X? ly nhi?u giai ?o?n / Phan vùng l?i d? li?u)

Tr?ng thái c?c b? r?t h?u ích khi chúng ta c?n th?c hi?n các phép t?ng h?p theo nhóm (group by aggregate). Nh?ng n?u c?n m?t k?t qu? d?a trên toàn b? d? li?u có s?n thì sao?

Ví d?, gi? s? chúng ta mu?n c?ng b? top 10 c? phi?u t?ng giá m?nh nh?t trong ngày—t?c là 10 c? phi?u có m?c t?ng cao nh?t t? giá m? c?a ??n giá ?óng c?a m?i ngày. R? ràng, b?t k? phép tính nào ???c th?c hi?n c?c b? trên t?ng instance c?a ?ng d?ng c?ng kh?ng ??, vì t?t c? 10 c? phi?u ??ng ??u có th? n?m trong các partition ???c gán cho các instance khác nhau.

Chúng ta c?n m?t cách ti?p c?n hai giai ?o?n. Tr??c tiên, tính toán m?c t?ng/gi?m hàng ngày cho t?ng m? c? phi?u. Vi?c này có th? th?c hi?n trên t?ng instance b?ng local state. Sau ?ó, ghi k?t qu? vào m?t topic m?i v?i ch? m?t partition.

Partition này s? ???c ??c b?i m?t instance duy nh?t c?a ?ng d?ng ?? tìm ra top 10 c? phi?u trong ngày. Topic th? hai, ch? ch?a d? li?u t?ng h?p hàng ngày c?a t?ng m? c? phi?u, s? nh? h?n ?áng k? và có ít l?u l??ng h?n so v?i topic ch?a d? li?u giao d?ch. Do ?ó, m?t instance duy nh?t có th? x? ly nó. ??i khi, có th? c?n thêm m?t s? b??c n?a ?? t?o ra k?t qu? mong mu?n.

So sánh v?i MapReduce

Lo?i hình x? ly nhi?u giai ?o?n này khá quen thu?c v?i nh?ng ai ?? t?ng l?p trình MapReduce, n?i th??ng ph?i th?c hi?n nhi?u b??c reduce liên ti?p. N?u anh em ?? t?ng vi?t MapReduce, anh em s? nh? r?ng c?n m?t ?ng d?ng riêng bi?t cho m?i b??c reduce.

Tuy nhiên, khác v?i MapReduce, h?u h?t các framework stream processing hi?n nay cho phép tri?n khai t?t c? các b??c x? ly trong m?t ?ng d?ng duy nh?t. Framework s? t? ??ng x? ly chi ti?t v? vi?c instance (ho?c worker) nào s? th?c hi?n m?i b??c, giúp quá trình tri?n khai ??n gi?n h?n nhi?u.

X? ly v?i tra c?u bên ngoài: Stream-Table Join

??i khi, stream processing c?n tích h?p v?i d? li?u bên ngoài lu?ng—ví d? nh? xác th?c giao d?ch d?a trên t?p h?p quy t?c ???c l?u trong c? s? d? li?u ho?c b? sung th?ng tin vào d? li?u clickstream b?ng th?ng tin v? ng??i dùng ?? th?c hi?n click.

Cách ti?p c?n tr?c ti?p: Tra c?u tr?c ti?p t? c? s? d? li?u

M?t cách ti?p c?n ??n gi?n ?? làm giàu d? li?u (data enrichment) là:

• V?i m?i s? ki?n click trong stream, tra c?u th?ng tin ng??i dùng trong c? s? d? li?u h? s?.
• Ghi l?i s? ki?n ?? ???c b? sung th?ng tin (bao g?m click g?c, ?? tu?i và gi?i tính c?a ng??i dùng) vào m?t Kafka topic m?i.

V?n ?? c?a cách ti?p c?n này:

1. Hi?u su?t kém do ?? tr? tra c?u: M?i l?n tra c?u d? li?u bên ngoài th??ng m?t 5–15ms, gay ?nh h??ng ?áng k? ??n t?c ?? x? ly.
2. T?i quá m?c lên c? s? d? li?u: Các h? th?ng stream processing th??ng x? ly 100K–500K event/s, trong khi c? s? d? li?u có th? ch? ch?u ???c 10K truy v?n/giay mà v?n gi? hi?u su?t ?n ??nh.
3. V?n ?? v? tính s?n sàng: N?u c? s? d? li?u g?p s? c? ho?c b? quá t?i, ?ng d?ng stream processing c?ng b? ?nh h??ng theo.

Gi?i pháp: L?u tr? d? li?u trong b? nh? cache

?? ??m b?o hi?u su?t và tính s?n sàng, chúng ta c?n l?u tr? d? li?u t? c? s? d? li?u vào b? nh? c?c b? trong ?ng d?ng stream processing. Tuy nhiên, qu?n ly b? nh? cache kh?ng h? ??n gi?n:

• N?u c?p nh?t cache quá th??ng xuyên, ta v?n gay t?i l?n lên c? s? d? li?u.
• N?u c?p nh?t quá ch?m, d? li?u trong cache s? b? l?i th?i.

Gi?i pháp t?i ?u: Change Data Capture (CDC)

Thay vì truy v?n c? s? d? li?u tr?c ti?p, chúng ta có th? l?ng nghe thay ??i t? c? s? d? li?u nh? m?t dòng s? ki?n (stream of events).

• K? thu?t này g?i là Change Data Capture (CDC)—m?t cách ?? theo d?i thay ??i trong c? s? d? li?u và chuy?n ??i chúng thành stream c?a các s? ki?n thay ??i d? li?u.
• Kafka Connect h? tr? nhi?u connector ?? th?c hi?n CDC và chuy?n ??i b?ng d? li?u trong c? s? d? li?u thành m?t stream s? ki?n.

Khi ?ó, ?ng d?ng stream processing s?:

1. Duy trì m?t b?n sao c?c b? c?a b?ng d? li?u.
2. C?p nh?t b?n sao này m?i khi có s? ki?n thay ??i t? c? s? d? li?u.
3. Khi có s? ki?n click m?i, ch? c?n tra c?u trong b? nh? c?c b? thay vì truy v?n c? s? d? li?u, giúp gi?m t?i ?áng k?.

K?t h?p Stream và Table: Stream-Table Join

Vì m?t trong các stream ch?a th?ng tin c?p nh?t c?a b?ng d? li?u c?c b?, ta g?i ph??ng pháp này là Stream-Table Join.

• Thay vì truy v?n d? li?u bên ngoài, ta k?t h?p d? li?u tr?c ti?p trong b? nh? c?c b?.
• Cách này m? r?ng quy m? t?t h?n và kh?ng làm ?nh h??ng ??n hi?u su?t c?a c? s? d? li?u c?ng nh? các ?ng d?ng khác ?ang s? d?ng nó.

Table-Table Join (K?t h?p Hai B?ng)

Chúng ta ?? th?o lu?n v? cách m?t b?ng (table) và m?t lu?ng s? ki?n c?p nh?t (stream of update events) th?c ch?t là t??ng ???ng nhau. Chúng ta c?ng ?? xem xét chi ti?t cách k?t h?p m?t stream v?i m?t table. Kh?ng có ly do gì mà chúng ta kh?ng th? áp d?ng nguyên t?c t??ng t? khi k?t h?p hai b?ng l?i v?i nhau.

??c ?i?m c?a Table-Table Join

• Kh?ng có c?a s? th?i gian (nonwindowed): K?t h?p hai b?ng lu?n d?a trên tr?ng thái hi?n t?i c?a c? hai b?ng t?i th?i ?i?m th?c hi?n phép join.
• Hi?u su?t cao nh? phan vùng h?p ly: Trong Kafka Streams, ta có th? th?c hi?n equi-join, t?c là c? hai b?ng ??u có cùng khóa chính (key) và ???c phan vùng gi?ng nhau. ?i?u này giúp phép join ???c phan tán hi?u qu? trên nhi?u máy ch? và phiên b?n ?ng d?ng, t?ng hi?u su?t x? ly.

H? tr? Foreign-Key Join

Ngoài equi-join, Kafka Streams c?ng h? tr? foreign-key join, t?c là khóa chính c?a m?t b?ng có th? k?t h?p v?i m?t tr??ng b?t k? c?a b?ng khác.

Streaming Join (K?t h?p hai lu?ng s? ki?n)

??i khi, chúng ta kh?ng ch? mu?n k?t h?p m?t stream v?i m?t table, mà mu?n k?t h?p tr?c ti?p hai lu?ng s? ki?n v?i nhau.

Th? nào là real event stream?

• Streams là kh?ng gi?i h?n (unbounded).
• Khi s? d?ng stream ?? bi?u di?n table, ta có th? b? qua ph?n l?n l?ch s? và ch? quan tam ??n tr?ng thái hi?n t?i.
• Tuy nhiên, khi k?t h?p hai stream, ta ph?i xét toàn b? l?ch s? c?a chúng, ghép n?i các s? ki?n có cùng key và x?y ra trong cùng m?t c?a s? th?i gian.
• Do ?ó, streaming join còn ???c g?i là windowed join.

Ví d? v? Streaming Join

Gi? s? chúng ta có:

• M?t lu?ng ch?a các truy v?n tìm ki?m mà ng??i dùng nh?p vào trang web.
• M?t lu?ng khác ch?a s? ki?n click, ghi l?i nh?ng k?t qu? mà ng??i dùng nh?p vào sau khi tìm ki?m.

Chúng ta mu?n ghép n?i các truy v?n v?i k?t qu? mà ng??i dùng ?? click ?? bi?t k?t qu? nào ???c quan tam nh?t ??i v?i t?ng truy v?n.

• R? ràng, chúng ta c?n k?t h?p các s? ki?n d?a trên t? khóa tìm ki?m.
• Tuy nhiên, chúng ta ch? quan tam ??n các s? ki?n x?y ra trong m?t kho?ng th?i gian ng?n, vì ng??i dùng th??ng click vào k?t qu? ch? vài giay sau khi tìm ki?m.
• Vì v?y, chúng ta s? gi? m?t c?a s? th?i gian ng?n trên m?i lu?ng và ghép n?i các s? ki?n trong c?a s? này.

Cách Kafka Streams th?c hi?n Streaming Join

• Kafka Streams h? tr? equi-join, trong ?ó c? lu?ng truy v?n và lu?ng click ??u ???c partition theo cùng m?t key (ví d?: user_id).
• Ví d?, t?t c? s? ki?n click c?a user_id:42 s? n?m trong partition 5 c?a clicks topic, và t?t c? s? ki?n tìm ki?m c?a user_id:42 c?ng n?m trong partition 5 c?a search topic.
• Kafka Streams ??m b?o r?ng partition 5 c?a c? hai ch? ?? ??u ???c gán cho cùng m?t tác v? (task).
• Tác v? này s? duy trì c?a s? join cho c? hai ch? ?? b?ng RocksDB nhúng trong b? nh? c?c b?.
• Nh? ?ó, nó có th? th?c hi?n phép join chính xác theo th?i gian th?c.

Out-of-Sequence Events (X? Ly S? Ki?n ??n Kh?ng Theo Th? T?)

X? ly các s? ki?n ??n kh?ng theo th? t? th?i gian là m?t thách th?c kh?ng ch? trong stream processing mà còn trong các h? th?ng ETL truy?n th?ng.

S? ki?n ??n sai th? t? x?y ra nh? th? nào?

• ?ay là tình hu?ng khá ph? bi?n, ??c bi?t trong các h? th?ng IoT
• Ví d?: Thi?t b? di ??ng m?t k?t n?i WiFi trong vài gi?, sau ?ó g?i hàng lo?t s? ki?n ?? thu th?p khi k?t n?i tr? l?i. Thi?t b? m?ng b? l?i (ví d?: switch h?ng) kh?ng g?i tín hi?u ch?n ?oán cho ??n khi ???c s?a ch?a. S?n xu?t c?ng nghi?p, n?i m?ng k?t n?i trong nhà máy th??ng kh?ng ?n ??nh, ??c bi?t ? các n??c ?ang phát tri?n.

?ng d?ng stream processing c?n làm gì ?? x? ly tình hu?ng này?

1. Nh?n di?n s? ki?n ??n sai th? t?
2. Xác ??nh kho?ng th?i gian có th? ch?p nh?n s? ki?n ??n mu?n
3. Có c? ch? x? ly s? ki?n mu?n ngay trong quá trình x? ly liên t?c
4. C?p nh?t k?t qu?

H? tr? x? ly s? ki?n kh?ng theo th? t? trong Kafka Streams

• Google Dataflow và Kafka Streams h? tr? event time ??c l?p v?i processing time.
• Các framework này cho phép:
• Kafka Streams API lu?n ghi k?t qu? t?ng h?p vào các result topics.

Reprocessing (X? ly l?i d? li?u)

M? hình quan tr?ng cu?i cùng trong stream processing là x? ly l?i s? ki?n. Có hai tr??ng h?p ph? bi?n:

1. C?p nh?t ?ng d?ng v?i phiên b?n m?i
2. S?a l?i và tính toán l?i k?t qu?

Cách th?c hi?n v?i Kafka Streams

Tr??ng h?p 1: Ch?y song song hai phiên b?n ?ng d?ng

Kafka l?u tr? toàn b? s? ki?n trong th?i gian dài, nên có th? th?c hi?n nh? sau:

• Ch?y phiên b?n m?i c?a ?ng d?ng v?i m?t consumer group m?i.
• C?u hình ?ng d?ng m?i ?? b?t ??u t? offset ??u tiên c?a topic ??u vào.
• Ti?p t?c x? ly ??n khi ?ng d?ng m?i b?t k?p ?ng d?ng c?.
• Khi s?n sàng, chuy?n client sang dùng lu?ng k?t qu? m?i.

Tr??ng h?p 2: Reset ?ng d?ng ?? x? ly l?i t? ??u

• Reset ?ng d?ng ?? b?t ??u x? ly l?i t? ??u các topic ??u vào.
• Xóa local state ?? tránh tr?n l?n k?t qu? gi?a phiên b?n c? và m?i.
• Có th? c?n xóa output stream c? ?? tránh nhi?u d? li?u.
• Kafka Streams có c?ng c? h? tr? reset state, nh?ng cách an toàn h?n là ch?y song song hai phiên b?n thay vì reset tr?c ti?p, vì: Có th? so sánh k?t qu? gi?a hai phiên b?n. Kh?ng có r?i ro m?t d? li?u ho?c l?i khi d?n d?p.

Truy v?n tr?ng thái ?ng d?ng (Interactive Queries)

Các ?ng d?ng stream processing có tr?ng thái c?c b?, phan tán trên nhi?u instance.

• Th?ng th??ng, k?t qu? ???c l?y t? output topic.
• Tuy nhiên, trong m?t s? tr??ng h?p, có th? ??c tr?c ti?p t? state store ?? t?i ?u hi?u su?t.
• Ví d?: n?u có m?t b?ng top 10 sách bán ch?y nh?t, thay vì ??c t? lu?ng k?t qu? liên t?c, có th? truy v?n tr?c ti?p b?ng ?ó.

Kafka Streams cung c?p API linh ho?t ?? truy v?n tr?ng thái c?a ?ng d?ng stream processing, giúp l?y d? li?u nhanh h?n mà kh?ng c?n ??i qua topic trung gian.