Зачем Kafka: журнал вместо отчёта

Сколько партиций: посчитать, а не угадать

20 мин

Чему научишься

считать число партиций: две нижние границы и одна верхняя, и брать ту нижнюю, которая выше
записывать чужую цифру допущением: что принято, откуда взято и что будет, если она вдвое меньше
называть цену лишней партиции числами: память продюсеров, файлы на брокере, секунды простоя консьюмер-группы
переезжать на топик с меньшим числом партиций и знать цену обоих способов переезда
ставить одну партицию осознанно: полный порядок ценой одного брокера и одного консьюмера

21:30. Шестьдесят четыре — с запасом

2 ноября 2184 года, диспетчерская «Хранилища-9». Эфир идёт, Читальный зал подключён, справочник антенн собирается. Утром ты поднял signal_raw на трёх партициях — наспех, лишь бы поймать несущую. Теперь расширяешь его до 64. С запасом: уменьшить потом нельзя, так пусть будет с избытком.

В 21:50 начинают отваливаться продюсеры эфира. Шестерым на 64 партициях нужно 384 мегабайта памяти, а столько на уровне связи нет. И тут же жалуется Читальный зал: к его группе подключился седьмой консьюмер, и остальные встали почти на одиннадцать секунд — пока группа заново делит партиции между консьюмерами, чтение стоит. Утром такая пауза длилась полсекунды: делить надо было три партиции, а теперь 64.

Запас, взятый «чтобы потом не переделывать», сломал обе стороны разом: и тех, кто пишет, и тех, кто читает.

КВЕРИ: Запас — тоже число. Его тоже считают.

Толстый поток янтарных капсул расходится веером из десятков тонких дорожек; капсулы бегут лишь по части из них. У основания веера перегруженная машина-продюсер: прибор в красном, вокруг громоздятся блоки памяти. Справа замерли посты консьюмеров со знаком паузы. Дежурный со спины стоит между веером и постами, рядом КВЕРИ, кот-интерфейс архива. — Шестьдесят четыре дорожки с запасом: кадры бегут по двенадцати, остальные пусты, а память продюсер держит под все.

Сначала — во что обошлись 64 партиции, числами. Ячейка печатает три вещи. Первая: цену партиций по учебной модели; сама модель напечатана сразу под заголовком первой части — сверь с ней таблицу. Вторая: signal_raw после расширения — сколько из 64 партиций вообще получили кадры — и попытку вернуть двенадцать. Третья: единственный способ получить двенадцать партиций под тем же именем. Смотри, сколько кадров в топике после него.

from arena_kafka import cluster, KafkaError, KF_BASE_TS_MS
import arena_kafka_ops  # смотровые окна: cluster.describe()
# KF_BASE_TS_MS — полночь 2 ноября 2184 года, начало смены
# Подключены заранее: fmt_seconds(ms) — секунды с десятыми, partitions_with_frames(desc) —
# сколько партиций топика получили хоть один кадр.

MIN = 60_000
HOUR = 60 * MIN

# Учебная модель цены партиции. Числа круглые, чтобы арифметика читалась.
PRODUCERS = 6           # продюсеров эфира на уровне связи
MB_PER_PARTITION = 1    # продюсер держит по мегабайту на КАЖДУЮ партицию топика
MS_PER_PARTITION = 170  # ребаланс группы: 170 мс на каждую партицию


print('== 1. Цена партиций по учебной модели')
print('продюсеров — %d, каждый держит по %d МБ на каждую партицию топика: какие ключи ему достанутся, заранее не знает' % (
    PRODUCERS, MB_PER_PARTITION))
print('ребаланс группы — пауза %d мс на каждую партицию' % MS_PER_PARTITION)
print('утром партиций было 3 — пауза %s с' % fmt_seconds(3 * MS_PER_PARTITION))
print('партиций  МБ памяти продюсеров        ребаланс, с')
for n in (12, 24, 64, 200):
    memory = PRODUCERS * n * MB_PER_PARTITION
    print('%8d  %20s  %17s' % (n, '{:,}'.format(memory).replace(',', ' '), fmt_seconds(n * MS_PER_PARTITION)))
