Потоци от събития

Накратко: Потокът от събития, като Kafka, е дневник, в който всяко събитие се записва веднъж, а всяка система го чете със своето темпо.

Проблем

Една поръчка, четири системи

Клиент 7 поръчва за 12,40 €. Складът трябва да приготви пакета, счетоводството да издаде фактура, пощата да прати имейл, а анализите да преброят продажбата.

Ако кодът на поръчката вика четирите системи една след друга, пощенският сървър, който не отговаря, проваля цялата поръчка. Или поръчката минава, а имейлът се губи. Всяка нова система (например точки за лоялност) значи промяна в кода на поръчката.

Таблицата orders пази какво е поръчката сега. Тя не пази кой вече е разбрал за нея и какво е станало преди. Трябва място, където магазинът казва веднъж „поръчката е направена“, а всеки, който се интересува, го прочита, когато може.

Идеята

Дневник от събития

Събитие е нещо, което вече е станало: „поръчка 61 е направена“, „поръчка 61 е платена“. Магазинът го записва в края на дневник и забравя за него. Складът, фактурите и пощата четат дневника всеки със своето темпо. Ако пощата е спряла за час, тя продължава оттам, докъдето е стигнала, и нищо не се губи.

Такъв дневник е поток от събития (event stream). Kafka го пази в теми (topic), например orders. Темата е разделена на дялове (partition), за да се пише и чете от много машини наведнъж. Всеки запис получава пореден номер в своя дял: отместване (offset). Номерата само растат.

поток от събития (като Kafka)симулатор

create topic orders partitions 3
produce orders key=customer-7 {"order":61,"total":12.40}
produce orders key=customer-2 {"order":62,"total":5.80}
produce orders key=customer-7 {"order":63,"total":31.90}
read orders

create topic orders partitions 3 produce orders key=customer-7 {"order":61,"total":12.40} produce orders key=customer-2 {"order":62,"total":5.80} produce orders key=customer-7 {"order":63,"total":31.90} read orders

▶ ИзпълниВърни командитеНова база

produce записва събитие и казва в кой дял и под кой номер е отишло. read показва всичко: p1:0 е дял 1, номер 0. На картинката всеки ред е дял, клетките са записите с номерата им (под номера е краят на ключа), а клетката с пунктир е мястото на следващия запис.

Записите не се променят и не се трият: команди като update и delete дават грешка. Ако нещо се поправи, това е ново събитие.

По-точно, за любопитните

Командите на симулатора са по-четим вариант на инструментите на Kafka (kafka-topics, kafka-console-producer, kafka-console-consumer, kafka-consumer-groups):

• create topic orders partitions 3: нова тема с 3 дяла; topics изброява темите;
• produce orders key=customer-7 {"order":61,"total":12.40}: записва събитие; ключът е по желание;
• read orders [partition 1] [from 0]: показва записите, без да помни нищо;
• consume orders group billing [as billing-2] [max 5] [from-beginning] [no-commit]: чете като група и записва докъде е стигнала;
• offsets orders: краят на всеки дял и мястото на всяка група;
• describe group billing: мястото, краят и изоставането на групата във всеки дял; groups изброява групите;
• reset group billing to 0 (или earliest, latest): мести мястото на групата.

no-commit е само на симулатора: показва потребител, който е прочел, но е паднал, преди да запише мястото си. Симулаторът чете дял по дял (първо 0, после 1, после 2). Истинският потребител взима порции от няколко дяла наведнъж и редът между тях не е определен.

Виж го

Редът се губи между дяловете

Без ключ записите се разпределят между дяловете (симулаторът ги редува: 0, 1, 2, 0). Поръчка 61 е направена, платена и изпратена. Пощата чете:

поток от събития (като Kafka)симулатор

# Събитията на поръчки 62 и 61, без ключ
produce orders {"order":62,"event":"placed"}
produce orders {"order":61,"event":"placed"}
produce orders {"order":61,"event":"paid"}
produce orders {"order":61,"event":"shipped"}
consume orders group email from-beginning

# Събитията на поръчки 62 и 61, без ключ produce orders {"order":62,"event":"placed"} produce orders {"order":61,"event":"placed"} produce orders {"order":61,"event":"paid"} produce orders {"order":61,"event":"shipped"} consume orders group email from-beginning

▶ ИзпълниВърни командитеНова база

