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

Партиции и ключ: где сохраняется порядок

18 мин

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

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

18:20. Двадцать, восемь и двадцать

2 ноября 2184 года. Читальный зал читает эфир группой из трёх консьюмеров — по консьюмеру на партицию. Сорок минут назад он упёрся в срок хранения, а теперь пишет о другом: первый консьюмер не успевает, второй почти простаивает. Счётчики на стойках показывают почему: в партицию 0 идёт 20 кадров в секунду, в партицию 1 — 8, в партицию 2 — снова 20.

У зала два предложения. Первое: «Уберите ключ — пусть Kafka раскладывает кадры поровну, по 16 на партицию». Второе: «Добавим в группу четвёртого консьюмера, пусть помогает».

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

КВЕРИ: Поровну — это когда не жалуется консьюмер. Правильно — это когда не жалуется запись.

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

Воспроизведём секунду эфира: двенадцать антенн по четыре кадра, ключ — имя антенны, топик на три партиции. Ячейка печатает три вещи: какие антенны легли в какую партицию; что прочитал каждый из четырёх консьюмеров группы reading_room и в каком порядке пришли кадры s02; в какую партицию попадают две антенны при другом числе партиций.

from arena_kafka import cluster, Producer, Consumer, partition_for

ANTENNAS = ['s%02d' % n for n in range(1, 13)]

# 1. Секунда эфира: двенадцать антенн по четыре кадра. Ключ — имя антенны.
cluster.create_topic('signal_raw', partitions=3)
producer = Producer()
for frame in range(4):
    for antenna in ANTENNAS:
        producer.send('signal_raw', {'antenna': antenna, 'frame': frame}, key=antenna)
producer.flush()
for partition, keys in cluster.keys_by_partition('signal_raw').items():
    print('партиция %d: антенны %s' % (partition, ' '.join(keys)))
print()

# 2. Группа Читального зала: четыре консьюмера на три партиции.
group = [Consumer('signal_raw', group_id='reading_room', member_id='c%d' % n,
                  auto_offset_reset='earliest') for n in range(1, 5)]
read = []
for consumer in group:
    messages = consumer.poll(max_records=100)
    read += messages
    partitions = [tp.partition for tp in consumer.assignment()]
    print('%s: партиции %s, кадров за секунду: %d' % (consumer.member_id, partitions, len(messages)))
print('кадры s02 в порядке чтения:', [m.value['frame'] for m in read if m.key == 's02'])
print()

# 3. Партиция ключа зависит от числа партиций: хеш ключа по модулю их числа.
for n in [3, 4, 6, 12]:
    print('партиций %2d: s03 -> %2d, s05 -> %2d' % (n, partition_for('s03', n), partition_for('s05', n)))

from arena_kafka import cluster, Producer, Consumer, partition_for

ANTENNAS = ['s%02d' % n for n in range(1, 13)]

# 1. Секунда эфира: двенадцать антенн по четыре кадра. Ключ — имя антенны.
cluster.create_topic('signal_raw', partitions=3)
producer = Producer()
for frame in range(4):
    for antenna in ANTENNAS:
        producer.send('signal_raw', {'antenna': antenna, 'frame': frame}, key=antenna)
producer.flush()
for partition, keys in cluster.keys_by_partition('signal_raw').items():
    print('партиция %d: антенны %s' % (partition, ' '.join(keys)))
print()

# 2. Группа Читального зала: четыре консьюмера на три партиции.
group = [Consumer('signal_raw', group_id='reading_room', member_id='c%d' % n,
                  auto_offset_reset='earliest') for n in range(1, 5)]
read = []
for consumer in group:
    messages = consumer.poll(max_records=100)
    read += messages
    partitions = [tp.partition for tp in consumer.assignment()]
    print('%s: партиции %s, кадров за секунду: %d' % (consumer.member_id, partitions, len(messages)))
print('кадры s02 в порядке чтения:', [m.value['frame'] for m in read if m.key == 's02'])
print()

# 3. Партиция ключа зависит от числа партиций: хеш ключа по модулю их числа.
for n in [3, 4, 6, 12]:
    print('партиций %2d: s03 -> %2d, s05 -> %2d' % (n, partition_for('s03', n), partition_for('s05', n)))

python · kafka

Партиция — единица параллельности

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

Как ключ выбирает партицию

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

