В этой статье я хочу показать как можно использовать Kafka в дата-инженерии и как её "пощупать".

Я не хотел бы повторяться по важным моментам, которые касаются архитектуры Kafka, поэтому рекомендую ознакомиться с данным видео.

В нём хорошо рассказано про основные концепции, которые будут дальше использоваться в статье, такие как:

Что такое producer.
Что такое consumer.
Что такое topic.
Что такое offset.
Что такое commit.
Что такое partition .
Что такое replication .

Весь код, который будет использоваться в статье будет доступен в моём репозитории.

Разворачивание сервиса

Начнём с того, что развернем Kafka локально в Docker. Для этого создадим docker-compose.yaml со следующим кодом:

version: '3.8'  
  
services:  
  zookeeper:  
    image: 'confluentinc/cp-zookeeper:7.7.0'  
    hostname: zookeeper  
    container_name: zookeeper  
    environment:  
      ZOOKEEPER_CLIENT_PORT: 2181  
      ZOOKEEPER_TICK_TIME: 2000  
    ports:  
      - '2181:2181'  
  
  kafka:  
    image: 'confluentinc/cp-kafka:7.7.0'  
    hostname: kafka  
    container_name: kafka  
    depends_on:  
      - zookeeper  
    environment:  
      KAFKA_BROKER_ID: 1  
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181  
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092  
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:19092  
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT  
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT  
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1  
    ports:  
      - '9092:9092'  
      - '19092:19092'  
  
  kafka-ui:  
    image: 'provectuslabs/kafka-ui:v0.7.2'  
    container_name: kafka-ui  
    ports:  
      - '8080:8080'  
    environment:  
      KAFKA_CLUSTERS_0_NAME: local  
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092  
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181  
    depends_on:  
      - kafka  
  
networks:  
  default:  
    name: kafka-network

Чтобы запустить все сервисы выполним команду:

docker-compose up -d

После этого у нас запустится Kafka, ZooKeeper и UI for Apache Kafka.

UI for Apache Kafka будет доступен по адресу http://localhost:8080/ через него можно будет: создавать topic, удалять topic, смотреть сообщения в topic и прочее. Очень удобный инструмент для работы с Kafka.

Создание и удаление topic

В данном разделе мы с вами попробуем создавать и удалять topic.

Создание и удаление topic через CLI

Чтобы создать topic нужно выполнить команды ниже.

Зайти в контейнер с Kafka:

docker exec -it kafka /bin/bash

Создание topic test в Kafka:

kafka-topics --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

Просмотр всех доступных topic в Kafka:

kafka-topics --list --bootstrap-server kafka:9092

Удаление topic test в Kafka:

kafka-topics --delete --topic test --bootstrap-server kafka:9092

Создание и удаление topic через Python

Если вам удобнее взаимодействовать с Kafka через Python, то это не проблема.

Для работы с Kafka нам понадобится библиотека confluent-kafka. В примерах ниже я использую версию 2.5.0. Весь код и список всех зависимостей находится в моём репозитории.

Точно также эти операции можно произвести без подключения к контейнеру c Kafka, а через Python.

Чтобы создать topic через Kafka:

from confluent_kafka.admin import AdminClient, NewTopic  
  
admin_client = AdminClient({'bootstrap.servers': 'localhost:19092'})  
  
  
def example_create_topics(a: AdminClient = None, topics: list[str] = None) -> None:  
    """  
    Функция для создани�� `topic` в Kafka  
    :param a: AdminClient с параметрами инициализации. Default `None`.    :param topics: Список `topic` для создания. Default `None`.    :return: Ничего не возвращает  
    """  
    new_topics = [NewTopic(topic, num_partitions=1, replication_factor=1) for topic in topics]  
        try:  
            f.result()  # The result itself is None  
            print("Topic {} created".format(topic))  
        except Exception as e:  
            print("Failed to create topic {}: {}".format(topic, e))  
  
  
example_create_topics(  
    a=admin_client,  
    topics=['test'],  
)

Важно: IDE может ругаться, что модуля NewTopic не существует, но он есть. Это официальный пакет. Это касается версии 2.5.0.

Чтобы удалить topic:

from confluent_kafka.admin import AdminClient  
  
