Ағынды өңдеу

Ағынды өңдеу дегеніміз не?

Ағынды өңдеу - бұл жағдайлардың ең аз кідірісі мен дұрыстығына кепілдік бере отырып, оқиғалардың шексіз тізбектілігіне үздіксіз реакция (транзакциялар журналы, кликтер, төлемдер, телеметрия). «Кезең ішінде жинақталған барлық заттарды аламыз» деген батшқа қарағанда, ағын деректерді түсуіне қарай өңдейді, жай-күйін ұстап тұрады және оқиғаның уақытын ескереді.

Негізгі ұғымдар

Оқиға (event) - 'event _ time' және бірегей 'event _ id' бар өзгермейтін факт.
Оқиға уақыты (event time) vs өңдеу уақыты (processing time) - біріншісі қайнар көзден келеді, екіншісі - оператор оқиғаны нақты көргенде.

Терезелер (windows) - оқиғаларды уақыт бойынша топтау:

Tumbling (жабылмайтын), Hopping/Sliding (жабылатын), Session (белсенділігі бойынша ажырау).
Су белгілері (watermarks) - терезелерді жабуға және кешігіп қалған деректерді күтуді шектеуге мүмкіндік беретін «оқиғалар Т сәтіне дейін келді» деген баға.
Кешігетін деректер (lateness) - 'event _ time' оқиғалары ағымдағы watermark кем; пысықтау ережелері жиі қолданылады.
Жай-күйі (state) - агрегаттарға, join 'oларға, дедупликацияларға арналған операторлардың жергілікті кестелері/қоймалары (keyed state).
Backpressure - downstream өткізу қабілетінің артуы кезіндегі қысым; хаттамамен және буферлермен басқарылады.

Сәулет негіздері

1. Дереккөз (source): оқиғалар брокері (Kafka/NATS/Pulsar), ДБ-дан CDC, кезектер, файлдар/лог-коллекторлар.
2. Ағындық қозғалтқыш: терезелерді, агрегаттарды, джойндарды, паттерндерді (CEP) есептейді, күй мен checkpoint's басқарады.
3. Қабылдағыш (sink): OLTP/OLAP ДБ, іздеу қозғалтқышы, кэш, топиктер, витриналарға/есептерге арналған сақтау орны.
4. Схемалар тізілімі: payload эволюциясын және үйлесімділігін бақылау.
5. Бақылау қабілеті: метрика, трейсинг, лагтар, дашбордтар және су белгілері.

Уақыт семантикасы және тәртіп

Әрқашан event time дегенді қалаңыз: бұл кідірістер мен іркілістерде жалғыз инвариант.
Оқиғалар тәртіптен тыс болуы мүмкін; тәртіп тек партия кілті шегінде ғана кепілдендіріледі.

Watermarks мүмкіндік береді:

терезелерді жабуға және нәтижелерді шығаруға;
кешігіп бара жатқан оқиғаларды қанша күткенімізді шектеу ('allowed _ lateness').
Кешігіп қалған оқиғалар үшін retractions/upserts пайдаланыңыз: агрегаттарды және түзетуші оқиғаларды қайта есептеу.

Жай-күйі мен сенімділігі

Keyed state: агрегаттардың деректері (сомалар, есептеуіштер, дедуп құрылымдары) шардингпен кілттер бойынша бөлінген.
Checkpoint/Savepoint: қалпына келтіру үшін күй түсірілімдері; savepoint - код нұсқасының көші-қоны үшін басқарылатын сурет.

Exactly-once әсері бойынша қол жеткізіледі:

транзакциялық «оқыды-өңдеді-жазды» (commit sink + оқу орны);
демпотенттік sinks (upsert/merge) + дедуп кестелері;
агрегаттарды нұсқалау (optimistic concurrency).

Терезелер, агрегаттар, join's

Терезелер:

