Событийно-ориентированные приложения с использованием Kafka и Python

В этой статье мы разработаем и реализуем событийно-ориентированное приложение с использованием Kafka в Python. Для примера мы возьмем заказ мебели в приложении типа IKEA. Это просто пример, а не то, что происходит на самом деле в IKEA.

Мы будем делать приложение на локальном компьютере, но для производственных сред вы можете использовать облачный провайдер, такой как AWS, GCP или Azure.

Давайте посмотрим, что у нас есть в приведенной выше архитектуре:

— Frontend (внешний интерфейс). Это может быть мобильное или веб-приложение, где пользователь заказывает товар. Когда пользователь выбирает и заказывает мебель с помощью приложения, внешний интерфейс обращается к бэкэнду.

— Orders Backend (бэкэнд заказов). Он принимает заказ из внешнего интерфейса со всеми данными, связанными с этим заказом, а затем записывает в Kafka тему под названием «order_details». Тема «order_details» будет содержать всю информацию, относящуюся к одному отдельному заказу. Это будет простой Python-файл. Вы можете задеплоить этот сервис вместе с остальными, как микросервисы в облако, используя, например, облачный запуск на GCP или Lambda на AWS.

— Transactions Backend (бэкэнд транзакций). Подписывается на тему «order_details» в Kafka, поэтому всякий раз, когда кто-то пишет в тему, бэкэнд транзакций будет читать сообщение и обрабатывать его в режиме реального времени. Бэкэнд транзакций будет выполнять обработку кредитных карт и некоторые другие проверки, чтобы убедиться, что заказ подтвержден. Как только заказ будет подтвержден, он отправит ответ в другую тему Kafka под названием «order_confirmed». Эта тема нужна, чтобы собирать все данные, которые относятся к подтвержденному заказу.

— Email Backend (бэкэнд электронной почты). Подписывается на тему «order_confirmed» и отправляет пользователю электронное письмо с подтверждением, когда заказ подтвержден. Он также может отправить сообщение в тему, например «order_email_sent».

— Analytics Backend (бэкэнд аналитики). Он подписывается на тему «order_confirmed» и выполняет по ней какую-то аналитику. Например, он может агрегировать общее количество заказов в этот день и общее количество доходов, полученных от разных заказов. Затем мы можем отправить результат аналитики по теме «analytics_result».

— Dashboard. У нас может быть служба для получения каких-либо данных из разных тем и отправки их на панель инструментов для визуализации. Здесь мы просто используем один сервис для них обоих в Python для простоты, но вы можете легко их разделить.

В этом посте нам понадобятся следующее:

kafka-python flask

Чтобы запустить Kafka локально, можно использовать следующий компоновочный файл с кластером Kafka с одним брокером, а также одним zookeeper’ом и некоторыми другими компонентами Kafka, такими как центр управления UI (UI control-center), реестр схем (schema-registry) и т.д.

## docker-compose-kafka.yml  version: "3"  services:   zookeeper:     image: confluentinc/cp-zookeeper:5.4.0     hostname: zookeeper     container_name: zookeeper     ports:       - "2181:2181"     environment:       ZOOKEEPER_CLIENT_PORT: 2181       ZOOKEEPER_TICK_TIME: 2000    broker:     image: confluentinc/cp-server:5.4.0     hostname: broker     container_name: broker     depends_on:       - zookeeper     ports:       - "9092:9092"       - "29093:29093"     environment:       KAFKA_BROKER_ID: 1       KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT       KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092       KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1       KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0       KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1       CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092       CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper:2181       CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1       CONFLUENT_METRICS_ENABLE: "true"       CONFLUENT_SUPPORT_CUSTOMER_ID: "anonymous"    kafka-tools:     image: confluentinc/cp-kafka:5.4.0     hostname: kafka-tools     container_name: kafka-tools     command: ["tail", "-f", "/dev/null"]     network_mode: "host"    schema-registry:     image: confluentinc/cp-schema-registry:5.4.0     hostname: schema-registry     container_name: schema-registry     depends_on:       - zookeeper       - broker     ports:       - "8081:8081"     environment:       SCHEMA_REGISTRY_HOST_NAME: schema-registry       SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: "zookeeper:2181"    control-center:     image: confluentinc/cp-enterprise-control-center:5.4.0     hostname: control-center     container_name: control-center     depends_on:       - zookeeper       - broker       - schema-registry     ports:       - "9021:9021"     environment:       CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'       CONTROL_CENTER_ZOOKEEPER_CONNECT: 'zookeeper:2181'       CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"       CONTROL_CENTER_REPLICATION_FACTOR: 1       CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1       CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1       CONFLUENT_METRICS_TOPIC_REPLICATION: 1       PORT: 9021

