Как технический менеджер в Яндексе я отвечаю за эксплуатацию больших кластеров, через которые мы передаём данные, — для этого мы используем YDB Topics, собственный аналог Apache Kafka, о котором я уже рассказывал.

Но если посмотреть не только на нас, но и на примеры других крупных систем, то становится понятно, почему энтерпрайзу сложно брать готовые опенсорс‑решения. Стоит лишь представить, что бывает, если команда с большими масштабами начинает использовать Apache Kafka.

В этой статье я покажу собирательный образ такой крупной системы из нашего опыта общения с большими распределёнными командами, и мы увидим разницу в TCO Apache Kafka и YDB Topics на конкретных цифрах.

Что требуется от шин передачи данных

У шин данных есть базовые свойства:

• Они должны передавать данные — быстро, много и отказоустойчиво.

• Есть требования к скорости. В норме шины должны передавать данные за десятки миллисекунд.

• На что очень часто не обращают внимания: эти системы должны стоить относительно дёшево. Чтобы эксплуатировать ваш кластер и передавать ваш объём данных, не должен требоваться отдел в 100 человек или целый дата‑центр. Нужно, чтобы систему обслуживало как можно меньше людей, чтобы она стоила как можно дешевле и занимала минимальное количество оборудования.

В 2013 году в Яндексе начали использовать Apache Kafka: на этой платформе работало несколько сотен серверов. Мы передавали через неё биллинговые данные, журналы и данные приложений — все виды данных Яндекса. В какой‑то момент в системе произошёл неприятный сбой, и мы не на шутку испугались, что ещё один сбой — и нас всех уволят. Мы поняли, что надо с этим что‑то делать, и начали строить свою систему, в которой могли быть уверены.

В 2017 году мы запустили систему YDB Topics в продакшене Яндекса, и перевели на неё все потоки данных. В 2022 году мы выложили её в опенсорс в составе YDB‑платформы (дальше вы знаете).

Сейчас, спустя 7 лет, в пике мы передаём данные вплоть до скорости 300 ГБ в секунду (а если это, например, «чёрная пятница», то даже больше). Этим опытом мы с удовольствием делимся с другими компаниями, и нередко показываем разницу для конкретной команды «на пальцах». Чтобы понять, как ведёт себя Apache Kafka и как на её месте себя чувствует YDB Topics, стоит заглянуть вглубь этих систем.

Заглянем в крупную систему: начнём с оборудования

Для примера представим себе действительно крупную инсталляцию, где для передачи данных используется 1500 серверов. Скорее всего, у такой большой системы будет учтён Disaster Recovery и кластеры будут распределены на несколько зон доступности.

Параметры кластеров у всех компаний разные, так что мы возьмём за точку отсчёта принятые в Яндексе 18 часов хранения (мы не разрешаем пользователям хранить данные меньше 18 часов). Почему мы так сделали? Когда мы решали, что важнее: надёжность или стоимость системы, — мы выбрали вариант с большей надёжностью. За 18 часов на больших объёмах пользователи успевают создать примерно 10 ПБ данных, которые в каждый момент времени находятся внутри шины. Это большие значения, но с учётом всех наших моделей отказа мы знаем, что самый суровый даунтайм не будет превышать эти 18 часов. Так что мы готовы пожертвовать местом на кластерах в пользу надёжности.

Диски

Крупные инсталляции нередко используют в серверах обычные HDD‑диски. Но с ними есть проблема. Одно из требований к шине данных — задержка. В норме шина должна передавать данные за десятки миллисекунд. А HDD‑диски очень не любят, когда на них идёт нагрузка, большое количество операций. В итоге придётся иметь много жёстких дисков, чтобы обеспечивать 10-миллисекундные задержки. Посчитаем, что будет, если заменить все HDD‑диски на NVMe.

Мы посчитали экономию в оборудовании при переходе с HDD на NVMe‑диски на основе числа операций ввода‑вывода (IOPS). У нас получилось, что при переходе на NVMe нам понадобится в 2–3 раза меньше серверов. На масштабах из нашего примера окажется: если на NVMe‑дисках нам достаточно 1500 серверов, то серверов с HDD нужно уже 4500.

Но тут другая проблема. HDD‑диски большие, места на них много, а NVMe — маленькие. Можно упереться в объём хранения, поэтому в этом месте стоит посмотреть, а как вообще можно хранить данные.

