У стримінгу є проблема тиші.

Коротка версія

Коли партиція Kafka замовкає — вона завершена чи просто повільна? Ззовні це виглядає однаково — а відповідь визначає, чи можна безпечно діяти на основі того, що ви вже маєте. Netflix, Uber і Walmart опублікували, як кожен із них це вирішує, і кожен чимось поступається: другим конвеєром, смугою паралелізму, коректністю під час повторного відтворення. Щоразу — добротна інженерія. Але ніхто з них не запитує в Kafka ту єдину річ, яку Kafka вже знає.

Я зіткнувся з цим на потоці даних про запаси для компанії з Fortune 50: тринадцять топіків Kafka, по три партиції в кожному, сотні мільйонів подій на день, кожна подія для певного товару мала йти в суворому порядку за часом події — читання семи днів історії з нуля й одразу перехід до живого пікового трафіку. Перший дизайн ставив тридцятихвилинне вікно перед усім, усе одно потребував другого завдання, щоб підбирати те, що вікно пропустило, і все одно випускав 5,5% записів у неправильному порядку.

Версія, що пішла у виробництво, не чекає на годинник. Вона читає офсети Kafka.

Ось як — і чому ця ідея старша за Kafka.

Компроміс, на який ідуть усі / §01

Це не проблема однієї команди. Чотири сильні інженерні організації, чотири добротні компроміси:

• Walmart закріплює кожну пару «товар–магазин» за єдиною партицією, тож її оновлює лише один консюмер. Впорядкування через закріплення ключів — жертвує паралелізмом.
• Uber запускає окремий конвеєр дозаповнення для переобчислення після запізнілих і невпорядкованих подій, бо потоковий шлях «може спричиняти неточності, відкидаючи будь-які події, що надходять після встановлення вотермарки». Він жертвує другим конвеєром.
• Netflix, дозаповнюючи конвеєри Flink, виявив, що повторно відтворена історія давала інший порядок і неконтрольоване зростання стану. Він жертвує коректністю і станом.
• Власна рекомендація Confluent для неактивної партиції Kafka: чекати на неї і зупинитися, або ігнорувати її й прийняти «багато запізнілих повідомлень, щойно партиція знову стане активною». Це жертва у вигляді зупинок або запізнілих даних.

Кожне рішення розумне. Спільна нитка в тому, що всі вони виводять повноту з часу — вікна, тайм-ауту, очікування, — тоді як офсети партицій точно кажуть, де закінчуються дані. Саме цю прогалину закриває цей підхід.

Чому потік перемішаний / §02

Топік Kafka поділений на партиції, і кожна партиція — це окрема колія. Усередині партиції порядок гарантований: запис один, потім запис одинадцять, завжди. Між партиціями ніщо нічого не координує — консюмери, які їх читають, ніколи не перебувають в одному місці в один час. Kafka робить саме те, що обіцяє, консюмери роблять саме те, що обіцяють, а об'єднаний вивід усе одно перемішаний для всього, де важлива послідовність.

Не можна діяти на «продано» раніше за «в наявності» для тієї самої одиниці. Тож перш ніж конвеєр щось випустить, він має відповісти на одне питання: чи маю я все до цього моменту? Майже кожна відповідь, що використовується сьогодні, — це таймер: вікно, тайм-аут неактивності, завдання звірки. Кожна з них — здогад щодо тиші.

Старе рішення: доповідати, а не чекати / §03

Є прецедент, і це не кращий годинник. Ранні залізниці працювали за системою часових інтервалів: наступний потяг вирушав після того, як минув фіксований інтервал, — отже, колія «мусить» бути вільною.

У червні 1889 року потяг застряг на крутому підйомі неподалік Арми в Ірландії. Задні вагони відчепили, вони покотилися вниз схилом і врізалися в потяг, що йшов позаду. Загинуло близько вісімдесяти людей.

Наступний потяг вирушив з Арми, щойно минули належні двадцять хвилин. Двадцять хвилин минуло — отже, колія мусить бути вільною. Ніхто в Армі не мав жодного способу дізнатися, що це не так.

Британія відповіла не кращим годинником. Парламент надав Торговельній раді повноваження вимагати на пасажирських залізницях абсолютної блокувальної системи: жоден потяг не входить на ділянку колії, доки сигнальний пост попереду не підтвердить, що ділянка вільна. Вільність колії перестала бути тривалістю, яку хтось висиджував, і стала фактом, про який хтось доповідає. Потяг, що застряг, і потяг, що відправився, більше не виглядали однаково, бо ніхто вже нічого не вичитував із тиші.

У цьому весь хід. Читання офсетів Kafka — це і є сигнальний пост.

Реалізація — три частини / §04

Принцип короткий: час рухається лише тоді, коли докази підтверджують, що він може. Щоб утілити це, знадобилися три частини, і найважче було не там, де очікують.

