Создание Apache Kafka потоков данных на Python: руководство разработчика

pip install confluent-kafka

docker-compose -f docker-compose-kafka.yml up -d

version: '3'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- "2181:2181"

kafka:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

from confluent_kafka.admin import AdminClient

admin_client = AdminClient({'bootstrap.servers': 'localhost:9092'})
metadata = admin_client.list_topics(timeout=10)

print("Доступные топики Kafka:")
for topic, topic_metadata in metadata.topics.items():
print(f" – {topic}")

from confluent_kafka import Producer
import json

# Настройка Producer
producer_config = {
'bootstrap.servers': 'localhost:9092',
'client.id': 'python-producer-1'
}

producer = Producer(producer_config)

# Функция для обработки результата отправки
def delivery_report(err, msg):
if err is not None:
print(f'Ошибка доставки сообщения: {err}')
else:
print(f'Сообщение доставлено в {msg.topic()} [{msg.partition()}] на позицию {msg.offset()}')

# Данные для отправки
data = {
'user_id': 123,
'action': 'purchase',
'product_id': 456,
'timestamp': 1634567890
}

# Сериализация и отправка
producer.produce(
topic='user-events',
key=str(data['user_id']),
value=json.dumps(data),
callback=delivery_report
)

# Важно вызвать flush() для отправки всех сообщений
producer.flush()

producer_config = {
'bootstrap.servers': 'localhost:9092',
'client.id': 'python-producer-advanced',

# Настройки для повышения пропускной способности
'batch.size': 32768, # Размер батча в байтах
'linger.ms': 5, # Время ожидания формирования батча

# Настройки для обеспечения надежности
'acks': 'all', # Требуем подтверждения от всех реплик
'retries': 3, # Количество повторов при ошибках
'retry.backoff.ms': 100, # Задержка между повторами

# Настройки для контроля памяти
'buffer.memory': 33554432, # 32 МБ для буфера сообщений
'max.in.flight.requests.per.connection': 5
}

import time
from confluent_kafka import Producer
import json
import random

producer = Producer(producer_config)

# Генерируем и отправляем 100,000 сообщений
for i in range(100000):
user_id = random.randint(1, 1000)
data = {
'user_id': user_id,
'action': 'view',
'item_id': random.randint(1, 10000),
'timestamp': int(time.time())
}

producer.produce(
topic='user-events',
key=str(user_id),
value=json.dumps(data),
callback=delivery_report
)

# Периодически вызываем poll() для обработки колбэков
if i % 1000 == 0:
producer.poll(0)

# В конце обязательно вызываем flush()
producer.flush(timeout=30)

from confluent_kafka import Consumer, KafkaError

# Настройка Consumer
consumer_config = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'python-consumer-group-1',
'auto.offset.reset': 'earliest',
'enable.auto.commit': True,
'auto.commit.interval.ms': 5000
}

consumer = Consumer(consumer_config)

# Подписка на топик
consumer.subscribe(['user-events'])

# Бесконечный цикл получения сообщений
try:
while True:
msg = consumer.poll(timeout=1.0)

if msg is None:
continue

if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
print(f'Достигнут конец партиции {msg.partition()}')
else:
print(f'Ошибка: {msg.error()}')
else:
# Обработка сообщения
print(f'Получено сообщение: {msg.value().decode("utf-8")}')
print(f' Топик: {msg.topic()}, Партиция: {msg.partition()}, Смещение: {msg.offset()}')

except KeyboardInterrupt:
pass
finally:
# Закрытие Consumer при завершении
consumer.close()

from confluent_kafka import Consumer, KafkaError, TopicPartition
import json
import time

# Расширенная конфигурация
consumer_config = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'python-consumer-advanced',
'auto.offset.reset': 'earliest',
'enable.auto.commit': False, # Отключаем автоматический коммит
'max.poll.interval.ms': 300000, # 5 минут на обработку батча
'session.timeout.ms': 30000, # 30 секунд таймаут сессии
'heartbeat.interval.ms': 10000, # 10 секунд между heartbeat
}

consumer = Consumer(consumer_config)
consumer.subscribe(['user-events'])

try:
while True:
# Получаем батч сообщений (до 100)
messages = consumer.consume(num_messages=100, timeout=1.0)

if not messages:
continue

# Обрабатываем полученные сообщения
for msg in messages:
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
print(f'Ошибка: {msg.error()}')
continue

# Десериализуем и обрабатываем сообщение
try:
value = json.loads(msg.value().decode('utf-8'))
process_message(value) # Предполагаемая функция обработки
except Exception as e:
print(f'Ошибка обработки: {e}')