Tumbling: қарапайым мерзімді есептер (минуттық, сағаттық).
Hopping/Sliding: «жылжымалы» метриктер (1 мин қадаммен 5 мин бұрын).
Session: теңшелетін сессиялар мен антифрод үшін табиғи.
Aggregations: sum/count/avg/approx-distinct (HyperLogLog), percentiles (TDigest/CKMS).
Stream-Stream join: екі тарапты да кілт пен уақыт бойынша буферлеуді талап етеді, 'allowed _ skew'.
Stream-Table join (KTable): анықтаманы немесе ағымдағы күйді қосу (мысалы, «пайдаланушының белсенді шектері»).

Кешігетін және қайталанатын деректермен жұмыс

Дедупликация: 'event _ id' немесе '(producer_id, sequence)' бойынша; TTL бар «көрінетін» кілттерді қайталау терезесі ≥ сақтаңыз.
Late events: жабылғаннан кейін 'X' ішінде терезені өңдеуге рұқсат етіңіз (retractions/upserts).
Жалған телнұсқалар: агрегаттарды іспеттес түзетіңіз және логтарға «ALREADY_APPLIED» белгілеңіз.

Масштабтау және өнімділік

Кілт бойынша шардалау: параллелизмді қамтамасыз етеді; «ыстық» кілттерді қадағалаңыз.
Backpressure: Параллельділікті шектеңіз, жариялау кезінде батшы мен компрессияны пайдаланыңыз.
Су белгілері: тым агрессивті орнатпаңыз - қатты watermarks күтуді қысқартады, бірақ late жаңартуларының үлесін арттырады.
Күйі: (RocksDB/state store/жадында) пішімін өлшемі мен қол жеткізу үлгілерін ескере отырып таңдаңыз; TTL тазалау.
Автоматты масштабтау: лаг, CPU, state өлшемі, GC уақыты бойынша.

Сенімділік және қайта іске қосу

Идемпотенттік sink немесе офсетті бекітумен транзакциялық коммит - дұрыстықтың негізі.
Қайта іске қосқаннан кейін қайта өңдеуге жол беріледі; әсері «тура бір рет» болуы тиіс.
DLQ/parking lot: проблемалық жазбаларды себептермен жеке ағынға жіберіңіз; қайта өңдеуді қамтамасыз етіңіз.

Бақылау (не өлшеу керек)

Дереккөздер бойынша (уақыт бойынша және хабарлама бойынша).
Watermark/current event time және late оқиғаларының үлесі.
Throughput/latency операторлары, p95/p99 end-to-end.
State size/rocksdb I/O, checkpoint жиілігі/ұзақтығы.
DLQ rate, дедупликация/ретрай пайызы.
CPU/GC/heap, үзіліс уақыты.

Қауіпсіздік және комплаенс

Деректерді жіктеу: PII/PCI сызбаларда белгілеңіз, минимумды сақтаңыз, state және snapshots шифрлаңыз.
Access control: state және sinks топиктеріне жеке ACL.
Ретенциялар: заңдық талаптарға сәйкес келеді (GDPR/ұмыту құқығы).
Аудит: логин 'event _ id', 'trace _ id', нәтиже: 'APPLIED/ALREADY _ APPLIED/RETRIED'.

Енгізу үлгілері

1. CDC → қалыпқа келтіру → домен оқиғалары: ДҚ-ның шикі өзгерістерін көрсетпеңіз, түсінікті бизнес-фактілерге көшіріңіз.
2. Outbox продюсерлерде: транзакция фактісі + оқиға - бір БД-транзакцияда.
3. Core vs Enriched: сыни ағында ең аз payload, байыту - асинхронды.
4. Replay-достық: проекциялар/витриналар логтан қайта жинақталуы тиіс.
5. Idempotency by design: operation/event key, upsert-схемалар, агрегаттар нұсқалары.

Тестілеу

Unit/Property-based: агрегаттар мен түрлендірулердің инварианттары.
Stream tests: out-of-order және дубликаттармен бекітілген оқиғалар ағыны → терезелерді және атаны тексеру.
Golden windows: эталондық терезелер/агрегаттар және рұқсат етілген кеш түзетулер.
Fault-injection: «әсер жазды» және «коммитал офсет» арасындағы құлау.
Replay tests: сөренің басынан қайта жиналуы = ағымдағы күйі.

Құны және оңтайландыру

