Объясните принципы работы Apache Kafka. Когда его применять?

Apache Kafka: Распределённая система обмена сообщениями

Apache Kafka — это высокопроизводительная распределённая платформа потоковой обработки данных, которая позволяет публиковать, подписываться и обрабатывать потоки записей в реальном времени.

Основные компоненты

1. Producer - создатель сообщений
2. Broker - сервер для хранения сообщений
3. Consumer - получатель сообщений
4. Topic - логическая группировка сообщений
5. Partition - физическое распределение данных

Архитектура Kafka

Producers → [Topic1: P0 | P1 | P2] → Consumers
            [Topic2: P0 | P1]       ↓
            Broker Cluster          Consumer Groups

Ключевые концепции

Topics и Partitions:

Topic: user_events
├─ Partition 0: [msg1, msg2, msg3, ...]
├─ Partition 1: [msg4, msg5, msg6, ...]
└─ Partition 2: [msg7, msg8, msg9, ...]

Каждая partition хранит упорядоченную последовательность сообщений.

Offset - позиция сообщения в partition:

Partition 0:
Offset 0: {user_id: 1, action: login}
Offset 1: {user_id: 2, action: purchase}
Offset 2: {user_id: 1, action: logout}

Пример: Producer и Consumer

Kafka Producer (отправитель):

from kafka import KafkaProducer
import json
from datetime import datetime

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

# Отправка сообщения
event = {
    'user_id': 123,
    'action': 'purchase',
    'amount': 99.99,
    'timestamp': datetime.utcnow().isoformat()
}

future = producer.send('user_events', value=event)

# Ожидание подтверждения
record_metadata = future.get(timeout=10)
print(f"Сообщение отправлено в partition {record_metadata.partition}")

Kafka Consumer (получатель):

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'user_events',
    bootstrap_servers=['localhost:9092'],
    group_id='analytics_group',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value
    print(f"Обработка события: {event}")
    # Отправка в базу
    save_to_db(event)

Consumer Groups

Много consumers в одной группе распределяют партиции:

Topic: orders (3 partitions)
├─ P0 → Consumer1 (group1)
├─ P1 → Consumer2 (group1)  Каждый consumer обрабатывает
└─ P2 → Consumer3 (group1)  свою partition

Another Consumer:
   Consumer4 (group2) → читает все 3 partition

Гарантии доставки

At-most-once (максимум один раз):

producer = KafkaProducer(
    acks=0  # Не ждём подтверждения от broker
)
# Быстро, но может потеряться

At-least-once (минимум один раз):

producer = KafkaProducer(
    acks='all'  # Ждём подтверждения от всех replicas
)
# Медленнее, но надёжнее

Exactly-once (ровно один раз):

producer = KafkaProducer(
    acks='all',
    retries=3,
    enable_idempotence=True  # Идемпотентная доставка
)

Consumer offset management

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'user_events',
    bootstrap_servers=['localhost:9092'],
    group_id='analytics_group',
    auto_commit_interval_ms=1000,  # Сохранять offset каждую секунду
    enable_auto_commit=True
)

for message in consumer:
    process(message.value)
    # Offset автоматически сохраняется

Реальный пример: Stream обработка с Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window

spark = SparkSession.builder \
    .appName("KafkaStreamProcessor") \
    .getOrCreate()

# Читаем из Kafka
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "user_events") \
    .load()

# Парсим JSON
schema = "user_id INT, action STRING, amount FLOAT, timestamp TIMESTAMP"
df = df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

# Aggrегация: доход за 5 минут
window_agg = df \
    .filter(col("action") == "purchase") \
    .groupBy(window("timestamp", "5 minutes")) \
    .agg({"amount": "sum"}) \
    .withColumnRenamed("sum(amount)", "revenue")

# Записываем обратно в Kafka
query = window_agg \
    .select("*") \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "revenue_5min") \
    .option("checkpointLocation", "/tmp/checkpoint") \
    .start()

query.awaitTermination()

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

from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    batch_size=16384,          # Размер батча (16KB)
    linger_ms=10,              # Ждать 10ms перед отправкой
    compression_type='snappy', # Сжатие
    acks='all',                # Надёжность
    retries=3                  # Повторы
)

Когда применять Kafka

Используй Kafka, если:

• Нужна обработка потоков в реальном времени
• Высокий объём сообщений (миллионы/сек)
• Нужна отказоустойчивость
• Требуется история сообщений (replaying)
• Много consumer приложений
• Асинхронная обработка

Примеры использования:

• Логирование событий
• Real-time аналитика
• System monitoring
• Microservices коммуникация
• Event sourcing
• Data pipeline

Альтернативы

• RabbitMQ - для простых очередей
• AWS Kinesis - managed solution
• Google Pub/Sub - cloud вариант
• Redis Streams - для легковесных решений

Лучшие практики

1. Используй правильную гарантию доставки (exactly-once для важных данных)
2. Мониторь consumer lag
3. Масштабируй partition count по объёму
4. Используй schema registry (Avro/Protobuf)
5. Логируй проблемы с доставкой

Вывод: Apache Kafka — это мощный инструмент для потоковой обработки данных и построения event-driven архитектур. Используй его для real-time аналитики, логирования и асинхронной коммуникации между системами.