Kafka Python: пример consumer

Kafka consumer в Python читает сообщения из топика Kafka. Для первого примера нужен работающий Kafka-брокер, установленная библиотека kafka-python и топик, в который уже попадают сообщения

Ниже будет минимальный consumer, маленький producer для проверки, чтение topic, partition, offset и пример обработки JSON-сообщений

Установка библиотеки

pip install kafka-python

В примерах предполагается, что Kafka доступна на localhost:9092, а топик называется orders. Если у вас другой адрес или топик, замените эти значения в коде

Минимальный consumer

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='demo-consumer',
    auto_offset_reset='earliest',
)

for message in consumer:
    text = message.value.decode('utf-8')
    print(text)

Что здесь важно:

• orders — имя топика
• bootstrap_servers — адрес Kafka-брокера
• group_id — группа consumer
• auto_offset_reset='earliest' — читать с начала, если у группы еще нет offset

Цикл for message in consumer будет ждать новые сообщения. Это нормальное поведение: consumer обычно работает постоянно

По этой теме полезно отдельно посмотреть Составьте еженедельное расписание занятий, чтобы расширить контекст и сравнить подходы

По этой теме полезно отдельно посмотреть Аннотации типов для декораторов в Python, чтобы расширить контекст и сравнить подходы

Producer для проверки

Чтобы проверить consumer, удобно отправить несколько сообщений отдельным скриптом

from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda value: value.encode('utf-8'),
)

producer.send('orders', 'order-1: paid')
producer.send('orders', 'order-2: created')
producer.flush()
producer.close()

Запустите consumer в одном терминале, producer — в другом. В окне consumer должны появиться отправленные строки

Вывести topic, partition и offset

Сообщение Kafka содержит не только value, но и служебные поля

for message in consumer:
    print('topic:', message.topic)
    print('partition:', message.partition)
    print('offset:', message.offset)
    print('value:', message.value.decode('utf-8'))

offset — позиция сообщения внутри partition. Он помогает понимать, какие сообщения уже прочитаны consumer-группой

Остановить consumer после нескольких сообщений

Для учебной проверки удобно не держать consumer бесконечно

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='demo-consumer-limited',
    auto_offset_reset='earliest',
)

limit = 5
count = 0

for message in consumer:
    print(message.value.decode('utf-8'))
    count += 1

    if count >= limit:
        break

consumer.close()

В production-коде consumer чаще работает постоянно, но для обучения и тестов лимит удобен

JSON-сообщения

В реальных проектах в Kafka часто отправляют JSON

Producer:

import json
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda value: json.dumps(value).encode('utf-8'),
)

producer.send('orders', {'order_id': 101, 'status': 'paid'})
producer.send('orders', {'order_id': 102, 'status': 'created'})
producer.flush()
producer.close()

Consumer:

import json
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers='localhost:9092',
    group_id='json-consumer',
    auto_offset_reset='earliest',
    value_deserializer=lambda value: json.loads(value.decode('utf-8')),
)

for message in consumer:
    data = message.value
    print(data['order_id'], data['status'])

value_deserializer превращает байты из Kafka в Python-словарь. После этого с сообщением можно работать как с обычными данными

Обработка плохого JSON

Если сообщения приходят из разных источников, JSON может быть неправильным. Тогда лучше не ронять весь consumer

import json
from json import JSONDecodeError

for message in consumer:
    try:
        data = json.loads(message.value.decode('utf-8'))
    except JSONDecodeError:
        print(f'Плохой JSON на offset {message.offset}')
        continue

    print(data)

continue пропускает плохое сообщение и переходит к следующему

Частые ошибки

Запускать consumer без работающей Kafka

Не создавать топик или писать в другой топик

Использовать один и тот же group_id и удивляться, почему старые сообщения больше не читаются с начала

Не понимать, что consumer может ждать новые сообщения и не завершаться сам

Не закрывать consumer в учебных скриптах

Парсить JSON без обработки ошибок

Практические задачи

Отправьте три текстовых сообщения через producer и прочитайте их consumer

Выведите topic, partition, offset и value

Сделайте consumer, который останавливается после пяти сообщений

Отправьте JSON с order_id и status

Добавьте обработку JSONDecodeError

Поменяйте group_id и посмотрите, как меняется чтение сообщений

Kafka consumer на Python создается через KafkaConsumer, подключается к топику и читает сообщения в цикле. Для первого запуска важно иметь Kafka-брокер, правильный bootstrap_servers, понятный group_id и тестовые сообщения в топике

Для реальных сценариев добавляйте обработку JSON, логирование offset, контроль ошибок и аккуратное завершение consumer