Apache Kafka для Data Engineers: Партиції, Consumer Groups та Потокова Обробка Даних

Apache Kafka залишається фундаментальним компонентом сучасних платформ обробки даних. Від потокового завантаження до Change Data Capture, від подієво-орієнтованих архітектур до побудови реального часу аналітичних конвеєрів — Kafka забезпечує надійну доставку мільярдів повідомлень щодня в таких компаніях, як Netflix, Uber та LinkedIn. Для інженера даних глибоке розуміння внутрішніх механізмів Kafka — партиціонування, гарантій доставки, управління зсувами — є обов'язковою компетенцією при проєктуванні масштабованих конвеєрів обробки даних.

Архітектура потокової обробки в Kafka

Kafka функціонує як розподілений незмінний журнал (distributed commit log). Продюсери записують повідомлення в топіки, кожен топік поділяється на партиції, а кожна партиція реплікується на декількох брокерах. Ця модель забезпечує три ключові властивості для конвеєрів даних: гарантований порядок повідомлень у межах партиції, конфігурований термін зберігання даних та горизонтальну масштабованість.

Типовий продакшн-кластер налічує від 3 до 30 брокерів залежно від обсягу даних та вимог до затримки. Кожен брокер зберігає сегменти партицій на диску, використовуючи механізм сторінок ядра операційної системи для досягнення надзвичайно швидких послідовних читань. Цей патерн, відомий як zero-copy, дозволяє Kafka досягати пропускної здатності понад 1 ГБ/с на брокер на сучасному обладнанні.

Зв'язок між кількістю партицій та паралелізмом є прямим: кількість партицій топіка визначає максимальну кількість одночасних споживачів у межах consumer group. Якщо топік має 6 партицій, максимум 6 споживачів можуть обробляти дані паралельно. Додавання більшої кількості споживачів просто залишає зайвих без роботи.

Kafka 4.0, випущений у березні 2025 року, остаточно усуває залежність від ZooKeeper. Режим KRaft управляє метаданими кластера безпосередньо в брокерах, використовуючи протокол консенсусу на базі Raft. Для інженерів даних це означає суттєво простіші розгортання — локальний кластер для розробки запускається одним файлом Docker Compose.

# docker-compose.yml — Kafka 4.x KRaft cluster (no ZooKeeper)
services:
  kafka-1:
    image: apache/kafka:4.2.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_NUM_PARTITIONS: 6
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2
    ports:
      - "9092:9092"

Параметр KAFKA_PROCESS_ROLES: broker,controller вказує, що кожен вузол виконує обидві ролі. У продакшн-середовищі прийнято розділяти ролі контролера та брокера на окремі вузли для ізоляції навантаження управління метаданими від обробки повідомлень. Фактор реплікації 3 з min.insync.replicas рівним 2 гарантує, що будь-яке підтверджене повідомлення з acks=all переживе падіння одного брокера.

Стратегії партиціонування для конвеєрів даних

Вибір ключа партиціонування визначає порядок повідомлень, розподіл навантаження та здатність масштабувати споживачів. Неправильний вибір ключа може призвести до розбалансованих партицій (hot partitions), де одна партиція накопичує непропорційно велике навантаження, а решта залишаються недозавантаженими.

Існують три основні стратегії вибору ключів партиціонування в контексті інженерії даних.

Партиціонування за бізнес-сутністю. Використання ідентифікатора основної сутності (customer_id, account_id) як ключа. Усі події одного клієнта потрапляють у ту саму партицію, що гарантує хронологічний порядок. Це найпоширеніша стратегія для конвеєрів замовлень, транзакцій та активності користувачів.

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

Партиціонування round-robin. Коли порядок не має значення (логи, загальні метрики), відсутність ключа розподіляє повідомлення рівномірно. Ця стратегія максимізує паралелізм ціною втрати будь-яких гарантій порядку.

# partition_strategy.py — Choosing the right partition key
from confluent_kafka import Producer
import json

def create_producer():
    return Producer({
        'bootstrap.servers': 'kafka-1:9092',
        'acks': 'all',
        'enable.idempotence': True,
        'max.in.flight.requests.per.connection': 5
    })