Терезелер мен watermark кідіріске/ресурстарға әсер етеді: терезе қаншалықты ұзын және 'allowed _ lateness' көп болса, соғұрлым көп state.
Кодекстер және компрессия: CPU/желісін теңгеріңіз.
Шығуда Batching: желілік қоңыраулар мен транзакциялар аз.
Ерте сүзу («pushdown»): артық нәрсені мүмкіндігінше дереккөзге жақын тастаңыз.

Антипаттерндер

Processing time - бұл event time → дұрыс емес талдау.
Синк → реставрациялар кезінде қосарланған әсерлер.
Жаһандық «мега-кілттер»: бір ыстық бөлім параллелизмді бұзады.
Көпшілік оқиғалар ретінде шикі CDC: БД схемаларының жылыстауы, эволюция кезіндегі морт.
DLQ жоқ: «улы» хабарлар бүкіл конвейерді бұғаттайды.
Watermark орнына бекітілген қатаң кідіріс: мәңгілік күту немесе деректерді жоғалту.

Домен мысалдары

Төлемдер/қаржы

'payment.' ағыны, антифрод терезесі (session + CEP), 'operation _ id' дедупы.
Exactly-once әсері бухгалтерлік ledger (upsert + нұсқа).

Маркетинг/жарнама

Sliding терезе CTR/конверсиялар, Join басу және '± Δ t' рұқсатымен көрсету, биддинг үшін агрегаттау.

iGaming/онлайн сервистер

Нақты-уақыт балансы/лимиттері, миссиялар/ашивкалар (session терезелер), антифрод-паттерндер және хабарландырулар.

Шағын үлгілер (жалған құжат)

Су белгілері және late-жаңартулары бар терезе

pseudo stream
.withEventTime(tsExtractor)
.withWatermark(maxAllowedLag=2m)
.window(TUMBLING, size=1m, allowedLateness=30s)
.keyBy(user_id)
.aggregate(sum, retractions=enable)
.sink (upsert_table )//idempotent upsert by (user_id, window_start)

Офсетті бекіткен транзакциялық sink

pseudo begin tx upsert target_table using event, key=(k)
update consumer_offsets set pos=:pos where consumer=:c commit

Шығарылған чек-парақ

event time және watermark стратегиясы анықталды; таңдалған және 'allowed _ lateness'.
Идемпотенттік sink немесе офсетті транзакциялық коммит.
Схемалар тізілімі мен үйлесімділік режимдері енгізілген; аддитивті эволюция.
Өлшемдер: лаг, watermark, p95/p99, DLQ, state өлшемі, checkpoint ұзақтығы.
Тесттер: out-of-order, дубликаттар, қайта іске қосу, replay.
Стате және снапшоттар үшін PII/ретенция саясаты.
Масштабтау жоспары және backpressure стратегиясы.
Терезе шарттары және түзетулер бойынша құжаттама (late updates).

FAQ

Event time қажет пе?
Егер метриканың дұрыстығы мен келісімділігі маңызды болса, иә. Processing time техникалық есеп/мониторинг үшін қолайлы, бірақ талдаманы бұрмалайды.

exactly-once қажет пе?
Нүктелі: сындарлы әсерлер үшін. Көбінесе at-least-once + іспеттес sink жеткілікті.

Терезелерді қалай таңдауға болады?
Бизнес-SLA-дан бастау: «соңғы 5 минут ішінде» → hopping, «пайдаланушы сессиялары» → session, «минуттық есептер» → tumbling.

Соңғы мәліметтермен не істеу керек?
Шектелген 'allowed _ lateness' дегенге рұқсат ету және түзетулерді шығару (upsert/retract). Клиенттік витринаны жаңарта білу керек.

Жиынтығы

Ағынды өңдеу - бұл төмен кідіріс қана емес, уақыттың, жағдайдың және келісімшарттардың тәртібі. event time, терезелер мен су белгілерін дұрыс таңдау, оған қоса демпотенттік әсерлер, бақылау және тестілер конвейерді сенімді, қайталанатын және үнемді етеді - және бизнеске «түннен кейін» емес, «осы жерде және қазір» шешімдерін береді.

Ағынды өңдеу