Пощата вижда „shipped“ на поръчка 61 преди „placed“ и праща „пратката ти е на път“ за поръчка, за която още не знае. Редът е гарантиран само вътре в един дял.

Затова се дава ключ. Програмата, която пише, смята от ключа число (хеш) и взима остатъка при деление на броя дялове. Същият ключ винаги дава същия дял. Хешът на симулатора е сборът от кодовете на знаците: за customer-7 той е 982, а 982 при деление на 3 дава остатък 1, затова дял 1. Kafka ползва друг хеш (murmur2), но идеята е същата.

Ключът е това, чийто ред има значение: номерът на поръчката за събитията на поръчката или клиентът, ако е важен редът на всичките му поръчки.

Пробвай

Събитията на една поръчка

Темата orders има 3 дяла. Запиши трите събития на поръчка 61 подред (placed, paid, shipped) с ключ order-61, така че всеки потребител да ги прочете в този ред.

поток от събития (като Kafka)симулатор

produce orders {"order":61,"event":"placed"}

produce orders {"order":61,"event":"placed"}

Провери▶ Изпълни

Идеята

Групи потребители

Всяка система чете като група потребители (consumer group): billing, email, warehouse. Групата помни докъде е стигнала във всеки дял. След четене тя записва това място (commit), за да продължи оттам и след рестарт. Две групи не си пречат: всяка получава всяко събитие.

поток от събития (като Kafka)симулатор

consume orders group billing from-beginning
consume orders group email from-beginning max 2
describe group email

consume orders group billing from-beginning consume orders group email from-beginning max 2 describe group email

▶ ИзпълниВърни командитеНова база

billing е прочела всичко, а email само два записа. Разликата между края на дяла и мястото на групата е изоставането (lag). На картинката всяка група има свой ред под всеки дял: оцветеното е прочетено, а ▲ е мястото, откъдето ще чете. Растящо изоставане значи, че групата не смогва.

Вътре в една група дяловете се поделят. Ако фактурите се бавят, пускаш второ копие със същата група, и всяко чете своите дялове:

поток от събития (като Kafka)симулатор

consume orders group warehouse as warehouse-1 from-beginning
# Второ копие на склада влиза в същата група
consume orders group warehouse as warehouse-2
produce orders key=customer-7 {"order":65,"total":4.30}
produce orders key=customer-2 {"order":66,"total":19.70}
consume orders group warehouse as warehouse-1
consume orders group warehouse as warehouse-2
describe group warehouse

consume orders group warehouse as warehouse-1 from-beginning # Второ копие на склада влиза в същата група consume orders group warehouse as warehouse-2 produce orders key=customer-7 {"order":65,"total":4.30} produce orders key=customer-2 {"order":66,"total":19.70} consume orders group warehouse as warehouse-1 consume orders group warehouse as warehouse-2 describe group warehouse

▶ ИзпълниВърни командитеНова база

Затова броят дялове е горната граница: при 3 дяла четвъртото копие стои без работа.

Нова група без from-beginning започва от края: вижда само събитията от този момент нататък. from-beginning важи само за група, която още няма записано място.

Отделът за анализи започва днес и иска всички поръчки досега. Прочети ги с нова група analytics.

поток от събития (като Kafka)симулатор

describe group billing

describe group billing

Провери▶ Изпълни

Виж го

Две фактури за една поръчка

Групата записва мястото си след като е прочела. Ако потребителят падне между обработката и записа, след рестарта получава същите събития отново:

поток от събития (като Kafka)симулатор

# billing чете две поръчки и пада, преди да запише докъде е стигнала
consume orders group billing from-beginning max 2 no-commit
# Рестарт: пак от последното записано място
consume orders group billing from-beginning max 2

# billing чете две поръчки и пада, преди да запише докъде е стигнала consume orders group billing from-beginning max 2 no-commit # Рестарт: пак от последното записано място consume orders group billing from-beginning max 2

▶ ИзпълниВърни командитеНова база

Поръчки 64 и 61 идват два пъти. Ако billing праща фактура при всяко четене, клиентът получава две. Затова потребителят трябва да е идемпотентен: второто прилагане на същото събитие не променя нищо. Най-честият начин е базата сама да откаже второто:

SQLбаза: магазин

-- Същото събитие идва два пъти: фактурата се пише веднъж
INSERT INTO invoices (order_id, total) VALUES (61, 12.40)
  ON CONFLICT (order_id) DO NOTHING;