print()

print('== 2. 21:30 — signal_raw расширен с 3 партиций до 64')
cluster.create_topic('signal_raw', partitions=3, retention_ms=6 * HOUR)
cluster.alter_partitions('signal_raw', partitions=64)
# Эфир с 21:30 до 21:50 — пока продюсеры держались. Ключ кадра — source_id.
cluster.fill('signal_raw', 48, KF_BASE_TS_MS + 21 * HOUR + 30 * MIN,
             KF_BASE_TS_MS + 21 * HOUR + 50 * MIN)
desc = cluster.describe('signal_raw')
print('партиций: %d, кадров за 21:30–21:50: %d' % (desc['partitions'], sum(desc['end'].values())))
print('партиций, куда попал хоть один кадр: %d из %d' % (partitions_with_frames(desc), desc['partitions']))
try:
    cluster.alter_partitions('signal_raw', partitions=12)
except KafkaError as e:
    print('alter_partitions(12) -> KafkaError:', e)
print()

print('== 3. Переезд с сохранением имени: погасить и завести заново')
cluster.delete_topic('signal_raw')
cluster.create_topic('signal_raw', partitions=12, retention_ms=6 * HOUR)
desc = cluster.describe('signal_raw')
print('партиций: %d, кадров в топике: %d' % (desc['partitions'], sum(desc['end'].values())))
start = cluster.now_ms()
cluster.fill('signal_raw', 48, start, start + 10 * MIN)
desc = cluster.describe('signal_raw')
print('через 10 минут эфира: кадров %d, партиций с кадрами: %d из %d' % (
    sum(desc['end'].values()), partitions_with_frames(desc), desc['partitions']))

from arena_kafka import cluster, KafkaError, KF_BASE_TS_MS
import arena_kafka_ops  # смотровые окна: cluster.describe()
# KF_BASE_TS_MS — полночь 2 ноября 2184 года, начало смены
# Подключены заранее: fmt_seconds(ms) — секунды с десятыми, partitions_with_frames(desc) —
# сколько партиций топика получили хоть один кадр.

MIN = 60_000
HOUR = 60 * MIN

# Учебная модель цены партиции. Числа круглые, чтобы арифметика читалась.
PRODUCERS = 6           # продюсеров эфира на уровне связи
MB_PER_PARTITION = 1    # продюсер держит по мегабайту на КАЖДУЮ партицию топика
MS_PER_PARTITION = 170  # ребаланс группы: 170 мс на каждую партицию

print('== 1. Цена партиций по учебной модели')
print('продюсеров — %d, каждый держит по %d МБ на каждую партицию топика: какие ключи ему достанутся, заранее не знает' % (
    PRODUCERS, MB_PER_PARTITION))
print('ребаланс группы — пауза %d мс на каждую партицию' % MS_PER_PARTITION)
print('утром партиций было 3 — пауза %s с' % fmt_seconds(3 * MS_PER_PARTITION))
print('партиций  МБ памяти продюсеров        ребаланс, с')
for n in (12, 24, 64, 200):
    memory = PRODUCERS * n * MB_PER_PARTITION
    print('%8d  %20s  %17s' % (n, '{:,}'.format(memory).replace(',', ' '), fmt_seconds(n * MS_PER_PARTITION)))
print()

print('== 2. 21:30 — signal_raw расширен с 3 партиций до 64')
cluster.create_topic('signal_raw', partitions=3, retention_ms=6 * HOUR)
cluster.alter_partitions('signal_raw', partitions=64)
# Эфир с 21:30 до 21:50 — пока продюсеры держались. Ключ кадра — source_id.
cluster.fill('signal_raw', 48, KF_BASE_TS_MS + 21 * HOUR + 30 * MIN,
             KF_BASE_TS_MS + 21 * HOUR + 50 * MIN)
desc = cluster.describe('signal_raw')
print('партиций: %d, кадров за 21:30–21:50: %d' % (desc['partitions'], sum(desc['end'].values())))
print('партиций, куда попал хоть один кадр: %d из %d' % (partitions_with_frames(desc), desc['partitions']))
try:
    cluster.alter_partitions('signal_raw', partitions=12)