def publish_order_event(producer, event):
    # Key by customer_id: all events for one customer stay ordered
    key = str(event['customer_id']).encode('utf-8')
    value = json.dumps(event).encode('utf-8')
    producer.produce(
        topic='order-events',
        key=key,
        value=value,
        headers=[('source', b'order-service'), ('version', b'2')]
    )
    producer.flush()

def publish_clickstream(producer, event):
    # Key by session_id: ordering within a session matters
    # NOT by user_id — one user may have multiple sessions
    key = str(event['session_id']).encode('utf-8')
    value = json.dumps(event).encode('utf-8')
    producer.produce(
        topic='clickstream',
        key=key,
        value=value
    )

Конфігурація enable.idempotence: True разом з acks: all активує ідемпотентне виробництво, запобігаючи дублікатам при повторних спробах відправки через таймаути мережі. Поєднання з max.in.flight.requests.per.connection: 5 зберігає порядок навіть при активних повторних спробах — покращення, впроваджене ще в Kafka 2.0, що усунуло попередню необхідність обмежувати кількість запитів у польоті до 1.

Consumer Groups та ручне управління зсувами

Consumer groups є основним механізмом паралелізму споживання в Kafka. Кожен споживач у межах групи отримує ексклюзивний набір партицій. Коли споживач виходить з ладу або додається новий, Kafka виконує ребалансування, що перерозподіляє партиції між активними членами групи.

Для конвеєрів даних рекомендованим патерном є вимкнення автоматичного коміту зсувів та ручне підтвердження після успішної обробки кожного пакету. Такий підхід гарантує семантику at-least-once: якщо споживач виходить з ладу до підтвердження, повідомлення будуть оброблені повторно при перезапуску. Компромісом є необхідність ідемпотентності логіки обробки для коректної обробки дублікатів.

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

# consumer_pipeline.py — Consumer group with manual offset management
from confluent_kafka import Consumer, KafkaError
import json

def create_consumer(group_id: str):
    return Consumer({
        'bootstrap.servers': 'kafka-1:9092',
        'group.id': group_id,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,  # Manual commit for at-least-once
        'max.poll.interval.ms': 300000,
        'session.timeout.ms': 45000,
        'isolation.level': 'read_committed'  # Only read committed transactions
    })

def process_batch(consumer: Consumer, batch_size: int = 500):
    """Process records in micro-batches for throughput."""
    consumer.subscribe(['order-events'])
    buffer = []

    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            raise Exception(msg.error())

        record = json.loads(msg.value().decode('utf-8'))
        buffer.append(record)

        if len(buffer) >= batch_size:
            # Process the full batch (write to warehouse, etc.)
            write_to_warehouse(buffer)
            # Commit AFTER successful processing
            consumer.commit(asynchronous=False)
            buffer.clear()

def write_to_warehouse(records: list):
    # Insert batch into data warehouse (BigQuery, Snowflake, etc.)
    pass

Параметр isolation.level: read_committed є критично важливим при використанні транзакцій продюсерами. З цією конфігурацією споживач читає лише повідомлення, підтверджені успішною транзакцією, уникаючи обробки даних з відхилених транзакцій.

Конфігурація max.poll.interval.ms: 300000 (5 хвилин) визначає максимальний час між викликами poll(). Якщо обробка пакету займає більше цього інтервалу, Kafka вважає споживача неактивним та ініціює ребалансування. Для великих пакетів, що записуються у повільні сховища даних, необхідно збільшити це значення.

Change Data Capture з Debezium та Kafka Connect

Change Data Capture (CDC) фіксує зміни на рівні рядків у вихідній базі даних та публікує їх як події у топіки Kafka. Цей патерн усуває необхідність пакетного ETL для синхронізації операційних баз даних з аналітичними сховищами.

Debezium — еталонний конектор CDC для Kafka Connect — читає журнал реплікації PostgreSQL (WAL), MySQL (binlog) або MongoDB (oplog) та перетворює кожну операцію INSERT, UPDATE та DELETE у структуровану подію. Конектор підтримує свою позицію в журналі, гарантуючи, що жодна зміна не буде пропущена навіть під час перезапусків.

Цей підхід має три визначальні переваги: латентність близька до нуля між зміною в базі та її поширенням, відсутність додаткового навантаження на вихідну базу (читання WAL є пасивним), та повне охоплення всіх операцій, включаючи видалення.