INSERT INTO invoices (order_id, total) VALUES (61, 12.40)
  ON CONFLICT (order_id) DO NOTHING;
SELECT order_id, total FROM invoices;

-- Същото събитие идва два пъти: фактурата се пише веднъж INSERT INTO invoices (order_id, total) VALUES (61, 12.40) ON CONFLICT (order_id) DO NOTHING; INSERT INTO invoices (order_id, total) VALUES (61, 12.40) ON CONFLICT (order_id) DO NOTHING; SELECT order_id, total FROM invoices;

▶ ИзпълниВърни заявкатаНова база

UNIQUE по номера на поръчката и ON CONFLICT DO NOTHING правят второто събитие безвредно.

По-точно, за любопитните

„Точно веднъж“ (exactly once) е трудно. Обратният ред (първо записваш мястото, после обработваш) дава „най-много веднъж“: при падане събитието се губи. Затова обикновено се избира „поне веднъж“ (at least once) плюс идемпотентен потребител. Kafka има идемпотентен производител и транзакции, които пазят „точно веднъж“ при четене и запис в самия Kafka. Имейл, който е излязъл, никоя транзакция не връща.

Идеята

Пак от началото

Отчетът на анализите е сбъркан: забравил е доставката. Поправката е готова, но старите поръчки вече са прочетени. Дневникът пази събитията, затова групата може да се върне назад и да ги прочете пак (replay):

поток от събития (като Kafka)симулатор

describe group analytics
# Отчетът е сбъркан. След поправката групата чете всичко наново
reset group analytics to 0
consume orders group analytics

describe group analytics # Отчетът е сбъркан. След поправката групата чете всичко наново reset group analytics to 0 consume orders group analytics

▶ ИзпълниВърни командитеНова база

reset group analytics to 0 мести мястото на групата в началото на всеки дял, а следващото четене минава всичко наново. Другите групи не усещат нищо. from-beginning тук не помага: групата вече има място и то печели.

Повторното четене е безопасно само за идемпотентен потребител. Отчет, който се смята наново, е наред. Фактури, които се пращат наново, не са.

Записите все пак не стоят вечно. Темата има срок на пазене (retention): събитията се трият, когато остареят (в Kafka по подразбиране след 7 дни) или когато надхвърлят зададен размер. Трие ги възрастта, а не потребителите: прочетеното от всички стои до срока, а непрочетеното изчезва след него. Симулаторът не моделира срока.

По-точно, за любопитните

В истинския Kafka kafka-consumer-groups --reset-offsets иска групата да е спряна и --execute, иначе само показва какво би направил. Има и теми с компактиране (log compaction): за всеки ключ се пази поне последният запис, а по-старите записи със същия ключ се трият.

Идеята

Състоянието е сборът от събитията

В таблицата products наличността на сиренето е едно число:

SQLбаза: магазин

SELECT name, stock
FROM products
WHERE id = 7;

SELECT name, stock FROM products WHERE id = 7;

▶ ИзпълниВърни заявкатаНова база

Числото е 18, но не казва защо. Колко са дошли, колко са продадени, имало ли е връщане? UPDATE пише върху старото и историята изчезва.

Event sourcing пази събитията като истина, а състоянието се смята от тях: наличността е сборът от всички промени. Същото в SQL, с текущата наличност след всяко събитие:

SQLбаза: магазин

SELECT id, why, change,
       SUM(change) OVER (ORDER BY id) AS stock
FROM stock_events
WHERE product_id = 7
ORDER BY id;

SELECT id, why, change, SUM(change) OVER (ORDER BY id) AS stock FROM stock_events WHERE product_id = 7 ORDER BY id;

▶ ИзпълниВърни заявкатаНова база

А така изглежда в потока. Ключът е продуктът, затова всичките му събития са в един дял и подред:

поток от събития (като Kafka)симулатор

create topic stock partitions 3
produce stock key=product-7 {"product":7,"change":24,"why":"delivery"}
produce stock key=product-7 {"product":7,"change":-4,"why":"order 58"}
produce stock key=product-7 {"product":7,"change":-3,"why":"order 61"}
produce stock key=product-7 {"product":7,"change":1,"why":"returned"}
read stock partition 2