except KafkaError as e:
    print('alter_partitions(12) -> KafkaError:', e)
print()

print('== 3. Переезд с сохранением имени: погасить и завести заново')
cluster.delete_topic('signal_raw')
cluster.create_topic('signal_raw', partitions=12, retention_ms=6 * HOUR)
desc = cluster.describe('signal_raw')
print('партиций: %d, кадров в топике: %d' % (desc['partitions'], sum(desc['end'].values())))
start = cluster.now_ms()
cluster.fill('signal_raw', 48, start, start + 10 * MIN)
desc = cluster.describe('signal_raw')
print('через 10 минут эфира: кадров %d, партиций с кадрами: %d из %d' % (
    sum(desc['end'].values()), partitions_with_frames(desc), desc['partitions']))

python · kafka

Откуда берётся число

На экране те самые 384 мегабайта и 10,9 секунды из сцены. Лишняя партиция обходится дорого сразу с двух сторон. Продюсер платит памятью за каждую партицию топика. А группа стоит всё время, пока заново решает, кто из консьюмеров какие партиции берёт, — и чем больше партиций делить, тем дольше. Это ребаланс (rebalance) — перераздача партиций между консьюмерами группы. Значит, у вопроса «сколько партиций» нет безопасного ответа «побольше». Число надо посчитать.

Источник чисел пришёл в 22:40 — заявка Читального зала на полный захват эфира с полуночи: пик 216 МБ/с (480 кадров в секунду по 450 КБ), девять консьюмеров на пике и строчка «одна партиция принимает 18 МБ/с». Партиция в группе достаётся ровно одному консьюмеру, поэтому консьюмер сверх числа партиций стоит без работы. Из этого и из заявки получается число. Счётных границ три:

нижняя граница по потоку — нужный поток, делённый на то, что тянет одна партиция: 216 ÷ 18 = 12. На меньшем числе партиций пик не пройдёт;
нижняя граница по числу консьюмеров — сколько их работает одновременно на пике: 9. Партиций меньше — лишние консьюмеры стоят без работы;
верхняя граница по ключу — сколько у ключа разных значений. Партицию выбирает source_id, а значений у него двенадцать, по числу антенн в справочнике. Больше двенадцати партиций кадры не займут ни при каком потоке: во второй части ячейки из 64 партиций кадры получили 12, остальные 52 пусты навсегда.

Берём ту нижнюю, которая выше, — 12 — и сверяем с верхней: тоже 12. Места для запаса между границами не осталось: на меньшем числе не пройдёт пик, больше двенадцати партиций ключ не займёт. Одна оговорка: двенадцать — потолок, а не гарантия ровной раскладки. В третьей части ячейки кадры легли в десять партиций из двенадцати, а деление 216 на 18 молча считает, что поток лёг поровну; почему по ключу он ложится неровно — разговор главы 2.

Цифра, которую ты не мерил

18 МБ/с на партицию ты не измерял — эту цифру ЗАЯВИЛ Читальный зал, вместе с пиком и числом консьюмеров. Мерить её в учебной Kafka нечем, и в настоящем разговоре такую цифру тоже чаще дают, чем меряют. Правило простое: такое число записывают допущением — что принято, откуда взято и что будет, если оно окажется вдвое меньше. Если партиция тянет 9 МБ/с, нижняя граница по потоку — 24, она перерастает верхнюю, и ответа в двенадцати партициях больше нет. В первом уроке главы ты записывал числом требование; здесь тем же способом записываешь допущение.

Три границы на одной шкале: две снизу — по консьюмерам и по потоку, одна сверху — по ключу; серое — где число не годится, свободной остаётся одна высота. Вечерние 64 стоят в серой зоне: 384 МБ памяти продюсеров и пауза 10,9 с против 72 МБ и 2 с у двенадцати.

Практика: напиши код