# Коммитим смещения после успешной обработки всего батча
consumer.commit()

except KeyboardInterrupt:
pass
finally:
consumer.close()

import os
import signal
import threading
import time
from confluent_kafka import Consumer

# Обработчик сигнала для корректного завершения
stop_event = threading.Event()

def signal_handler(sig, frame):
print('Получен сигнал остановки')
stop_event.set()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

# Создаем несколько потоков потребителей
def consumer_worker(worker_id):
consumer = Consumer({
'bootstrap.servers': 'localhost:9092',
'group.id': 'python-consumer-group',
'auto.offset.reset': 'earliest',
'enable.auto.commit': True
})

consumer.subscribe(['user-events'])

try:
while not stop_event.is_set():
msg = consumer.poll(timeout=1.0)
if msg and not msg.error():
print(f'Воркер {worker_id}: Получено сообщение: {msg.value().decode("utf-8")}')
finally:
consumer.close()
print(f'Воркер {worker_id} завершил работу')

# Запуск нескольких потоков-потребителей
workers = []
for i in range(3): # Создаем 3 потребителя в группе
worker = threading.Thread(target=consumer_worker, args=(i,))
workers.append(worker)
worker.start()

# Ожидание завершения
stop_event.wait()
for worker in workers:
worker.join()

print('Сервис остановлен')

from confluent_kafka import Producer, KafkaException
import json
import time
import socket
import logging