{
  "name": "postgres-cdc-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-primary",
    "database.port": "5432",
    "database.user": "replication_user",
    "database.dbname": "production",
    "topic.prefix": "cdc",
    "table.include.list": "public.orders,public.customers,public.products",
    "slot.name": "debezium_slot",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_publication",
    "snapshot.mode": "initial",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "cdc\\.public\\.(.*)",
    "transforms.route.replacement": "warehouse.$1"
  }
}

Поле snapshot.mode: initial вказує, що Debezium спочатку захоплює повний знімок налаштованих таблиць, а потім переходить до потокової передачі інкрементальних змін. Трансформація RegexRouter перейменовує топіки з cdc.public.orders на warehouse.orders, полегшуючи логічну організацію топіків за призначенням.

Replication slot (debezium_slot) гарантує, що PostgreSQL не видалить сегменти WAL, які ще не були спожиті. Критично важливо моніторити розмір слоту: якщо конектор зупиняється на тривалий час, слот може накопичити гігабайти WAL, впливаючи на дисковий простір сервера бази даних. Рекомендується налаштувати обмеження через параметр max_slot_wal_keep_size у PostgreSQL.

Exactly-Once семантика: транзакції в Kafka

Семантика exactly-once гарантує, що кожне повідомлення обробляється рівно один раз, навіть при збоях мережі або перезапусках споживачів. Kafka реалізує exactly-once через транзакції, що охоплюють повний цикл consume-transform-produce в одній атомарній операції.

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

# exactly_once_pipeline.py — Transactional consume-transform-produce
from confluent_kafka import Consumer, Producer
import json

def transactional_pipeline():
    consumer = Consumer({
        'bootstrap.servers': 'kafka-1:9092',
        'group.id': 'etl-transformer',
        'enable.auto.commit': False,
        'isolation.level': 'read_committed'
    })

    producer = Producer({
        'bootstrap.servers': 'kafka-1:9092',
        'transactional.id': 'etl-transformer-001',
        'acks': 'all',
        'enable.idempotence': True
    })

    producer.init_transactions()
    consumer.subscribe(['raw-events'])

    while True:
        msg = consumer.poll(1.0)
        if msg is None or msg.error():
            continue

        # Begin atomic transaction
        producer.begin_transaction()
        try:
            raw = json.loads(msg.value())
            enriched = transform(raw)

            # Write transformed record to output topic
            producer.produce(
                'enriched-events',
                key=msg.key(),
                value=json.dumps(enriched).encode('utf-8')
            )

            # Commit consumer offset within the same transaction
            producer.send_offsets_to_transaction(
                consumer.position(consumer.assignment()),
                consumer.consumer_group_metadata()
            )
            producer.commit_transaction()
        except Exception:
            producer.abort_transaction()

def transform(record: dict) -> dict:
    # Apply business logic transformations
    record['processed_at'] = '2026-04-20T00:00:00Z'
    record['pipeline_version'] = '2.1'
    return record

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

Exactly-once вносить вимірне збільшення латентності (від 50 до 200 мс додатково на транзакцію). Для конвеєрів, де латентність є критичною, а логіка обробки вже є ідемпотентною, семантика at-least-once з ручним комітом може бути доцільнішим вибором. Транзакції Kafka гарантують exactly-once в межах екосистеми Kafka (consume-transform-produce). Для наскрізної exactly-once гарантії до зовнішньої системи (бази даних, сховища) необхідно поєднувати транзакції Kafka з ідемпотентними записами у призначення.

Share Groups: нова модель споживання у Kafka 4.x

Kafka 4.0 запроваджує Share Groups — модель споживання, що дозволяє декільком споживачам обробляти повідомлення з однієї партиції одночасно. На відміну від класичних consumer groups, де кожна партиція призначається виключно одному споживачу, Share Groups розподіляють окремі повідомлення між членами групи.

Ця модель особливо корисна для навантажень, де порядок не є критичним, але паралелізм є пріоритетом: обробка сповіщень, незалежні валідації, завдання збагачення без стану. Share Groups дозволяють масштабувати споживачів за межі кількості партицій, усуваючи історичне обмеження Kafka.