Стандартные опенсорс‑решения, та же Apache Kafka, в основном реплицируют свои данные, делают несколько копий. Они берут пользовательские данные и делают ещё две копии в разных местах. Для нашего примера это означало бы, что 10 ПБ пользовательских данных превратились бы в 30. То есть 1500 серверов умножаем на три и снова получаем 4500 серверов. Но есть способ лучше, который называется Erasure‑кодирование.

Напомню суть этого математического фокуса: вы берёте четыре блока данных, добавляете к ним два блока контрольных сумм, в итоге эти шесть блоков могут выдержать выпадение любых двух из них. Та же Apache Kafka выдержит выпадение одного из трёх, а YDB Topics выдерживает выпадение двух из шести. То есть, чтобы всё работало, требуется коэффициент всего 1,5 от числа данных.

Итого: за использование Apache Kafka на этом объёме «заплатим» дополнительными 2250 серверами (4500-1500*1,5).

У NVMe‑дисков есть ещё один параметр, на который часто не обращают внимания. Он называется DWPD, или Drive Writes Per Day. Параметр показывает, сколько раз вы можете полностью перезаписать диск данными до того, как он выйдет из строя. Если вы пишете много данных, часто их перезаписываете, время наработки на отказ сильно сокращается. Вы начинаете часто менять диски, для этого покупать новые — бизнес будет недоволен. Поэтому, кроме всего прочего, экономия в объёмах записываемых данных ещё и увеличивает сроки жизни ваших дисков, а значит уменьшает стоимость эксплуатации всей системы.

Добавление серверов

Железо приходит в дата‑центры не постоянно, а порциями: заказ, дозаказ и так далее. Ниже — реальный график, который показывает, как добавляются серверы в Яндексе.

Когда это происходит, мы хотим, чтобы вся нагрузка моментально начинала разъезжаться на появившиеся мощности, чтобы была полная автоматика: серверы приехали, добавились в кластер, всё работает.

Но кроме добавления серверов, существует процесс уменьшения кластеров. Он возникает при миграции между дата‑центрами. Например, вы жили в одном дата‑центре, он уменьшается или выключается, и вам нужно всё перевести в соседний. В ответ на это один дата‑центр растёт, второй — уменьшается.

Этот постоянный процесс «меньше‑больше» Apache Kafka очень не любит: у платформы до сих пор нет возможности удобно уменьшать размеры кластеров. Для крупных инсталляций это критично, потому что держать одновременно несколько больших кластеров просто не хватает оборудования.

Как добиться отказоустойчивости

В шине данных передаётся несколько типов трафика, в зависимости от строгости SLA.

1. Самые критичные данные, например, биллинговые. Это то, что ни в коем случае нельзя потерять и нужно доставить в том порядке, в каком пришло. Шина должна всегда принимать и отдавать данные. Здесь требуются максимальные гарантии из возможных.

2. Данные, требующие долгосрочного хранения, например, журналы работы приложений — логи. Многие крупные компании находятся под большим количеством аудитов: ISO, PCI DSS, ГОСТ, политики комплаенса.
   С точки зрения логов нет особой необходимости доставлять их за миллисекунды. Более того, при передаче в них могут появляться дубли. В этом тоже нет проблемы, потому что принимающая система — база данных — сама их дедуплицирует, выровняет. Здесь гарантии могут быть послабее.

3. Пользовательские данные. Разработчики отправляют данные в шину, а с другой стороны они или другие разработчики их забирают, строят свои процессы поверх этого и сами управляют нужными гарантиями (exactly once или at least once, гарантии порядка, надёжности).

Последний пункт чуть раскроем в следующем разделе.

Виды инсталляций

Чтобы работать с трафиком, YDB Topics поддерживает два варианта инсталляций.

Первая называется «Кросс‑ДЦ». Эта инсталляция для самых критичных видов трафика, то есть для тех же биллинговых данных. Для пользователя она выглядит просто как один большой кластер. При этом эта инсталляция полностью переживает выпадение дата‑центра. Пользователь даже не замечает, если что‑то происходит: для него система продолжает работать всегда.

Второй тип инсталляции называется «Федерация». В этом режиме пользователи отлично знают, что у нас несколько дата‑центров, знают, как система себя ведёт при выпадении ДЦ, и учитывают эти модели в своих приложениях.