logging.basicConfig(level=logging.INFO, 
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s')
logger = logging.getLogger('kafka-producer')

class RobustProducer:
def __init__(self, bootstrap_servers):
self.config = {
'bootstrap.servers': bootstrap_servers,
'client.id': f'python-producer-{socket.gethostname()}',
'retries': 5,
'retry.backoff.ms': 100,
'acks': 'all',
'linger.ms': 5,
'batch.size': 32768,
'compression.type': 'snappy', # Сжатие данных
}
self._producer = None
self._connect()

def _connect(self):
try:
if self._producer is not None:
self._producer.flush()
self._producer = Producer(self.config)
logger.info("Producer подключен к Kafka")
except KafkaException as e:
logger.error(f"Ошибка при создании Producer: {e}")
raise

def _delivery_report(self, err, msg):
if err is not None:
logger.error(f'Ошибка доставки: {err} для сообщения: {msg.value()[:30]}...')
# Здесь можно добавить сохранение сообщения для повторной отправки
else:
logger.debug(f'Сообщение доставлено в {msg.topic()}[{msg.partition()}] @ {msg.offset()}')

def send(self, topic, value, key=None, headers=None, max_retries=3):
retry_count = 0
while retry_count <= max_retries:
try:
# Сериализация данных
value_bytes = json.dumps(value).encode('utf-8')
if key:
key_bytes = str(key).encode('utf-8')
else:
key_bytes = None

# Отправка сообщения
self._producer.produce(
topic=topic,
value=value_bytes,
key=key_bytes,
headers=headers,
callback=self._delivery_report
)

# Периодически вызываем poll для обработки колбэков
self._producer.poll(0)
return True

except BufferError:
# Буфер сообщений переполнен, ждем и пробуем снова
logger.warning("Буфер Producer переполнен, ожидание...")
self._producer.poll(1)
retry_count += 1

except KafkaException as e:
logger.error(f"Ошибка Kafka при отправке: {e}")
if retry_count < max_retries:
time.sleep(0.5 * (retry_count + 1)) # Экспоненциальный backoff
retry_count += 1
try:
self._connect() # Пробуем переподключиться
except:
pass
else:
raise

return False

def flush(self, timeout=None):
if self._producer:
return self._producer.flush(timeout)

def close(self):
if self._producer:
self._producer.flush()
# Закрытия Producer напрямую нет в confluent-kafka
# Он закроется при сборке мусора
self._producer = None

from confluent_kafka import Consumer, KafkaError, KafkaException
import json
import time
import socket
import logging
import sys

logging.basicConfig(level=logging.INFO, 
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s')
logger = logging.getLogger('kafka-consumer')

class RobustConsumer:
def __init__(self, bootstrap_servers, group_id, topics, error_handler=None):
self.config = {
'bootstrap.servers': bootstrap_servers,
'group.id': group_id,
'auto.offset.reset': 'earliest',
'enable.auto.commit': False,
'max.poll.interval.ms': 600000, # 10 минут
'session.timeout.ms': 30000,
'heartbeat.interval.ms': 10000,
'max.partition.fetch.bytes': 1048576, # 1MB
}
self.topics = topics if isinstance(topics, list) else [topics]
self.error_handler = error_handler
self._consumer = None
self._running = False

def _connect(self):
try:
if self._consumer is not None:
self._consumer.close()

self._consumer = Consumer(self.config)
self._consumer.subscribe(self.topics)
logger.info(f"Consumer подключен к Kafka и подписан на топики: {self.topics}")
return True
except KafkaException as e:
logger.error(f"Ошибка при создании Consumer: {e}")
return False

def process_batch(self, process_message_func, batch_size=100, timeout=1.0):
"""Обрабатывает батч сообщений с помощью переданной функции"""
if not self._connect():
return False

self._running = True

try:
while self._running:
try:
messages = self._consumer.consume(batch_size, timeout)
if not messages:
continue

processed_offsets = {} # Отслеживаем обработанные смещения по партициям

for msg in messages:
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
logger.debug(f"Достигнут конец партиции {msg.partition()}")
else:
error_msg = f"Ошибка при получении сообщения: {msg.error()}"
logger.error(error_msg)
if self.error_handler:
self.error_handler(error_msg)
else:
try:
# Обрабатываем сообщение
value_str = msg.value().decode('utf-8')
data = json.loads(value_str)
process_message_func(data, msg)

# Запоминаем позицию для коммита
topic_partition = f"{msg.topic()}{msg.partition()}"
processed_offsets[topic_partition] = (msg.topic(), msg.partition(), msg.offset())

except json.JSONDecodeError:
logger.error(f"Ошибка декодирования JSON: {value_str[:100]}")
except Exception as e:
logger.error(f"Ошибка обработки сообщения: {e}")
if self.error_handler:
self.error_handler(f"Ошибка обработки сообщения: {e}")

# Коммитим обработанные смещения
if processed_offsets:
try:
for _, (topic, partition, offset) in processed_offsets.items():
self._consumer.commit(message=msg, asynchronous=False)
except KafkaException as e:
logger.error(f"Ошибка при коммите смещений: {e}")

except KafkaException as e:
logger.error(f"Ошибка Kafka: {e}")
# Пробуем переподключиться
time.sleep(1)
self._connect()

finally:
self.close()

def close(self):
self._running = False
if self._consumer:
try:
self._consumer.close()
except:
pass

from prometheus_client import Counter, Gauge, start_http_server
import threading
import time

# Метрики Producer
producer_messages_total = Counter('kafka_producer_messages_total', 
'Total number of messages sent', 
['topic', 'status'])
producer_message_size = Counter('kafka_producer_message_size_bytes', 
'Size of messages sent in bytes', 
['topic'])
producer_batch_size = Gauge('kafka_producer_batch_size', 
'Current batch size in messages', 
['topic'])
producer_latency = Gauge('kafka_producer_latency_ms', 
'Producer latency in ms', 
['topic'])

# Метрики Consumer
consumer_messages_total = Counter('kafka_consumer_messages_total', 
'Total number of messages consumed', 
['topic', 'partition'])
consumer_message_size = Counter('kafka_consumer_message_size_bytes', 
'Size of messages consumed in bytes', 
['topic'])
consumer_lag = Gauge('kafka_consumer_lag', 
'Consumer lag in messages', 
['topic', 'partition'])
consumer_processing_time = Gauge('kafka_consumer_processing_time_ms', 
'Message processing time in ms', 
['topic'])

# Запуск сервера метрик Prometheus
def start_metrics_server(port=8000):
start_http_server(port)
print(f"Metrics server started on port {port}")

# Запуск в отдельном потоке
metrics_thread = threading.Thread(target=start_metrics_server)
metrics_thread.daemon = True
metrics_thread.start()

# Примеры использования в коде Producer
def send_with_metrics(producer, topic, value, key=None):
start_time = time.time()
size = len(json.dumps(value))

producer_batch_size.labels(topic=topic).set(producer._producer.flush(0))

try:
producer.send(topic, value, key)
producer_messages_total.labels(topic=topic, status='success').inc()
except Exception as e:
producer_messages_total.labels(topic=topic, status='error').inc()
raise

producer_message_size.labels(topic=topic).inc(size)
producer_latency.labels(topic=topic).set((time.time() – start_time) * 1000)

# Примеры использования в коде Consumer
def process_with_metrics(consumer, message):
start_time = time.time()
topic = message.topic()
partition = message.partition()
size = len(message.value())

consumer_messages_total.labels(topic=topic, partition=partition).inc()
consumer_message_size.labels(topic=topic).inc(size)

# Обработка сообщения...

consumer_processing_time.labels(topic=topic).set((time.time() – start_time) * 1000)

# Обновление лага (требует дополнительных запросов к Kafka)
# consumer_lag.labels(topic=topic, partition=partition).set(get_consumer_lag())