admin_client = AdminClient({'bootstrap.servers': 'localhost:19092'})  
  
  
def example_delete_topics(a: AdminClient = None, topics: list[str] = None) -> None:  
    """  
    Функция для удаления `topic` в Kafka.  
    :param a: AdminClient с параметрами инициализации. Default `None`.    :param topics: Список `topic` для удаления. Default `None`.    :return: Ничего не возвращает.  
    """  
    fs = a.delete_topics(topics, operation_timeout=30)  
  
    # Wait for operation to finish.  
    for topic, f in fs.items():  
        try:  
            f.result()  # The result itself is None  
            print("Topic {} deleted".format(topic))  
        except Exception as e:  
            print("Failed to delete topic {}: {}".format(topic, e))  
  
  
example_delete_topics(  
    a=admin_client,  
    topics=['test'],  
)

Больше примеров использования библиотеки confluent_kafka в официальном GitHub проекта.

Kafka CLI

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

wget https://archive.apache.org/dist/kafka/3.8.0/kafka_2.13-3.8.0.tgz

Затем распаковать:

tar -xzf kafka_2.13-3.8.0.tgz

После выполнения данных команд мы можем использовать CLI для взаимодействия с Kafka.

Важно: Все исполняемые файлы находятся в папке bin. Поэтому стоит обратить внимание, что все скрипты будут выполнять из неё.

Чтобы перейти в папку bin нужно выполнить команду:

cd kafka_2.13-3.8.0/bin/

Запись в Kafka через CLI

Чтобы произвести запись в Kafka выполним команду:

echo 'Hello, Kafka!' | sh kafka-console-producer.sh --broker-list localhost:19092 --topic test

Или так:

echo 'Hello, Kafka!' | ./kafka-console-producer.sh --broker-list localhost:19092 --topic test

Важно: Мне привычнее вызывать скрипт командой sh, но можно и через ./.

Ещё можно создать producer в интерактивном режиме командой:

sh kafka-console-producer.sh --broker-list localhost:19092 --topic test

После создания такого producer у нас появляется возможность писать все сообщения, которые хотим.