Также можно использовать Kafka-UI и Conduktor. UI-центр управления просто показывает сообщения, которые отправляются, когда UI-страница темы открыта. Для Kafka-UI вы можете добавить следующие коды в свой файл компоновки вместо части центра управления:

kafka-ui: image: provectuslabs/kafka-ui container_name: kafka-ui ports: - "8080:8080" restart: always environment: - KAFKA_CLUSTERS_0_NAME=local - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=broker:29092 - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2181 - KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL=PLAINTEXT - KAFKA_CLUSTERS_0_SCHEMAREGISTRY=http://schema-registry:8081

Затем нам нужно запустить docker-compose -f docker-compose-kafka.yml up -d, чтобы запустить Kafka со всеми компонентами. Обратите внимание, что Kafka должна быть запущена, когда мы хотим протестировать внешние и внутренние службы, которым необходимо отправлять или получать данные из тем Kafka.

Мы можем проверить, все ли работает, с помощью docker-compose -f docker-compose-kafka.yml ps команды:

NAME COMMAND SERVICE STATUS PORTS broker "/etc/confluent/dock…" broker running 0.0.0.0:9092->9092/tcp, 0.0.0.0:29093->29093/tcp kafka-tools "tail -f /dev/null" kafka-tools running kafka-ui "/bin/sh -c 'java $J…" kafka-ui running 0.0.0.0:8080->8080/tcp schema-registry "/etc/confluent/dock…" schema-registry running 0.0.0.0:8081->8081/tcp zookeeper "/etc/confluent/dock…" zookeeper running 2888/tcp, 0.0.0.0:2181->2181/tcp, 3888/tcp

В дополнение к вышеупомянутому файлу docker-compose для Kafka у нас будет еще один для наших микросервисов. Вы можете добавлять сервисы один за другим в файл docker-compose и тестировать их.

Обратите внимание, что нужно для подключения ваших producers и consumers к брокеру Kafka: 

• Если вы используете ту же сеть, поместив микросервис в тот же файл docker-compose, что и Kafka. Или используя отдельный файл docker-compose и настроив сеть, как сеть Kafka, вы можете использовать broker:29092.

• Если вы запускаете свой сервис локально на том же компьютере без его докеризации, вы можете использовать в своем коде его localhost:9092.

• Если вы хотите запустить Kafka на одной машине, а свои службы на другой машине, вам нужно использовать в своем коде <kafka machine ip>:29093.

Начнем с backend-сервисов. Интерфейс мы добавим позже.

Бэкэнд заказов

Теперь давайте перейдем к backend’у заказов. Мы докеризуем приложение и помещаем его в отдельный файл docker-compose с именем docker-compose-services.yml и настраиваем сеть так же, как сеть Kafka.  orders_backend.py — это flask-приложение и выглядит оно следующим образом:

# orders_backend.py  import json import time  from kafka import KafkaProducer from flask import Flask, jsonify, request  ORDER_KAFKA_TOPIC = 'order_details' # KAFKA_SERVER_ADDRESS = 'localhost:9092' KAFKA_SERVER_ADDRESS = 'broker:29092' # KAFKA_SERVER_ADDRESS = '47.93.191.241:29093`  app = Flask(__name__)  ## from inside docker compose network - when add the service to compose file -> orders_backend:v1 producer = KafkaProducer(bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                               value_serializer=lambda x: json.dumps(x).encode('utf-8'))  # post endpoint to get user id , order id, user email, and order details @app.route('/order', methods=['POST']) def order():     user_id = request.json['user_id']     order_id = request.json['order_id']     user_email = request.json['user_email']     order_details = request.json['order_details']     order = {}     order['user_id'] = user_id     order['order_id'] = order_id     order['user_email'] = user_email     order['order_details'] = order_details     order['time'] = time.time()     producer.send(ORDER_KAFKA_TOPIC, order)     print("Sent order details {} to kafka topic: {}".format(order, ORDER_KAFKA_TOPIC))     return jsonify(order)   if __name__ == '__main__':     app.run(host="0.0.0.0", port=5002, debug=True)