Кросс‑ДЦ. Этот кластер, который развёрнут поверх трёх реальных дата‑центров, для пользователей выглядит как просто одна большая инсталляция, куда можно писать данные, не задумываясь, произойдёт ли что‑то. Такая система выдерживает выпадение одного дата‑центра плюс выпадение стойки в соседнем.

Это во многом стандартная для нас модель отказа: любой сервис должен жить при минус одном ДЦ и минус одной стойке в соседнем (никто не может застраховать от экскаватора, перерубающего оптику, и от случайного отказа оборудования в другом ДЦ одновременно). YDB Topics устойчив к этому за счёт того, что мы построены поверх YDB‑платформы, которая умеет такие случаи обрабатывать.

Федерация. В этом случае пользователи отлично знают, что у них есть кластеры, что дата‑центров несколько. Если ДЦ выключается, пользователи тоже об этом знают.

Чтобы система работала при выключении ДЦ, есть отдельная сущность, которая командует: «Так, у нас выпал первый дата‑центр; пожалуйста, половину трафика переведите сюда, а другую половину — в третий дата‑центр». Потоки данных переключаются автоматически на другие ДЦ, пользователи получают данные и счастливы.

Почему пользователи выбирают этот вариант? Если мы используем Кросс‑ДЦ‑инсталляцию, нужно три полных реплики, чтобы полностью выдерживать любые варианты отказов. При использовании федерации мы работаем в модели Erasure‑кодирования, что существенно дешевле с точки зрения объёма дисков. Разница между Кросс‑ДЦ‑кластером и федеративным на внутреннем биллинге существенна, и люди деньгами голосуют за такие инсталляции. По нашим наблюдениям, больше 90% данных передаётся именно в режиме федерации, а не в режиме Кросс‑ДЦ.

Резервирование

Что происходит, если выключается один дата‑центр? Apache Kafka говорит: «Если вам нужно георезервирование, возьмите два дата‑центра, поставьте их рядышком и настройте между ними репликацию данных с помощью репликатора или Mirror Maker». Но Apache Kafka не говорит, сколько мощности нужно держать в этом случае про запас.

Например, у моей компании из примера есть два одинаковых дата‑центра. Если один дата‑центр выключается, все данные, которые писатели продолжают создавать, начинают уходить во второй.

То есть нужно в каждом дата‑центре иметь стопроцентный запас мощности, чтобы система работала. На объёмах с 500 серверами в каждом дата‑центре нужно иметь 500 серверов про запас на тот случай, если какой‑то дата‑центр выключится.

Если же мы можем работать с большим количеством дата‑центров, то цифра становится лучше. Потому что когда «умирает» один дата‑центр, потоки из него можно поделить по двум оставшимся.

В результате в резерве уже не 500 серверов, а всего 250 в каждом дата‑центре. Чем больше дата‑центров, тем меньше степень резервирования железа. Поэтому, чем на большее число площадок можно распределить данные, тем выгоднее такое размещение с точки зрения объёма резервирующего оборудования.

Но дата‑центры зачастую не одинаковые, сделать полностью одинаковые кластеры в разных дата‑центрах не всегда возможно, (об этом Apache Kafka тоже не говорит).

Часто бывает, что вам нужно получить железо, а его нет в том дата‑центре, где вы хотите его получить. Оно или занято, или недоступно по причине обновления. Поэтому в реальности потоки будут распределяться не 50% в один ДЦ, 50% в другой, а по более сложным формулам. Например, 83% в один, 17% в другой.

Возвращаясь к модели федерации, скажу, что именно поэтому в основном многие пользователи предпочитают эти кластеры. Они дешевле за счёт того, что гораздо меньше железа тратится на такое резервирование. В случае Кросс‑ДЦ запас железа очень большой, с федерацией он гораздо ниже. И что ещё тут важно — всё распределение трафика выполняется автоматически силами SDK, пользователи про это даже не знают.

Разделение слоёв Compute и Storage

YDB‑платформа явно делит слои хранения и слои Compute. Есть отдельные процессы, которые отвечают за хранение данных, а рядом можно поднять произвольное количество динамических нод, которые занимаются вычислениями. Они будут получать данные из системы хранения, что‑то с ними делать, и эти данные могут рассылаться в произвольное количество динамических нод для обработки.

Почему это важно? Базовый примитив системы — Tablet, или «таблетка», как часто говорят на русском. Проще всего его воспринимать как маленький кусочек таблички, в котором хранится набор данных.