Перша: черга таймерів — це і є сортування. Кожен запис паркується в ключованому стані під своїм власним часом події, з таймером часу події саме на цю мілісекунду. Поки що нічого не випускається. Коли вотермарка досягає цієї позначки часу, таймер спрацьовує, і все, що там запарковано, виходить разом. Flink спрацьовує таймери ключа у висхідному порядку позначок часу, тож ми ніколи нічого не сортуємо. Це один метод і один елемент стану — складність ніколи не ховалася тут.

Друга: потік без часу події стає годинником. Один із тринадцяти топіків узагалі не мав часу події — він лише публікував дійсні товари, якими збагачувалося все інше. Тож йому не було чого сортувати, і він міг робити те, чого не могли інші: випускати вотермарки з часу обробки, завжди найпізнішу можливу позначку часу. Заковика: годинник конвеєра — це мінімум серед його входів, тож потік, що біжить у «зараз», нічого не змінює, доки історичні потоки ще рахуються. Я побудував цю частину першою, і вона не дала жодного ефекту — вона не була помилковою, вона була не в тому порядку. Керівний потік має значення лише тоді, коли інші залишають мінімум, а потік залишає мінімум, коли його позначено неактивним.

Третя: Kafka вирішує, коли потік завершено, — не тайм-аут. Вбудований механізм неактивності Flink — це таймер: якийсь час немає даних — позначити неактивним. Але тайм-аут не відрізнить «запускається» від «завершено», а запуск — найгірший випадок: з'єднання формуються з різною швидкістю, один топік сидить тихо, тайм-аут спрацьовує, інші просувають годинник повз часи подій, яких він ще не прочитав, і його перший запис прибуває вже запізнілим. Це не затримка, це порушення. Тож тайм-аут мав піти. Під час запуску, перш ніж читати бодай один запис, завдання запитує в Kafka останній офсет кожної партиції й утримує його. Топік неактивний лише тоді, коли кожна партиція досягла цього числа — не більшість, кожна. Жодної тривалості для налаштування, бо шлагбаум — це факт від Kafka.

// once, at job startup
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

// as records arrive
if (offset + 1 >= startupEndOffsets.get(partition)) markCaughtUp(partition);

Що це дало / §05

Вікно тримало кожен запис, доки не закривалося, — тридцять хвилин лише щоб зчитати історію, і знову після кожного перезапуску. Тепер запис випускається, щойно вотермарка його проходить: у середньому менш ніж 100 мілісекунд, включно із записами, які притримували, щоб вони не вийшли не по порядку. Запис, що прибув на першій хвилині, виходить на першій хвилині. З упорядкуванням та сама історія: два шари неактивності за тайм-аутом випускали 5,5% записів не по порядку; один шар — 1,5%; неактивність, яку визначають офсети Kafka на момент запуску, — нуль. Вікно зникло, і завдання підбирання теж — одне завдання читає сім днів історії, а потім працює наживо.

Чого це не обіцяє / §06

Детермінований — не означає ясновидющий. Подія існує для конвеєра лише з моменту, коли вона потрапляє в Kafka. Якщо продюсер притримує запис і той потрапляє туди після того, як годинник уже пройшов його час події, він виходить одразу, поза чергою, і лічильник порушень це фіксує. Жоден потоковий процесор не може впорядкувати подію, якої ще не бачить. Усередині цієї межі гарантія точна, а відставання вотермарки визначає, наскільки широка ця межа.

Те саме питання — усюди / §07

Побачивши це раз, ви бачитимете це в кожній системі, яка визначає істину з потоків, що не надходять одночасно.

Річковий водомірний пост замовкає під час повені. Річка стабільна — чи вода дісталася датчика? У 2018 році USGS повідомила про водомірні станції, виведені з ладу ураганом «Флоренс», — саме ті, що були найважливішими. «Немає новин» — хибне прочитання цієї тиші.

Камерна установка. Знімайте атлета кількома камерами — і кожна є власним потоком, наче партиція; ви не можете злити 3D-позу для миті, доки кожна камера її не передала. Мовчазна камера могла втратити кадр — або спостерігати за людиною, що завмерла. На i3d.ai це те саме питання, десятки разів на секунду.

Агент. ШІ-агент викликає інструмент і не чує нічого у відповідь. Агенти чудово вміють трактувати тишу як успіх — помилки не прийшло, отже, далі до наступного кроку з машинною швидкістю. Агент, що діє на обліковому записі клієнта в тому стані, в якому той був годину тому, — це потяг, відправлений за таймером. Рішення — те саме, що знайшли залізниці: жоден крок не відбувається на тиші; кожен крок відбувається на підтвердженні, на яке можна вказати.

Найдешевший виграш у надійності в стримінгу — не швидший рушій і не більше вікно. Це перестати виводити «завершено» з тиші й запитати систему, яка вже знає. Вікно висиджує найгірший випадок для кожного запису; перевірка офсетів чекає рівно стільки, скільки вимагають докази, і ні мілісекундою більше.

Спрощена еталонна реалізація — генератор, обробник, трекер наздоганяння — це наступне, що я публікую.