У него есть конечная точка публикации для получения заказа и публикации его в теме Kafka с именем «order_details».

Затем вы можете легко докеризовать этот сервис. Вот докерфайл:

FROM python:3.9.7-slim RUN pip install -U pip RUN pip install pipenv WORKDIR /app COPY [ "Pipfile", "Pipfile.lock", "./" ] RUN pipenv install - system - deploy COPY [ "orders_backend.py", "./" ] EXPOSE 5002 ENTRYPOINT ["python", "orders_backend.py"]

Затем мы можем создать образ, используя:

docker build -t orders_backend:v1 .

Файл docker-compose будет выглядеть следующим образом:

# docker-compose-services.yml  version: "1"  services:   orders_backend:     restart: always     image: orders_backend:v1     ports:       - "5002:5002"     networks:       - ikea-ordering-kafka_default  networks:   ikea-ordering-kafka_default:     external: true

Мы можем использовать postman для проверки:

Мы также можем видеть сообщения по теме в пользовательском интерфейсе:

Бэкэнд транзакций

Это простой сервис для прослушивания темы «order_details», выполнения какой-либо обработки данных и отправки подтвержденного сообщения в тему «order_confirmed»:

# transactions_backend.py  import json import time from kafka import KafkaConsumer, KafkaProducer  OERDER_KAFKA_TOPIC = 'order_details' ORDER_CONFIRMED_KAFKA_TOPIC = 'order_confirmed' # KAFKA_SERVER_ADDRESS = 'localhost:9092' KAFKA_SERVER_ADDRESS = 'broker:29092' # KAFKA_SERVER_ADDRESS = '47.93.191.241:29093`  consumer = KafkaConsumer(OERDER_KAFKA_TOPIC, bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                             value_deserializer=lambda x: json.loads(x.decode('utf-8'))) producer = KafkaProducer(bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                             value_serializer=lambda x: json.dumps(x).encode('utf-8'))  while True:     for message in consumer:         print("Received order details: {}".format(message.value))         user_id = message.value['user_id']         order_id = message.value['order_id']         user_email = message.value['user_email']         order_details = message.value['order_details']         time = message.value['time']         ## do some suff on the order and check the confirmation         order_confirmed = {}         order_confirmed['user_id'] = user_id         order_confirmed['order_id'] = order_id         order_confirmed['user_email'] = user_email         order_confirmed['order_details'] = order_details         order_confirmed['time'] = time         order_confirmed['status'] = 'confirmed'         producer.send(ORDER_CONFIRMED_KAFKA_TOPIC, order_confirmed)         print("Sent order details {} to kafka topic: {}".format(order_confirmed, ORDER_CONFIRMED_KAFKA_TOPIC))

Докер-файл выглядит следующим образом:

FROM python:3.9.7-slim RUN pip install -U pip RUN pip install pipenv WORKDIR /app COPY [ "Pipfile", "Pipfile.lock", "./" ] RUN pipenv install - system - deploy COPY [ "transactions_backend.py", "./" ] ENTRYPOINT ["python", "transactions_backend.py"]

Вы можете создать этот образ и обновить docker-compose-services.yml файл:

# docker-compose-services.yml  version: "1"  services:   orders_backend:     restart: always     image: orders_backend:v1     ports:       - "5002:5002"     networks:       - ikea-ordering-kafka_default      transactions_backend:     restart: always     image: transactions_backend:v1     ports:       - "5003:5003"     networks:       - ikea-ordering-kafka_default  networks:   ikea-ordering-kafka_default:     external: true

Затем, ещё раз протестировав сервисы с помощью postman, мы можем увидеть сообщения, приходящие в тему:

Бэкэнд электронной почты

Код этой службы выглядит следующим образом:

