Как взаимодействуют партиции Kafka, консьюмеры и консьюмер-группы?

Взаимодействие партиций Kafka, консьюмеров и консьюмер-групп

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

1. Структура Kafka: Топики и партиции

Топик — это логический канал для публикации данных. Каждый топик разделяется на партиции, которые распределены между брокерами для параллельной обработки:

Топик: orders
├── Партиция 0 (Брокер 1)  [msg1, msg2, msg3, msg4]
├── Партиция 1 (Брокер 2)  [msg5, msg6, msg7, msg8]
└── Партиция 2 (Брокер 3)  [msg9, msg10, msg11]

Каждая партиция — это отсортированный лог (append-only log)

Чем больше партиций, тем выше пропускная способность:

# Создание топика с 3 партициями
from kafka.admin import KafkaAdminClient, NewTopic

admin_client = KafkaAdminClient(bootstrap_servers=['localhost:9092'])
topic = NewTopic(
    name='orders',
    num_partitions=3,
    replication_factor=2
)
admin_client.create_topics([topic])

2. Консьюмеры и распределение партиций

Консьюмер — это приложение, которое читает сообщения из топика. Консьюмер-группа — это набор консьюмеров, которые вместе обрабатывают один топик.

Правило распределения

Кafka автоматически распределяет партиции между консьюмерами в группе:

Топик: orders (3 партиции)

Консьюмер-группа: analytics
├── Consumer 1 → Партиция 0
├── Consumer 2 → Партиция 1
└── Consumer 3 → Партиция 2

Консьюмер-группа: billing
├── Consumer A → Партиция 0
├── Consumer B → Партиция 1
└── Consumer C → Партиция 2

Важно: одна партиция всегда обрабатывается только одним консьюмером в группе. Это гарантирует порядок обработки.

3. Пример: несбалансированное распределение

from kafka import KafkaConsumer
import json

# Консьюмер-группа: analytics (3 консьюмера для 3 партиций)
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='analytics',  # Группа
    auto_offset_reset='earliest',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

# Если добавить 4-й консьюмер:
# Consumer 1 → Партиция 0 ✓
# Consumer 2 → Партиция 1 ✓
# Consumer 3 → Партиция 2 ✓
# Consumer 4 → (ничего)   ✗ будет в режиме ожидания

# Если удалить 1 консьюмер (осталось 2):
# Consumer 1 → Партиции 0, 1  (перебалансировка)
# Consumer 2 → Партиция 2     (перебалансировка)

4. Offsets (смещения) и прогресс обработки

Каждый консьюмер отслеживает offset — позицию в партиции, до которой он прочитал сообщения:

# Пример обработки с отслеживанием offset
for message in consumer:
    try:
        order = message.value
        process_order(order)
        # Kafka автоматически коммитит offset после успешной обработки
        consumer.commit()
    except Exception as e:
        # Если ошибка — offset не коммитится, будет переобработка
        logger.error(f"Error processing order: {e}")

Партиция 0:
[0][1][2][3][4][5][6][7][8][9]
         ↑ current offset = 3
         
След. сообщение будет с offset = 4

5. Консьюмер-группы для параллельной обработки

Разные группы могут читать один топик независимо:

# Группа 1: реал-тайм аналитика
consumer_analytics = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='analytics',
    auto_offset_reset='latest'  # Читаем только новые сообщения
)

# Группа 2: billing (может иметь свой offset)
consumer_billing = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='billing',
    auto_offset_reset='earliest'  # Читаем с начала (первый раз)
)

# Оба консьюмера читают ВСЕ 3 партиции, но независимо друг от друга

6. Координация: Rebalancing

Когда в группу добавляется или удаляется консьюмер, происходит перебалансировка (rebalancing):

# До перебалансировки (3 консьюмера)
Consumer 1 reads:  P0, P1
Consumer 2 reads:  P2
Consumer 3 reads:  (idle)

# После добавления Consumer 3
# Kafka переераспределяет партиции:
Consumer 1 reads:  P0
Consumer 2 reads:  P1
Consumer 3 reads:  P2

# Во время перебалансировки:
# - Консьюмеры останавливают обработку
# - Происходит переассignment партиций
# - Обработка возобновляется
# - Это может вызвать задержку (stop-the-world)

7. Гарантии обработки

# Three delivery semantics (три типа гарантий)

# 1. At-most-once (максимум один раз)
# Риск: потеря сообщений
consumer = KafkaConsumer(
    'orders',
    group_id='group',
    auto_commit_interval_ms=5000,  # Коммитим часто
    enable_auto_commit=True
)

# 2. At-least-once (минимум один раз)
# Риск: дублирование сообщений
consumer = KafkaConsumer(
    'orders',
    group_id='group',
    enable_auto_commit=False  # Ручной коммит
)

for message in consumer:
    try:
        process(message)
        consumer.commit()  # Коммитим только после успеха
    except:
        # Без коммита — переобработка при перезапуске
        pass

# 3. Exactly-once (ровно один раз) — самое сложное
# Требует идемпотентности + внешнее хранилище

8. Пример production架構

from kafka import KafkaConsumer
import logging
import json
from datetime import datetime

logger = logging.getLogger(__name__)

class OrderProcessor:
    def __init__(self, group_id='order_processor'):
        self.consumer = KafkaConsumer(
            'orders',
            bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
            group_id=group_id,
            auto_offset_reset='earliest',
            enable_auto_commit=False,  # Ручной коммит
            value_deserializer=lambda x: json.loads(x.decode('utf-8')),
            max_poll_records=100,  # Батч из 100 сообщений
            session_timeout_ms=30000
        )
    
    def process_batch(self):
        """Обрабатываем батч сообщений"""
        messages = self.consumer.poll(timeout_ms=5000)
        
        for topic_partition, records in messages.items():
            logger.info(
                f"Processing {len(records)} messages "
                f"from partition {topic_partition.partition}"
            )
            
            for record in records:
                try:
                    order = record.value
                    self.validate_and_store_order(order)
                except Exception as e:
                    logger.error(
                        f"Failed to process order {record.key}: {e}",
                        exc_info=True
                    )
                    # Не коммитим, будет переобработка
                    raise
            
            # Успешно обработали батч — коммитим
            self.consumer.commit()
    
    def validate_and_store_order(self, order):
        """Бизнес-логика обработки заказа"""
        if not order.get('order_id'):
            raise ValueError('Missing order_id')
        
        # Сохраняем в БД
        store_to_database(order)
        logger.info(f"Processed order {order['order_id']}")

# Запуск
if __name__ == '__main__':
    processor = OrderProcessor(group_id='order_processor_1')
    
    while True:
        try:
            processor.process_batch()
        except Exception as e:
            logger.error(f"Processing error: {e}")
            # Переподключиться и начать заново
            continue

9. Мониторинг консьюмер-групп

# Посмотреть все группы
kafka-consumer-groups --bootstrap-server localhost:9092 --list

# Описание группы (lag, offset, partition assignment)
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group analytics --describe

# Пример вывода:
# GROUP    TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# analytics orders  0          100             150             50
# analytics orders  1          200             200             0
# analytics orders  2          150             180             30

Ключевые выводы

1. Партиции = параллелизм. Больше партиций = больше консьюмеров могут читать параллельно
2. Одна партиция = один консьюмер в группе (гарантирует порядок)
3. Консьюмер-группы позволяют несколько приложений читать один топик независимо
4. Offsets отслеживают прогресс. Коммит = гарантия обработки
5. Rebalancing — это перераспределение при изменении группы (может вызвать задержку)
6. Lag = отставание консьюмера от end-of-log (показатель здоровья системы)