| Feature | Consumer Groups | Share Groups | |---|---|---| | Partition assignment | Exclusive (one consumer per partition) | Shared (multiple consumers per partition) | | Ordering guarantee | Per-partition ordering preserved | No ordering guarantee | | Use case | Ordered event streams | Task distribution, work queues | | Acknowledgement | Offset-based (commit position) | Per-record (acknowledge individually) | | Max parallelism | Limited by partition count | Limited by consumer count |

Share Groups не замінюють традиційні consumer groups. Обидві моделі співіснують та відповідають різним потребам. Для ETL-конвеєрів, де порядок подій є критичним, класичні consumer groups залишаються єдиним правильним вибором. Для розподілу обчислювальних завдань (генерація звітів, відправка сповіщень, незалежні валідації) Share Groups пропонують кращу масштабованість без обмеження кількістю партицій.

Питання для технічних співбесід з Apache Kafka

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

Як Kafka гарантує порядок повідомлень? Kafka гарантує порядок виключно в межах однієї партиції. Для збереження порядку пов'язаних подій необхідно використовувати ключ партиціонування, що групує ці події в одній партиції (наприклад, customer_id для подій одного клієнта). Глобального порядку між партиціями не існує.

У чому різниця між at-least-once та exactly-once? При at-least-once повідомлення може бути оброблене більше одного разу, якщо споживач виходить з ладу після обробки, але до підтвердження зсуву. Exactly-once використовує транзакції для атомарного підтвердження зсуву споживача та виробництва трансформованого повідомлення. Ця тема тісно пов'язана з патернами ETL та ELT.

Як працює ребалансування consumer groups? Коли споживач приєднується або залишає групу, Kafka перерозподіляє партиції між активними членами. Під час ребалансування обробка зупиняється. Протокол Cooperative Sticky Assignor мінімізує переривання, перерозподіляючи лише необхідні партиції.

Що відбувається, коли споживач занадто довго обробляє повідомлення? Якщо час між викликами poll() перевищує max.poll.interval.ms, Kafka вважає споживача неактивним та ініціює ребалансування. Рішення: зменшити розмір пакету, збільшити інтервал або обробляти в окремому потоці.

Чому Kafka 4.x позбавився ZooKeeper? ZooKeeper додавав операційну складність, обмежував масштабованість метаданих та вимагав окремого кластера з власним управлінням. KRaft інтегрує управління метаданими безпосередньо в брокери Kafka, спрощуючи розгортання та прискорюючи відновлення після збоїв.

Коли доцільно використовувати Share Groups замість consumer groups? Share Groups підходять, коли порядок повідомлень не є вимогою та потрібно масштабувати споживачів за межі кількості партицій. Вони ідеальні для навантажень типу черги завдань: обробка сповіщень, незалежні валідації, збагачення без стану.

Як запобігти hot partitions? Моніторинг затримки (lag) по кожній партиції та аналіз розподілу ключів. Якщо один ключ концентрує занадто багато повідомлень, можна використати складений ключ, додати випадковий суфікс (ціною втрати порядку) або перепартиціонувати топік. Цей принцип розподілу даних тісно пов'язаний з моделюванням даних у контексті інженерії даних.

Підсумки

Apache Kafka є центральним елементом сучасних архітектур інженерії даних реального часу. Ключові висновки для опанування Kafka як інженера даних:

• Kafka 4.x усуває ZooKeeper на користь KRaft, спрощуючи розгортання та масштабуючи управління метаданими до мільйонів партицій
• Вибір ключа партиціонування визначає порядок, паралелізм та розподіл навантаження — customer_id для конвеєрів замовлень, session_id для клікстриму
• Consumer groups з ручним комітом зсувів забезпечують семантику at-least-once — найпоширеніший патерн для конвеєрів завантаження у сховища даних
• CDC з Debezium захоплює зміни на рівні рядків з PostgreSQL, MySQL чи MongoDB без навантаження на операційну базу даних запитами екстракції
• Транзакції Kafka забезпечують exactly-once в межах екосистеми Kafka, обгортаючи consume-transform-produce в атомарну операцію
• Share Groups у Kafka 4.x дозволяють масштабувати споживачів за межі кількості партицій для навантажень, де порядок не є критичним
• На технічних співбесідах очікується глибоке розуміння партиціонування, семантик доставки, ребалансування consumer groups та відмінностей між KRaft і ZooKeeper