Но этот компонент на самом деле хитрый. Он использует подход Replicated State Machine, или RSM. Основная идея такая: у вас есть набор данных, он лежит в нашей системе хранения. У вас есть входящая очередь команд в этот компонент. Компонент специально написан однопоточным, он забирает команду, забирает своё состояние из системы хранения, что‑то выполняет и потом транзакционно говорит: «Я команду обработал, новое состояние сохранил». Это означает, что мы в любое время этот компонент можем «убить», остановить сервер, сделать всё что угодно. Он будет поднят в соседнем месте, точно так же поднимет своё состояние и свою очередь команд и продолжит вычисления. Таким образом очень удобно переживать любые виды отказов.

Это полезно при балансировке нагрузки. В платформе YDB есть специальный компонент Hive, который следит, чтобы уровень Compute был равномерно загружен. Когда появляются новые серверы, этот компонент обнаруживает свободные мощности, берёт часть таблеток и переносит их на эти серверы. Пользователи даже не замечают, что что‑то происходит, процесс очень быстрый.

Реакция на сбой тоже занимает десятки миллисекунд. Как только нода перестала отвечать, в течение 20 миллисекунд таблетка будет поднята на другом сервере, возьмёт нагрузку и продолжит вычисление. Пользователи часто этого не замечают, потому что не ставят алерты на такие маленькие значения. Обычно алерты настроены на десятки секунд, поэтому задержки уровня 20 миллисекунд не замечаются. Всё происходит прозрачно, и люди не замечают этих ситуаций.

Обновление кластера

Есть интересный процесс обновления кластера, когда необходимо выкатить новую версию на сотни серверов в дата‑центре. Проходит он аналогично. Берём ноду, выключаем её, система балансировки видит, что есть таблетки, которые надо перенести, переносит их на свободные, обновляем версию, появляется пустой сервер, обратно возвращаем все таблетки.

В реальности, конечно, бывают неудачные релизы, и в этом случае возникает откат. Он повторяется ровно так же в обратную сторону. А ещё бывает очень интересно, когда релиз хороший, но старый сервер вернулся из починки, на нём «проснулась» старая версия и начала отправлять данные. И весь кластер начал себя вести странно, потому что не ожидал такого. В любом случае версии, которые мы строим, совместимы между собой вперёд и назад в пределах одной мажорной версии, чтобы можно было быстро их раскатить и вернуть обратно, даже иногда повторять это много раз.

Как управлять кластером

Коммунальный или выделенный кластер?

Итак, мы создали кластер, разобрались с балансировкой, и теперь возникает вопрос: что видят пользователи, когда начинают в него заходить?

Вот как это может выглядеть для коммунального кластера с учётом квот:

Если нужно было бы выдать всё, что люди заказали в виде мощности на этом примере, понадобился бы объём оборудования для передачи 360 ГБ в секунду. При этом видно, что по факту люди используют не больше 100 ГБ в секунду. За счёт того, что система коммунальная, можно сэкономить 70% оборудования. На примере крупной системы в самом начале — без коммунального кластера вместо 1500 серверов было бы 6000. Цифра внушительная.

Правда, выделенные кластеры могут понадобиться для отказоустойчивости, потому что бывает неудачный релиз, может быть сбой, а мы хотим, чтобы кластеры были к этому устойчивы. Такие критичные инфраструктурные кластеры в нашей практике существуют, но по факту их очень мало. Они нужны в основном для надёжности особо критичных систем. Надёжности никогда не бывает мало, можно, конечно, все кластеры сделать выделенными, но это потребует огромного количества железа.

В свою очередь, Apache Kafka рекомендует создавать маленькие кластеры, но команда из примера потеряла бы на этом уже 3000 серверов.

Ресурсная модель

С одной стороны, YDB Topics очень похожи на Apache Kafka: есть точно такие же понятия «топик», «партиции». При этом есть понятие, которого нет в Apache Kafka, — аккаунт. Это количество мощности, которое выделено команде.

Команда заказывает железо, получает его, и оно появляется внутри аккаунта. О том, что железо есть, пользователи узнают из информационных систем заказа оборудования. Сам аккаунт создается автоматически, а пользователи «живут» внутри этих аккаунтов.

Квотирование ресурсов