create topic stock partitions 3 produce stock key=product-7 {"product":7,"change":24,"why":"delivery"} produce stock key=product-7 {"product":7,"change":-4,"why":"order 58"} produce stock key=product-7 {"product":7,"change":-3,"why":"order 61"} produce stock key=product-7 {"product":7,"change":1,"why":"returned"} read stock partition 2

▶ ИзпълниВърни командитеНова база

Събитията дават историята наготово: кой какво е направил и кога, наличността към всеки минал ден, и ново изчисление върху старите събития. Цената е, че за текущото състояние трябва да се минат всички събития. Затова програмите пазят и готово състояние (снимка), което обновяват от потока. Събитията остават завинаги, така че и форматът им трябва да се мени внимателно.

Упражнение

Връщане, копие и корекция

1. Групите billing и analytics са прочели всичко в orders.

Сметката на анализите е поправена. Върни групата analytics в началото на orders, за да прочете всичко наново. Не пипай billing и не чети още.

поток от събития (като Kafka)симулатор

describe group analytics

describe group analytics

Провери▶ Изпълни

2. billing е прочела всичко, а после идва поръчка 65 (в дял 2).

Пусни второ копие на фактурите с име billing-2 в същата група, така че да поеме част от дяловете и да прочете новото в тях.

поток от събития (като Kafka)симулатор

describe group billing

describe group billing

Провери▶ Изпълни

3. В темата stock са събитията на сиренето (продукт 7): доставка 24, продажби 4 и 3.

Клиентът от поръчка 61 върна едно сирене. Добави събитието {"product":7,"change":1,"why":"returned"} така, че да е в историята на сиренето, след продажбите.

поток от събития (като Kafka)симулатор

read stock partition 2

read stock partition 2

Провери▶ Изпълни

Идеята

Кога поток, кога не

Поток от събития е добър, когато за едно нещо трябва да научат много системи, всяка със свое темпо, и когато историята има стойност: поръчки и плащания, движения на склада, кликове и следене на поведение, дневници на сървърите. Той не е добър, когато искаш отговор веднага („има ли сирене на склад?“, „колко е общата сума?“), търсене и промяна на запис. За това е SQL базата, която обикновено стои до потока и се пълни от него.

По-точно, за любопитните

Kafka е най-разпространеният: теми, дялове, групи, мястото на групите се пази в самия Kafka (в темата __consumer_offsets). Redpanda приема същия протокол като Kafka, затова програмите и инструментите на Kafka работят с нея. RabbitMQ е преди всичко опашка, от която съобщението изчезва след потвърждение. Неговите streams са отделен вид опашка: дневник, който се чете от произволно място и може да се чете наново. Amazon Kinesis Data Streams нарича дяловете shards и по подразбиране пази записите 24 часа.

В една страница в браузъра същата идея е шината за събития (event bus, publish/subscribe): един бутон съобщава „добавено в количката“, а броячът и сумата се абонират. Разликата е, че шината не пази нищо: който не слуша в момента, пропуска събитието. Дневникът пази събитията и позволява да се прочетат пак.

Тест

Провери се

1. Събитията на поръчка 61 („placed“, „paid“, „shipped“) трябва да се четат в този ред. Какво гарантира това?

Трите събития да имат един и същ ключ, например order-61Темата да има много дяловеПотребителят да чете с from-beginning

2. Групите billing и email четат темата orders. Колко пъти се чете поръчка 61?

Толкова пъти, колкото дяла има тематаПо веднъж във всяка група, тоест два пътиВеднъж: която група стигне първа

3. billing прочита поръчка 61, праща фактурата и пада, преди да запише докъде е стигнала. Какво става след рестарта?

Базата знае, че фактурата е пратена, и пропуска поръчкатаПоръчка 61 идва пак. Ако billing не помни кои поръчки е фактурирал, клиентът получава две фактуриПоръчка 61 се губи завинаги

4. Кое е вярно за записите в темата?

Не се променят. Трият се, когато остареят (например след 7 дни), независимо кой ги е прочелПотребителят ги трие, след като ги прочетеUPDATE поправя грешен запис на място

Провери

Започни урока отначало

Свързани уроци

• Нужно отпреди: INSERT: нови редове · UPDATE: промяна на редове · UPSERT: вмъкни или обнови · Ограничения: правила за данните
• Продължава в: Как се избира база
• Задачи по темата: Видове бази данни