Напиши функцию, которая считает число партиций по такой заявке, как у Читального зала: plan_partitions(peak_mb_s, per_partition_mb_s, peak_consumers, key_values) — сколько партиций ставить, или None, если подходящего числа нет. В заявке три числа, и все три ЗАЯВЛЕНЫ, а не замерены тобой: пик — 216 МБ/с, одна партиция принимает 18 МБ/с, консьюмеров на пике — 9. Четвёртое число твоё, из справочника: у ключа source_id двенадцать разных значений.

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

None — не ошибка функции, а честный ответ: при таком ключе подходящего числа партиций нет. Запас на рост в функцию не закладывай: она считает границы, а запас — отдельное решение поверх них.

def plan_partitions(peak_mb_s, per_partition_mb_s, peak_consumers, key_values):
    '''Число партиций по заявке — или None, если ключ столько не разведёт.'''
    # Нижняя граница по потоку: пик делим на то, что тянет одна партиция.
    return -(-peak_mb_s // per_partition_mb_s)

python · kafka

Цена лишней партиции

Каждая партиция сверх верхней границы ничего не даёт, но за неё платят, и ты уже посчитал сколько. Статей расхода три:

память продюсера. По учебной модели продюсер держит по мегабайту на каждую партицию топика: какие ключи ему достанутся, он заранее не знает. Шесть продюсеров на 64 партициях — 384 МБ, на двенадцати — 72;
файлы и память брокера. Каждая партиция — свой набор сегментов со своими оффсетами, то есть свои открытые файлы на диске брокера и своя память под них. 64 партиции — 64 таких набора;
время ребаланса. Пока группа заново решает, кто из консьюмеров какие партиции берёт, консьюмеры стоят. По учебной модели — 170 мс на партицию: 10,9 с на 64 партициях против 2,0 с на двенадцати. Механику разбирает глава 3.

Переезд: когда партиций оказалось слишком много

Уменьшить число партиций нельзя — во второй части ячейки брокер ответил отказом. Остаётся переезд, и способов два, у каждого своя цена.

С новым именем. Поднимаешь второй топик на двенадцати партициях, переключаешь на него продюсеров, какое-то время читаешь оба топика и складываешь в одно место, а дочитанный старый гасишь. Останавливать продюсеров не нужно. Платишь двумя топиками в чтении — и порядком на границе: у каждого ключа старые сообщения лежат в одном топике, новые — в другом, и пока консьюмер берёт оба, их взаимный порядок не гарантирован.

С сохранением имени. Даёшь консьюмерам дочитать, гасишь старый топик и заводишь под тем же именем новый — это третья часть ячейки: двенадцать партиций и ноль кадров. Продюсеры на время замены ждут. Платишь всей историей, которая лежала в топике.

Станция сегодня выбирает второе. Эфир в signal_raw и так живёт шесть часов, а Читальный зал к 23:10 его дочитал — терять почти нечего. Переезд назначен на полночь: с 00:00 signal_raw принимает эфир на двенадцати партициях.

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

23:10. Одна партиция и пустое табло

В той же заявке есть ещё строчка: ответы диспетчерской из signal_ack Читальный зал хочет разбирать вчетвером. А signal_ack ты утром завёл на ОДНОЙ партиции — «потому что там важен порядок».

Четыре консьюмера на одной партиции — это один работающий и трое без работы. Добавить им работы нельзя никак, кроме как переделать топик. Одна партиция — это полный порядок всех ответов, но и потолок: один брокер и один консьюмер. Это законный выбор, если назвать его цену, и ты её называешь: ответов четыре в час, порядок в них важнее скорости, одного консьюмера им хватает. Читальный зал получает ответ «один консьюмер, и вот почему».

В 23:10 ты ставишь на стойку табло отставания — экран, где виден лаг: насколько консьюмеры отстали от конца журнала. Пока оно пустое: читать его начнут завтра.

КВЕРИ: Запомни, как выглядит пустое табло. Больше ты его таким не увидишь.

Что в этом уроке настоящее, а что учебное

Числа цены — учебная модель, взятая круглой, чтобы арифметика читалась: 170 мс паузы группы на каждую партицию и мегабайт памяти продюсера на каждую партицию топика. Настоящая Kafka память продюсера так не считает: у продюсера один общий запас памяти, и он делит его между партициями по мере надобности. Поэтому 384 мегабайта из сцены — оценка сверху; как эта память устроена и что бывает, когда она кончается, разбирает урок «Производительность продюсера: батчи, linger.ms, сжатие» из трека «Углублённо». Паузу ребаланса тоже не считают «по 170 мс за партицию»: больше всего она зависит от того, как быстро консьюмеры группы заново в неё войдут, а число партиций лишь добавляет работы каждому.

18 МБ/с на партицию — не модель, а цифра из заявки Читального зала: в учебной Kafka нет ни диска, ни сети, а — копии той же партиции на других брокерах — она ведёт в памяти, без передачи данных, так что пропускную способность партиции в ней не померить. Переезд в песочнице мгновенный: delete_topic и create_topic не занимают времени, и продюсеры не ждут. Порядок сообщений ключа на границе переезда ячейка не показывает — двух топиков и переключения продюсеров на ходу здесь нет.

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

Вопрос с собеседования

Как это спрашивают на собеседовании

«Сколько партиций вы поставите на такой топик?» — вопрос почти обязательный, и кандидат, который сразу называет число, уже ответил мимо. Ждут метода: две нижние границы, одна верхняя и одно честное допущение. Нижняя по потоку (throughput floor) — нужный поток, делённый на то, что тянет одна партиция. Нижняя по консьюмерам (consumer parallelism floor) — сколько их работает на пике. Верхняя по ключу (key cap) — сколько у ключа разных значений: партиций больше не даст параллельности, лишние останутся пустыми. Берут ту нижнюю, что выше, и сверяют с верхней; если между ними осталось место, добавляют запас на рост и округляют вверх до кратного числу брокеров, чтобы партиции легли поровну. Не дали пик и число консьюмеров — их надо спросить, и это часть правильного ответа.

Отдельно оценят, как ты обойдёшься с цифрой «сколько тянет одна партиция». Сильный ответ не называет её по памяти и не берёт из статьи, а говорит, откуда она — свой замер или чужая заявка — и что будет, если она вдвое меньше.

Второй вопрос — на цену: «Почему тогда не поставить тысячу?» Ждут минимум три статьи расхода: память продюсера, растущую с числом партиций; открытые файлы и память брокера; время ребаланса группы (rebalance) — пока группа заново решает, кто из консьюмеров какие партиции берёт.

Третий — «А если ошиблись?» В меньшую сторону — партиции добавляют, и надо сказать, что раскладка ключей после этого сдвинется. В большую — уменьшить нельзя: живут с этим или переезжают (topic cutover). Хороший кандидат сразу описывает оба переезда: с новым именем и чтением двух топиков (dual-read ), когда останавливаться нельзя, и с сохранением имени, когда историей можно пожертвовать.

Проверь себя

Представь, что в заявке Читального зала пик в полтора раза выше — 324 МБ/с; остальное прежнее: 18 МБ/с на партицию, девять консьюмеров, ключ source_id. Коллега говорит: «Поток в полтора раза больше — и партиций в полтора раза больше, ставь 18». Что ответить?

Главное из урока

что	число	откуда
нижняя граница по потоку	216 ÷ 18 = 12	заявка; 18 МБ/с — допущение, не замер
нижняя граница по консьюмерам	9	заявка
верхняя граница по ключу	12	справочник: двенадцать значений `source_id`
ответ	12	нижняя и верхняя сошлись
цена 64 партиций	384 МБ памяти, 10,9 с паузы	учебная модель
`signal_ack`	1 партиция	полный порядок, один консьюмер — цена названа

Число партиций не выбирают, а считают: из двух нижних границ берут ту, что выше, и сверяют с верхней. Цифру, которую не мерил, записывают допущением вместе с источником. Лишняя партиция стоит памяти, файлов и секунд и не даёт ничего. Уменьшить нельзя — только переезд: с новым именем платишь порядком на границе и двумя топиками в чтении, с сохранением имени — историей.

КВЕРИ: Утром три наугад, вечером шестьдесят четыре наугад. Двенадцать — первое число партиций за день, которое ты посчитал.