Apache Kafka говорит, что квоты обеспечиваются внутри каждого конкретного брокера. По факту системы распределённые, и нельзя заставить пользователя прийти в один конкретный брокер. Поэтому в YDB Topics используются распределённые квоты. Пользователь может прийти на любой из сотен серверов, начать писать данные, переместиться на другой сервер, при этом он всегда будет ограничен действующими квотами на аккаунт или топик.

Есть ещё одна интересная квота — на холодный старт кластера. Наши разработчики пишут очень эффективный код, при холодном запуске кластера мы стартуем и начинаем изо всех сил нагонять отставание, забирая данные из соседних дата‑центров с огромной скоростью. В результате можно уйти далеко за сотни терабит данных, перегружая все каналы связи, что будет тормозить работу остальных систем. Тут и пригодится квота, которая помогает ограничить аппетиты всего кластера, снизив нагрузку на каналы передачи и позволив и другим системам тоже стартовать (и это тоже распределённые квоты, а не локальные квоты на уровне одного брокера).

Отдельный вид квот используется при выпадении одного дата‑центра.

Например, у нас есть два дата‑центра, где мы выдали пользователям по 100 МБ в секунду в обоих. Если первый дата‑центр выключается, весь поток данных идёт во второй. Но там квота 100 МБ в секунду, то есть технически принять поток не позволяет железо. В этот момент можно расширить квоту и разрешить пользователю временно написать больше данных. А когда дата‑центр возвращается, сжимаем эту квоту и показываем пользователю, что уже всё хорошо.

Пару слов про аудит

Если у компании много аудита, комплаенса, то логи собираются во всевозможных сочетаниях: по консьюмерам, партициям, читателям, IP‑адресам. В опенсорсе Apache Kafka этого нет — появляется только в коммерческой версии (в опенсорсе аудит же не нужен!)

Self-service

Как я уже писал, а 2013 году у нас была Apache Kafka и команда из 12 сисадминов, которые с утра до ночи обрабатывали входящие запросы: создайте топик, настройте мощность, заведите ещё один топик или кластер. Но если увеличить масштаб системы до нескольких тысяч серверов, то будет нелишним отдать работу с кластерами в самих пользователей, иначе для обработки запросов потребуются уже сотни человек. Чтобы этого не допустить, необходимые операции можно перевести в режим self‑service: например, создание топиков, правила чтения и так далее. Так пользователи могут что‑то перенастроить, даже когда параллельно катится релиз самого YDB Topics.

Для того чтобы люди могли настроить алерты, в YDB Topics мы даём сотни метрик.

Алерты бывают всевозможные: с точностью до партиции, читателя, трафика между дата‑центрами. Всё это сохраняется на долгое время, чтобы пользователи могли своими силами разбираться с проблемами.

Мы даём пользователям готовые шаблоны, чтобы они понимали, с чего начинаются проблемы: например, с 90% нехватки квоты или с 80%. Так им становится попроще.

Подведу итог. Опенсорс зачастую не совсем подходит для больших компаний.

Если бы крупная компания из начального примера выбрала чистый опенсорс, на поддержке было бы 120 человек вместо одного, а вместо 1500 серверов — 8000.

А серверы нужно не только обслуживать — им требуется место в дата‑центре, чего часто не хватает.

Что дальше?

YDB Topics — это основная шина в Яндексе, через которую мы передаём большой объём данных. Когда мы вышли в опенсорс, поняли, что несмотря на то, что наши протоколы классные и эффективные, перестраивать всё ПО, переписывать наши протоколы нереально, поэтому мы занялись поддержкой протокола Apache Kafka. Сейчас мы полностью совместимы c Kafka на чтение и на запись.

Мы часто сталкиваемся с тем, что комьюнити нас не поддерживает, когда мы в опенсорсе хотим сделать оптимизацию с перспективой экономии, например, 1% ядер, — многие считают, что это не нужно: столько кода, так сложно. Но на больших масштабах 1% экономии — это десятки серверов, очень большие цифры. Сейчас мы уже догнали Apache Kafka, сопоставимы с ней и видим, куда улучшаться: по процессору, сети, объёму хранения.

Мы открыли весь исходный код YDB Topics, он выложен в составе платформы YDB на Github. Мы по‑прежнему уверены, что задачи, которые решают компании‑гиперскейлеры во многом уникальны. Поэтому искренне считаем, что можно создать комьюнити, которое сумеет совместно развивать технологии, позволяющие быстрее, проще и выгоднее решать задачи любого масштаба. Так что присоединяйтесь!