После выполнения команды у нас появится [> и после чего мы сможем вводить сообщения для Kafka.

Для выхода из интерактивного режима несколько раз нажмите CTRL + C.

Чтение из Kafka через CLI

Важно: topic в Kafka можно читать "с конца" и "с начала".

Чтобы начать читать с самого начала:

sh kafka-console-consumer.sh --bootstrap-server localhost:19092 --topic test --from-beginning

Чтобы начать читать с конца и получать только новые сообщения:

sh kafka-console-consumer.sh --bootstrap-server localhost:19092 --topic test

Kafka Python

Как было описано выше мы можем взаимодействовать с Kafka через Python. Поэтому сейчас рассмотрим также операции записи и чтения с использованием Python.

Запись в Kafka через Python

Я приведу пример той записи, которая может появиться в вашей Kafka – это информация о пользователе.
Запись будет содержать: uuid, first_name, last_name, middle_name.

Вы можете запустить код ниже и в topic my_topic начнут записываться значения.

import json  
import time  
from confluent_kafka import Producer  
from faker import Faker  
import uuid_utils as uuid  
  
  
def generate_list_of_dict() -> dict[str, str]:  
  
    fake = Faker(locale='ru_RU')  
  
    return {  
        'uuid': str(uuid.uuid7()),  
        'first_name': fake.first_name(),  
        'last_name': fake.last_name(),  
        'middle_name': fake.middle_name(),  
    }  
  
  
# Define the Kafka configuration  
conf = {'bootstrap.servers': "localhost:19092"}  
  
# Create a Producer instance with the above configuration  
producer = Producer(conf)  
  
  
while True:  
    # Define some data to send to Kafka  
    data = generate_list_of_dict()  
  
    # Convert the data to a JSON string  
    data_str = json.dumps(data)  
  
    # Produce a message to the "my_topic" topic  
    producer.produce(topic="my_topic", value=data_str)  
  
    # Flush the producer to ensure all messages are sent  
    producer.flush()  
  
    # Sleep for a second before producing the next set of messages  
    time.sleep(3)

Важно: Если topic ранее не был создан, то он создастся при первой записи.

Чтение из Kafka через Python

Для того чтобы прочитать значения из Kafka нам необходимо создать consumer. Функция ниже имеет возможность прочитать topic с самого начала и с определённого offset.

from confluent_kafka import Consumer, KafkaError, TopicPartition  
  
  
def consume_messages(topic: str = None, offset: int = None) -> None:  
    conf = {  
        'bootstrap.servers': 'localhost:19092',  
        'group.id': 'mygroup',  
        'auto.offset.reset': 'earliest'  
    }  
  
    consumer = Consumer(conf)  
  
    if offset is not None:  
        partitions = consumer.list_topics(topic).topics[topic].partitions  
        for partition in partitions:  
            consumer.assign([TopicPartition(topic, partition, offset)])  
    else:  
        consumer.subscribe([topic])  
  
    try:  
        while True:  
            msg = consumer.poll(1.0)  
            if msg is None:  
                continue  
            if msg.error():  
                if msg.error().code() == KafkaError:  
                    print('Reached end of partition')  
                else:  
                    print(f'Error: {msg.error()}')  
            else:  
                print(f'Received message: {msg.value().decode("utf-8")}')  
    except KeyboardInterrupt:  
        pass  
    finally:  
        consumer.close()  
  
  
# Читать с начала  
consume_messages('test')  
  
# Читать с определенного offset  
# consume_messages('test', offset=5)

Ранее мы читали topic в Kafka без использования групп и поэтому атрибут --from-beginning срабатывал каждый раз при вызове (каждый раз создавалась новая группа).

Но при создании consumer через Python указание group.id является обязательным и поэтому мы можем столкнуться со следующей проблемой: если мы один раз прочитали topic, то при перезапуске кода мы начнем читать только новые сообщения и даже атрибут auto.offset.reset не поможет.

А всё это происходит, потому что мы произвели commit (фиксацию) offset для группы.

Чтобы проверить на каком сейчас offset находится группа необходимо выполнить команду в Kafka:

sh kafka-consumer-groups.sh --bootstrap-server localhost:19092 --group mygroup --describe

И мы увидим, что мы прочитали все сообщения. Поэтому offset стоит на последнем сообщении в topic.

CURRENT-OFFSET говорит о том на каком offset находится группа.
LOG-END-OFFSET текущий последний доступный offset для topic

Вообще, это не проблема, потому что данный offset можно "сбросить", для этого необходимо выполнить команду:

sh kafka-consumer-groups.sh --bootstrap-server localhost:19092 --group mygroup --to-earliest --reset-offsets --execute --topic test

Также можно прочитать topic заново изменив group.id. Но это делать не рекомендуется.

Использование Kafka в дата-инженерии

В дата-инженерии Kafka частый гость, потому что Kafka позволяет быстро и за дёшево покрыть множество бизнес-задач, таких как:

CDC

При реализации CDC вы можете встретиться с Kafka, потому что она является "стандартом" при работе с такого вида событиями.

Если вы хотите понять что такое CDC и какую роль там занимает Kafka вы можете изучить мою статью: CDC на примитивах.

Event-driven

Так как Kafka позволяет нам получать изменения "моментально". В этом определении есть определённые нюансы, но это тема для другого разговора.

Если вернуться к мысли выше, то получая все события "моментально" мы можем на них реагировать.

Для примера: покупатель заходит на сайт нашего интернет-магазина и при заходе в какую-то категорию или раздел мы можем сделать ему какое-то предложение или перестроить для него страницу, в зависимости от его предпочтений или условий заложенных ранее.

Real-time Analytics

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

Для примера: маркетинговые акции. Мы запускаем какую-то акцию и сразу смотрим на важные для нас показатели. В зависимости от получаемых значений мы можем изменять условия акции, условия размещения и прочее.

Резюме

Kafka популярный инструмент, поэтому найти литературу, видео и примеры использования – не проблема. В данной статье я показал только верхушку айсберга, который можно изучать и изучать.

Если говорить про взаимодействие c Kafka, то CLI и Python – это не единственные инструменты, к ним можно добавить: PySpark, ClickHouse, Java и прочее.

Кстати, про то как читать из Kafka при помощи ClickHouse было описано в моей статье: CDC на примитивах.

Для более глубокого изучения инструмента рекомендую ознакомиться с книгой: Apache Kafka. Потоковая обработка и анализ данных" (авторы - Нархид Н., Шапира Г., Палино Т., год издания - 2019). В ней описывается много тонкостей и подводных камней при работе с Kafka. Уже вышло второе издание, я его не читал, но судя по наполнению; учтены новые моменты, поэтому порекомендовал бы изучать второе издание.

Ну и самое главное – Теория без практики мертва, практика без теории слепа. Поэтому попробуйте Kafka, даже на pet-проектах или в рамках данной статьи.

Также если вам необходима консультация/менторство/мок-собеседование и другие вопросы по дата-инженерии, то вы можете обращаться ко мне. Все контакты указаны по ссылке.

Инфраструктура для data engineer Kafka