один ключ — всегда одна партиция. Все кадры s02 лежат в партиции 0, и консьюмер получил их в том порядке, в каком их отправили: 0, 1, 2, 3;
разные ключи делят партиции как повезёт. Хеш ничего не знает о нагрузке: двенадцать антенн легли как пять, две и пять, поэтому у c2 восемь кадров в секунду, а у c1 и c3 по двадцать;
сменилось число партиций — ключи переехали. s03 при трёх партициях лежит в партиции 0, при четырёх — в 2, при двенадцати — в 6. Поэтому партиции не добавляют между делом: их число считают заранее, это урок «Сколько партиций: посчитать, а не угадать».

Порядок Kafka обещает только внутри партиции. Между партициями его нет: консьюмер может прочитать сообщение из партиции 2 раньше, чем более старое сообщение из партиции 0.

Без ключа продюсер раздаёт буфер антенны по партициям, и консьюмер читает кадры не в том порядке. С ключом все кадры антенны лежат в одной партиции и читаются так, как их отправили.

Без ключа порядка нет

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

Ключ выбирают по вопросу «где нужен порядок». Нужен по антенне — ключ антенна. Нужен по заказу — номер заказа. Не нужен вовсе — ключ можно не ставить. А неровные 20, 8 и 20 отказом от ключа не лечат: сколько партиций нужно, глава посчитает вечером, а почему ключ и тогда ложится неровно — разговор главы 2.

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

Ночная смена оставила перекладчик: он забирает у антенны буфер — четыре кадра подряд — и отправляет их в signal_raw. Ключ в нём забыли, и Читальный зал получает кадры одной антенны вразнобой. Почини send_buffer(producer, antenna, frames): кадры каждой антенны должны читаться в том порядке, в каком их отправили, а эфир — по-прежнему идти во все три партиции. Функция replay_buffers(send_buffer) уже подключена, её код скрыт. Она заводит signal_raw на три партиции, прогоняет через твою функцию буферы двенадцати антенн, читает топик группой reading_room и возвращает два словаря: номера кадров каждой антенны в порядке чтения и партиции, куда легли её кадры. Нижние строки заготовки печатают это для s01 — их вывод виден, когда жмёшь «Проверить».

def send_buffer(producer, antenna, frames):
    """Отправить буфер антенны в signal_raw: кадры одной антенны идут подряд."""
    for frame in frames:
        producer.send('signal_raw', frame)


# Самопроверка: в каком порядке Читальный зал прочитает кадры s01.
order, parts = replay_buffers(send_buffer)
print('s01: кадры', order['s01'], 'из партиций', sorted(parts['s01']))

python · kafka

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

Правила настоящие: один ключ — одна партиция, порядок только внутри партиции, сменилось число партиций — ключи переехали, в группе партицию читает один консьюмер.

Хеш ключа учебная Kafka считает функцией crc32 — так по умолчанию делает клиент librdkafka, на нём стоит confluent-kafka-python. Клиент на Java и kafka-python считают murmur2, и у них антенны легли бы по партициям иначе, чем 5, 2 и 5, — но так же неровно. Без ключа учебная Kafka раздаёт сообщения по кругу, настоящие клиенты — батчами или случайно; порядок антенны ломается при любом способе. Имя консьюмера member_id учебная Kafka принимает для наглядности: настоящему консьюмеру его выдаёт Kafka при входе в группу.

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

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

«Как Kafka гарантирует порядок?» Только внутри партиции. Сообщения, порядок которых важен, получают один ключ: продюсер берёт хеш ключа по модулю числа партиций, и одинаковые ключи попадают в одну партицию. Общего порядка у топика из нескольких партиций нет.

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

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

Английские имена: partition key, partitioner, hot partition — партиция, в которую пришлось слишком много.

Проверь себя

Заказ проходит статусы «создан», «оплачен», «отгружен». Продюсер отправляет их без ключа в топик из шести партиций, а витрина записывает последний прочитанный статус. Что увидит витрина?

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

что	как устроено	у нас
партиция	в группе её читает один консьюмер: работающих консьюмеров не больше, чем партиций	три партиции — `c4` без работы
ключ	хеш ключа по модулю числа партиций: один ключ — одна партиция	антенны легли 5, 2, 5 — это 20, 8 и 20 кадров в секунду
порядок	только внутри партиции	кадры `s02` пришли 0, 1, 2, 3
без ключа	нагрузка ровнее, порядок антенны теряется	буфер `s01` прочитан как 0, 3, 1, 2
число партиций	сменилось — ключи переехали	`s03`: партиция 0 при трёх, 6 при двенадцати

Ключ выбирают по вопросу «где нужен порядок». Неровную нагрузку отказом от ключа не лечат.

КВЕРИ: Ключ оставили. Зал поворчал и согласился: запись дороже.