Что такое Kafka?

Что такое Kafka

Apache Kafka — это распределённая платформа потоковой обработки данных (streaming platform), которая используется для построения real-time приложений. Она позволяет отправлять, хранить и обрабатывать большие потоки данных с низкой задержкой.

Основные характеристики

Kafka — это:

• Брокер сообщений (message broker)
• Система потоковой обработки (stream processing)
• Платформа для архитектуры на основе событий (event-driven architecture)

Kafka обрабатывает миллиарды сообщений в день в компаниях: LinkedIn, Netflix, Airbnb, Uber, Twitter.

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

┌─────────────┐     ┌─────────────┐
│ Producer 1  │     │ Producer 2  │
└──────┬──────┘     └──────┬──────┘
       │                   │
       │  publish          │
       └───────────┬───────┘
                   ↓
        ┌──────────────────────┐
        │   Kafka Broker       │
        │  (Topic = queue)     │
        │  Partition 0,1,2...  │
        └──────────────────────┘
                   ↑
       ┌───────────┴───────────┐
       │ consume               │
 n      │
┌──────────────┐      ┌──────────────┐
│ Consumer 1   │      │ Consumer 2   │
└──────────────┘      └──────────────┘

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

Topic — очередь сообщений, логическая единица публикации-подписки:

# Пример: Topic "orders" содержит все сообщения о заказах
topic_name = "orders"  # Topic

Partition — раздел топика для параллельной обработки:

Topic: orders
├─ Partition 0: msg1, msg2, msg3
├─ Partition 1: msg4, msg5, msg6
└─ Partition 2: msg7, msg8, msg9

Каждая partition может быть обработана отдельным consumer

Producer — отправитель сообщений:

from kafka import KafkaProducer
import json

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

# Отправка сообщения
order = {"order_id": 123, "customer": "John", "total": 99.99}
producer.send("orders", value=order)
producer.flush()

Consumer — получатель сообщений:

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=["localhost:9092"],
    value_deserializer=lambda m: json.loads(m.decode("utf-8"))
)

# Получение сообщений
for message in consumer:
    print(f"Заказ: {message.value}")
    # Обработка

Consumer Group — группа consumers, которые вместе обрабатывают топик:

from kafka import KafkaConsumer

# Consumer Group: "order-processors"
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=["localhost:9092"],
    group_id="order-processors",  # Это consumer group
    auto_offset_reset="earliest"
)

for message in consumer:
    print(f"Consumer обработал: {message.value}")

Если запустить несколько consumers с одним group_id, Kafka автоматически распределит partitions между ними.

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

Kafka предоставляет три уровня гарантий (acks):

producer = KafkaProducer(
    bootstrap_servers=["localhost:9092"],
    acks="all",  # Гарантии:
    # acks=0 (fire-and-forget): нет гарантий
    # acks=1 (leader ack): лидер получил
    # acks="all" (all in-sync replicas): все replicas получили
)

Offsets и хранение истории

Kafka хранит сообщения и позволяет consumers читать историю:

consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=["localhost:9092"],
    auto_offset_reset="earliest",  # Начать с первого сообщения
    # auto_offset_reset="latest"  # Начать с последнего
)

for message in consumer:
    print(f"Partition: {message.partition}")
    print(f"Offset: {message.offset}")  # Номер в partition
    print(f"Value: {message.value}")

Использование Kafka с Python

1. Установка:

pip install kafka-python

2. Простой producer-consumer:

# producer.py
from kafka import KafkaProducer
import time
import json

producer = KafkaProducer(
    bootstrap_servers=["localhost:9092"],
    value_serializer=lambda v: json.dumps(v).encode()
)

for i in range(10):
    message = {"event": "click", "user_id": i}
    producer.send("events", value=message)
    time.sleep(1)

producer.close()

# consumer.py
from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "events",
    bootstrap_servers=["localhost:9092"],
    group_id="my-group",
    value_deserializer=lambda m: json.loads(m.decode()),
    auto_offset_reset="earliest"
)

for message in consumer:
    print(f"Got message: {message.value}")

Сравнение с RabbitMQ и Redis

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

1. Обработка логов в реальном времени:

# Producer отправляет логи
producer.send("logs", value={
    "timestamp": "2024-01-01 10:00:00",
    "level": "ERROR",
    "message": "Database connection failed"
})

# Multiple consumers обрабатывают: алертинг, архивация, индексация

2. Event sourcing (архитектура событий):

# Все изменения в системе - события в Kafka
# User created
producer.send("user-events", value={"event_type": "created", "user_id": 1})

# User updated
producer.send("user-events", value={"event_type": "updated", "user_id": 1})

# User deleted
producer.send("user-events", value={"event_type": "deleted", "user_id": 1})

3. Микросервисы коммуникация:

Order Service -> Kafka -> Inventory Service
             -> Kafka -> Payment Service
             -> Kafka -> Notification Service

Преимущества Kafka

1. Высокая пропускная способность — миллионы сообщений в секунду
2. Хранение истории — переиграть события в любой момент
3. Масштабируемость — горизонтальное масштабирование через partitions
4. Надежность — репликация, гарантии доставки
5. Простота — publish-subscribe модель понятна всем

Недостатки

1. Сложность — требует настройки и понимания архитектуры
2. Потребление памяти — нужно много ресурсов для высокой нагрузки
3. Обработка упорядоченности — сложно гарантировать порядок без partitions

В заключение: Kafka — это решение для high-load систем, требующих обработки больших потоков данных с низкой задержкой и надежными гарантиями доставки.