# email_backend.py  import json import time  from kafka import KafkaConsumer, KafkaProducer # from flask import Flask, jsonify, request  ORDER_CONFIRMED_KAFKA_TOPIC = 'order_confirmed' EMAIL_SENT_KAFKA_TOPIC = 'order_email_sent' # KAFKA_SERVER_ADDRESS = 'localhost:9092' KAFKA_SERVER_ADDRESS = 'broker:29092' # KAFKA_SERVER_ADDRESS = '47.93.191.241:29093`  producer = KafkaProducer(bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                             value_serializer=lambda x: json.dumps(x).encode('utf-8'))  consumer = KafkaConsumer(ORDER_CONFIRMED_KAFKA_TOPIC, bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                             value_deserializer=lambda x: json.loads(x.decode('utf-8')))   def send_email(user_id, order_id, user_email, order_details, time, status):     print("Sending email to user: {} with order details: {}".format(user_email, order_details))     # send email to user     # ...     # ...     # ...     # ...        return True   while True:     for message in consumer:         # read data from consumer and call the send_email() function         print("Received order details: {}".format(message.value))         user_id = message.value['user_id']         order_id = message.value['order_id']         user_email = message.value['user_email']         order_details = message.value['order_details']         time = message.value['time']         status = message.value['status']         email_send_status = send_email(user_id, order_id, user_email, order_details, time, status)         email_sent = {}         email_sent['user_id'] = user_id         email_sent['order_id'] = order_id         email_sent['user_email'] = user_email         email_sent['order_details'] = order_details         email_sent['time'] = time         email_sent['status'] = email_send_status         producer.send(EMAIL_SENT_KAFKA_TOPIC, email_sent)         print("Sent email details {} to kafka topic: {}".format(email_sent, EMAIL_SENT_KAFKA_TOPIC))

Докер-файл также похож на предыдущие с небольшими изменениями имени python-файла. Затем вы можете создать образ, обновить файл docker-compose и запустить его. После отправки нескольких новых сообщений через postman мы можем увидеть сообщения в теме в пользовательском интерфейсе:

Бэкэнд аналитики

Код следующего сервиса получает подтвержденный заказ и рассчитывает общее количество заказов и общий доход:

# analytics_backend.py  import json import time  from kafka import KafkaConsumer, KafkaProducer  ORDER_CONFIRMED_KAFKA_TOPIC = 'order_confirmed' ANALYTICS_KAFKA_TOPIC = 'analytics_result' # KAFKA_SERVER_ADDRESS = 'localhost:9092' KAFKA_SERVER_ADDRESS = 'broker:29092' # KAFKA_SERVER_ADDRESS = '47.93.191.241:29093`  producer = KafkaProducer(bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                             value_serializer=lambda x: json.dumps(x).encode('utf-8'))  consumer = KafkaConsumer(ORDER_CONFIRMED_KAFKA_TOPIC, bootstrap_servers=[KAFKA_SERVER_ADDRESS], security_protocol="PLAINTEXT",                             value_deserializer=lambda x: json.loads(x.decode('utf-8')))  total_revenue = 0 total_orders_count = 0  while True:     for message in consumer:         # read data from consumer and do some analytics on it         print("Received order details: {}".format(message.value))         order_details = message.value['order_details']         total_revenue += int(order_details['price'])         total_orders_count += 1         analytics = {}         analytics['total_revenue'] = total_revenue         analytics['total_orders_count'] = total_orders_count         producer.send(ANALYTICS_KAFKA_TOPIC, analytics)         print("Sent analytics details {} to kafka topic: {}".format(analytics, ANALYTICS_KAFKA_TOPIC))

Докер-файл снова точно такой же, как и предыдущий, но с небольшой модификацией. Создайте образ, обновите файл компоновки и, наконец, запустите его.

Мы можем отправить несколько новых сообщений через postman и увидеть сообщения в теме в пользовательском интерфейсе:

Вот и все на этот раз. Надеемся, вы получили общее представление о том, как использовать Kafka в своих проектах.

А чтобы еще больше узнать о том, как разработчикам можно использовать Kafka в работе, вы можете прийти на наш курс «Apache Kafka для разработчиков». Это углублённый курс с практикой на Java или Golang и платформой Spring+Docker+Postgres, который переведёт вас на новый уровень владения инструментом.