Инфраструктура Apache / BI
Apache Superset
Установка
Сборка образа
FROM apache/superset:latest
USER root
RUN pip install --no-cache-dir \
psycopg2-binary \
redis
USER supersetДополнительные драйвера:
| БД | Драйвер |
|---|---|
| MySQL | mysqlclient |
| MSSQL | pymssql |
| ClickHouse | clickhouse-connect |
| Oracle | cx_Oracle |
docker build -t test/superset6_1 .Но это не поехало. Нужно использовать инструкцию по сборке образа из git.
Делаем Dockerfile и добавляем нужные драйвера. В инструкции предлагается тег master однако с ним не поехало. Указал конкретную цифру версии.
# change this to apache/superset:5.0.0 or whatever version you want to build from;
# otherwise the default is the latest commit on GitHub master branch
FROM apache/superset:6.0.0
USER root
# Set environment variable for Playwright
ENV PLAYWRIGHT_BROWSERS_PATH=/usr/local/share/playwright-browsers
# Install packages using uv into the virtual environment
RUN . /app/.venv/bin/activate && \
uv pip install \
# install psycopg2 for using PostgreSQL metadata store - could be a MySQL package if using that backend:
psycopg2-binary \
# add the driver(s) for your data warehouse(s), in this example we're showing for Microsoft SQL Server:
#pymssql \
# package needed for using single-sign on authentication:
Authlib \
# openpyxl to be able to upload Excel files
openpyxl \
# Pillow for Alerts & Reports to generate PDFs of dashboards
Pillow \
# install Playwright for taking screenshots for Alerts & Reports. This assumes the feature flag PLAYWRIGHT_REPORTS_AND_THUMBNAILS is enabled
# That feature flag will default to True starting in 6.0.0
# Playwright works only with Chrome.
# If you are still using Selenium instead of Playwright, you would instead install here the selenium package and a headless browser & webdriver
playwright \
&& playwright install-deps \
&& PLAYWRIGHT_BROWSERS_PATH=/usr/local/share/playwright-browsers playwright install chromium
# Switch back to the superset user
USER superset
CMD ["/app/docker/entrypoints/run-server.sh"]
Собираем образ
docker build -t my-superset:6 .
Скачивается git проекта
git clone --depth=1 https://github.com/apache/superset.gitURI:
postgresql://testuser:testpass@pgdataout:5432/testdb
python -m pip install psycopg2-binary
Агент
import psycopg2
from psycopg2 import sql
def execute_query(query: str, params: tuple = None):
"""
Выполняет SQL-запрос к PostgreSQL и возвращает результат.
:param query: SQL-запрос
:param params: параметры запроса (для безопасной подстановки)
:return: результат fetchall() или None
"""
connection = None
try:
connection = psycopg2.connect(
host="localhost",
port=5432,
database="your_database",
user="your_user",
password="your_password"
)
with connection.cursor() as cursor:
cursor.execute(query, params)
# Если запрос возвращает данные (SELECT)
if cursor.description:
result = cursor.fetchall()
return result
# Если это INSERT / UPDATE / DELETE
connection.commit()
return None
except Exception as e:
print(f"Ошибка выполнения запроса: {e}")
if connection:
connection.rollback()
raise
finally:
if connection:
connection.close()
rows = execute_query(
"SELECT id, name FROM users WHERE age > %s",
(18,)
)
print(rows)
execute_query(
"INSERT INTO users (name, age) VALUES (%s, %s)",
("Ivan", 25)
)
HTML обертки
Обертка для прокручивания дашбордов
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Dashboard Rotation</title>
<style>
body { margin: 0; }
iframe {
width: 100vw;
height: 100vh;
border: none;
}
</style>
</head>
<body>
<iframe id="dashboardFrame"></iframe>
<script>
const dashboards = [
"http://192.168.1.196/superset/dashboard/1/?standalone=1",
"http://192.168.1.196/superset/dashboard/2/?standalone=1",
];
let index = 0;
const frame = document.getElementById("dashboardFrame");
function rotate() {
frame.src = dashboards[index];
index = (index + 1) % dashboards.length;
}
rotate();
setInterval(rotate, 15000); // 15 секунд
</script>
</body>
</html>
Обертка для переключения tab
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<title>Superset Dashboard Carousel</title>
<style>
body {
margin: 0;
background: #111;
}
iframe {
width: 100vw;
height: 100vh;
border: none;
}
</style>
</head>
<body>
<iframe
id="supersetFrame"
src="http://192.168.1.191:8088/superset/dashboard/p/kgA6aQpyVpb/?standalone=1"
></iframe>
<script>
const ROTATION_INTERVAL = 10000; // 10 секунд
function startTabCarousel() {
const iframe = document.getElementById("supersetFrame");
iframe.onload = function () {
const iframeDoc = iframe.contentWindow.document;
setInterval(() => {
const tabs = iframeDoc.querySelectorAll('[role="tab"]');
const activeTab = iframeDoc.querySelector('[role="tab"][aria-selected="true"]');
if (!tabs.length || !activeTab) return;
const tabArray = Array.from(tabs);
const currentIndex = tabArray.indexOf(activeTab);
const nextIndex = (currentIndex + 1) % tabArray.length;
tabArray[nextIndex].click();
}, ROTATION_INTERVAL);
};
}
startTabCarousel();
</script>
</body>
</html>
Директории
- mydata
- static
- superset_db_data
- superset_home
docker-compose.yml
services:
superset:
image: amancevice/superset:6.0.0
container_name: superset_app_next
restart: unless-stopped
depends_on:
- superset_db
- redis
environment:
SUPERSET_SECRET_KEY: "hj9uozuQOM2RTilH5Dwfdo7MqcslMaogDYMY6Xo9IyYS_mYqN5QQAsff9oSUToqvCc0GihnARc4IYQTi022rHw"
SUPERSET_CONFIG_PATH: /etc/superset/superset_config.py
DATABASE_DB: superset
DATABASE_HOST: superset_db
DATABASE_USER: superset
DATABASE_PASSWORD: superset
DATABASE_PORT: 5432
REDIS_HOST: redis
volumes:
- ./superset_home:/app/superset_home
- ./superset_config.py:/etc/superset/superset_config.py
command:
- "sh"
- "-c"
- |
superset fab create-admin \
--username admin \
--firstname Admin \
--lastname User \
--email admin@example.com \
--password admin
superset db upgrade &&
superset init &&
gunicorn --bind 0.0.0.0:8088 \
--workers 4 \
--timeout 120 \
--limit-request-line 0 \
--limit-request-field_size 0 \
'superset.app:create_app()'
networks:
- superset_network
superset_db:
image: postgres:16
container_name: superset_metadata_db_next
restart: unless-stopped
environment:
POSTGRES_DB: superset
POSTGRES_USER: superset
POSTGRES_PASSWORD: superset
volumes:
- ./superset_db_data:/var/lib/postgresql/data
networks:
- superset_network
redis:
image: redis:7
container_name: superset_redis_next
restart: unless-stopped
networks:
- superset_network
nginx:
image: nginx:latest
container_name: superset_nginx_next
restart: unless-stopped
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- ./static:/usr/share/nginx/html:ro
depends_on:
- superset
networks:
- superset_network
pgdataout:
image: postgres:16
container_name: postgres_data_next
environment:
POSTGRES_DB: "testdb"
POSTGRES_USER: "testuser"
POSTGRES_PASSWORD: "testpass"
PGDATA: "/var/lib/postgresql/data/pgdata"
volumes:
- ./mydata:/var/lib/postgresql/data
ports:
- "5433:5432"
networks:
- superset_network
networks:
superset_network:
driver: bridge
После первого запуска кусок
superset fab create-admin \
--username admin \
--firstname Admin \
--lastname User \
--email admin@example.com \
--password password
superset db upgrade &&
superset init &&нужно убрать.
Или его сразу убрать, однако в первый раз запустить команды
docker compose run --rm superset superset db upgrade
docker compose run --rm superset superset fab create-admin ...
docker compose run --rm superset superset initОписание сервисов
| superset | Основной сервис |
| superset_db |
Сервис хранения метаданных. Если не будет, то метаданные будут храниться в /var/lib/superset/superset.db При работе через docker после удаления контейнера все настройки будут сброшены. |
| redis | Кэширующий сервер. |
| nginx | Распределяющий прокси. |
| pgdataout | Не нужен для работы superset. Это для хранения данных дашбордов, указывается в виде dataset. |
nginx.conf
events {}
http {
upstream superset {
server superset:8088;
}
server {
listen 80;
client_max_body_size 50M;
# Статические файлы
location /alterinfo/ {
alias /usr/share/nginx/html/;
index index.html;
}
location / {
proxy_pass http://superset;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_redirect off;
}
}
}
superset_config.py
import os
#SECRET_KEY = os.environ.get("SUPERSET_SECRET_KEY")
SECRET_KEY = "hj9uozuQOM2RTilH5Dwfdo7MqcslMaogDYMY6Xo9IyYS_mYqN5QQAsff9oSUToqvCc0GihnARc4IYQTi022rHw"
CACHE_CONFIG = {
"CACHE_TYPE": "RedisCache",
"CACHE_DEFAULT_TIMEOUT": 300,
"CACHE_REDIS_HOST": "redis",
"CACHE_REDIS_PORT": 6379,
}
CELERY_CONFIG = {
"broker_url": "redis://redis:6379/0",
"result_backend": "redis://redis:6379/0",
}
SQLALCHEMY_DATABASE_URI = (
"postgresql+psycopg2://"
"superset:superset@superset_db:5432/superset"
)
# Использование html в компоненте markdown
ESCAPE_MARKDOWN_HTML = False
HTML_SANITIZATION = False
# Обновление конфигурации безопасности в части источников подгрузки изображений. Добавлено https
from superset.config import TALISMAN_CONFIG as TALISMAN_DEFAULTS
TALISMAN_CONFIG = TALISMAN_DEFAULTS.copy()
csp = TALISMAN_CONFIG.get("content_security_policy", {}).copy()
img_src = csp.get("img-src", [])
img_src += [
"https:",
]
csp["img-src"] = img_src
TALISMAN_CONFIG["content_security_policy"] = csp
Базовый compose для старта в режиме разработки:
services:
superset:
image: apache/superset:latest
container_name: superset
restart: unless-stopped
ports:
- "8088:8088"
environment:
- SUPERSET_SECRET_KEY=change_key_here
volumes:
- ./superset_data:/app/superset_home
command:
- "sh"
- "-c"
- |
superset fab create-admin \
--username admin \
--firstname Admin \
--lastname User \
--email admin@example.com \
--password admin && \
superset db upgrade && \
superset init && \
superset run -p 8088 --with-threads --reload --debugger --host 0.0.0.0Но в стандартном не оказалось драйверов, странно...
Kafka
Архитектура
Apache Kafka — opensource платформа хранения и потоковой передачи событий в режиме реального времени.
Событийная архитектура (Event driven architecture, EDA): шаблон, в котором производство, обнаружение, потребление и реакция на события играют центральную роль. Преимущества:
- Асинхронное взаимодействие, позволяющее функционировать системе при неустойчивой работе части компонентов
- Увеличение гибкости
- Снижение риска каскадных ошибок
Недостатки:
- Согласованность данных.
- Задержки между изменением данных и получением информации об изменении.
- Повышенные операционные издержки
Примеры задач EDA:
Объединяет создание значительных объемов данных, которые асинхронно получаются из различных источников и анализируются позже. Примеры:
- анализ активности пользователей
- сбор логов в задачах администрирования
- детекция мошенничества
- предсказание аварии железа
Ключевые компоненты
Кластерная структура стандартная:
Производители создают и отправляют данные, брокеры распределяют данные по потребителям, потребители получают и обрабатывают данные.
Процессы внутри одного брокера:
| Точка | Критерий выбора | Дополнительно |
| Topic | Topic name | Задается Producer при отправке сообщения. Обязательный параметр |
| Partition | Может задаваться Producer при отправке сообщения для определения Partition. Необязательный параметр. | |
| Key |
Выбор раздела по hash функции от ключа. Необязательный параметр. Но не гарантируется и зависит от количества разделов и количества ключей. Если не указан ни Key, ни Partition, то сообщения отправляются всем Consumer по кругу |
|
| Group | ||
| Consumer |
Offset
Смещение точки непрочитанного сообщения. В рамках Partition. Commit = фиксация.
Группы
- Сообщение из одной parititon не может поступать нескольким consumer в пределах группы
- Один consumer может получать сообщения из нескольких partition
- При нескольких группах consumer, каждая группа будет получать все сообщения
- Одиночный consumer без группы равносилен группе из одного пользователя
Kafka: введение
Сообщение
- Ключ - метаданные для управления записью сообщения в разделы
- Схема - способ разобрать сообщение. Может быть JSON, XML, Apache Avro
- Смещение - точка считывания, добавляется Kafka
Topic - центр объединения данных.
Иерархия распределения данных: Topic - Partition - message. Каждая partition связана с одним единственным потребителю (consumer). Упорядочиваются в пределах Partition.
- Сообщение одной partition не может поступать нескольким consumer в пределах группы
Производители
Потребители
Могут объединяться в группы. Чтение каждого раздела только одним членом группы. Групп может быть много. Если в одной группе consumer больше чем partition то оставшиеся простаивают. Принадлежность - какой раздел какому потребителю.
Брокер
Отдельный сервер Kafka - брокер. Объединяются в кластер, один из брокеров - контроллер. Если раздел нескольким брокерам, то происходит репликация. Основной брокер - ведущий. Механизмы репликации только в пределах одного кластера. MirrorMaker - репликация между кластерами.
| broker.id | Целочисленный идентификатор, с 0, уникальный. |
| listeners |
протокол://имя_хоста:порт Перечисляются через запятую. Имя хоста: конкретный ip - соответствующий интерфейс, 0.0.0.0 - все интерфейсы, не указан - интерфейс по умолчанию. Протокол: PLAINTEXT, SSL Если порт менее 1024 то Kafka от имени root, |
| log.dirs | Директории размещения логов. |
Python: kafka-python
Установка
Kafka с web интерфейсом:
services:
kafka:
image: apache/kafka:4.2.0
container_name: kafka
hostname: kafka
restart: unless-stopped
ports:
- "9092:9092" # INTERNAL (docker network)
- "29092:29092" # EXTERNAL (LAN access)
environment:
# =========================
# KRaft
# =========================
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
# =========================
# Listeners
# =========================
KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL://:29092,CONTROLLER://:9093
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://192.168.1.195:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
# =========================
# Single-node safety
# =========================
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
# Optional tuning
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
volumes:
- ./kafka_data:/var/lib/kafka/data
kafka-ui:
image: provectuslabs/kafka-ui:latest
container_name: kafka-ui
restart: unless-stopped
ports:
- "8080:8080"
environment:
KAFKA_CLUSTERS_0_NAME: local
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
depends_on:
- kafka
На https://kafka.apache.org/quickstart/ есть информация по запуску без docker
Проверка внутренними средствами
Создание темы
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh \
--create \
--topic demo \
--bootstrap-server kafka:9092 \
--partitions 3 \
--replication-factor 1Запуск продюсера (сервер на 1.120)
docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh \
--topic demo \
--bootstrap-server 192.168.1.120:29092Запуск консъюмера (сервер на 1.120)
docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh \
--topic demo \
--from-beginning \
--bootstrap-server 192.168.1.120:29092.
Topics
Тема
Сообщение:
- значение (value) - данные, обязательное
- (partition number)
- ключ (key) -
Создание темы (topic)
| Поле | Тип данных | Описание | ||||||||||
| Topic Name | Строка | Название темы. | ||||||||||
| Number of partitions | Целое число |
Количество разделов — управление производительностью и масштабируемостью кластера. Единицы параллелизма в Kafka. Разделы определяют скорость и масштабирование.
Пример: для 10 разделов можно запустить до 10 потребителей, и каждый будет обрабатывать свой раздел. Если потребителей больше, чем разделов, то лишние потребители будут простаивать и тратить ресурсы впустую . Дополнительные факторы
Практические рекомендации
Выбор "Number of partitions" — это всегда компромисс между желаемой производительностью, требованиями к порядку данных и операционными затратами на поддержку кластера. |
||||||||||
| Cleanup policy |
Список.
|
Политика управления сроком жизни данных. По умолчанию delete. Политика delete (Удаление) Удаление старых данных по истечении заданного времени или при превышении лимита размера. Как это работает: удаление когда старше retention.ms (по умолчанию 7 дней), или когда общий размер раздела превышает retention.bytes. Происходит в фоновом режиме с заданным интервалом. Когда применять: потоки событий и логов, где важна вся история. Например, логи приложений, метрики, трекинговые события. Здесь каждое сообщение самоценно, и старые данные со временем теряют актуальность . Политика compact (Сжатие) Эта политика работает по принципу "сохранить только последнее известное состояние" для каждого ключа сообщения . Как это работает: В фоновом режиме для каждого уникального ключа (key) в разделе остается только последнее по времени сообщение, а все промежуточные удаляет. Сообщения обязаны иметь ключ (key), без ключа будут удалены при сжатии. Когда применять: для хранения состояние или конфигурации. Например:
Комбинированная политика delete,compact Как это работает: сначала удаляются старые данные по времени/размеру, а затем сжимается оставшийся лог. Это полезно, когда нужно ограничить общий объем данных и гарантировать, что для каждого ключа в пределах этого срока хранения сохранится только последнее значение.
Как выбрать? Важна ли полная история событий или только их текущее состояние? Если история — выбирайте delete. Если состояние — выбирайте compact. |
||||||||||
| Min in sync replicas | Целое число |
Минимальное количество брокеров-подписчиков, которые должны подтвердить запись, чтобы она считалась успешной. Иначе — это порог, защищающий от потери данных при сбое брокеров. Работает в паре с настройкой продюсера acks (подтверждения). У топика replication.factor = 3 (3 копии данных на разных брокерах), а min.insync.replicas = 2.
Ключевой момент: если в кластере внезапно упадут 2 брокера из 3, и в живых останется только 1 (лидер), то условие min.insync.replicas=2 перестанет выполняться. В этом случае брокер откажется принимать новые сообщения и вернет продюсеру ошибку NotEnoughReplicasException. Компромисс: Надежность vs Доступность Это выбор между Consistency (согласованностью) и Availability (доступностью) из теоремы CAP. Высокое значение (например, 2 или 3): ✅ Плюс: Максимальная надежность. Вы гарантированно не потеряете данные, даже если несколько брокеров выйдут из строя. ❌ Минус: Риск недоступности. Если выйдет из строя больше брокеров, чем min.insync.replicas, топик станет read-only, пока не починят брокеры или не уменьшите параметр. Низкое значение (например, 1): ✅ Плюс: Максимальная доступность. Топик всегда будет принимать данные, даже если у вас работает всего 1 брокер из 3. ❌ Минус: Риск потери данных. Если лидер упадет в момент, когда он единственный хранил новое сообщение (остальные реплики еще не получили его), это сообщение будет потеряно безвозвратно. Правило настройки (Золотое правило) При настройке топика нужно строго соблюдать соотношение: min.insync.replicas ≤ replication.factor И, что важнее, для реальной отказоустойчивости min.insync.replicas должно быть ≤ (replication.factor - 1). |
||||||||||
| Replication factor | Целое число |
Фундамент отказоустойчивости. Определяет, сколько всего копий каждого раздела будет храниться в кластере Kafka. Есть топик с replication.factor = 3. Это значит, что каждый раздел этого топика будет физически скопирован на 3 разных брокера. Для каждого раздела Kafka выбирает:
Репликация происходит на уровне разделов, а не топиков целиком. Если в топике 10 разделов и replication.factor=3, то в кластере будет храниться 30 копий этих разделов (10 разделов × 3 реплики).
Что происходит при сбое:
На что влияет replication.factor
Правила настройки
RF=1 — любая потеря брокера = потеря данных. Только для dev/test. RF=2 — переживает падение 1 брокера. Но при обслуживании кластера (rolling restart) вы остаетесь без запаса. RF=3 — стандарт индустрии. Переживает падение 2 брокеров, позволяет обслуживать кластер без простоя.
Если у вас 3 брокера, вы не можете поставить RF=5 — Kafka не сможет разместить все реплики. Максимум = число брокеров.
Классическая production-конфигурация: replication.factor = 3 Это дает: данные записаны минимум на 2 брокера, и кластер переживает падение 1 брокера без потери доступности на запись.
|
||||||||||
| Time to retain data | Целое число, ms |
определяет время хранения сообщения в топике перед удалением. По умолчанию 7 дней (168 часов). Есть два уровня настройки: на уровне брокера и на уровне конкретного топика.
Удаление происходит по сегментам, а не по сообщениям Kafka не удаляет отдельные сообщения. Лог раздела разбит на сегменты (segments). Параметр retention.ms определяет, когда закрытый (неактивный) сегмент становится кандидатом на удаление. Что это значит на практике: Активный сегмент (тот, в который сейчас пишутся данные) никогда не удаляется, даже если его сообщения «просрочены». Проверка на удаление запускается периодически (по умолчанию каждые 5 минут), а не мгновенно. Дополнительная задержка в 1 минуту (file.delete.delay.ms) перед физическим удалением файла. Пример из практики (низкая нагрузка) Вы хотите хранить данные 1 день, но у вас мало сообщений. Активный сегмент не «закрывается» (не достигает лимита по размеру segment.bytes или времени segment.ms), поэтому он живет долго. Результат: сообщения могут храниться до 2 дней (1 активный сегмент + 1 неактивный), хотя вы задали retention.ms=1 день. Kafka позволяет комбинировать два независимых ограничения:
Практические рекомендации |
||||||||||
| Maximum message size in bytes | Целое число, bytes | Потолок одного пакета. Но пакеты могут объединяться, так что потолок может быть превышен. | ||||||||||
| Custom parameters | Разные | Дополнительные параметры для тонкой настройки. |
Python + Kafka
Библиотеки
- confluent_kafka от разработчиков
- aiokafka асинхронная библиотека
confluent_kafka
AdminClient
Модуль автоматизации административных задач.
Настройки:
from confluent_kafka.admin import AdminClient, NewTopic
config = {
'bootstrap.servers': '192.168.1.195:29092'
}
admin_client=AdminClient(config)Создание темы:
def create_partition():
topic = 'test_topic'
new_topic = NewTopic(topic, num_partitions=2, replication_factor=1)
futures = admin_client.create_topics([new_topic])
for topic_name, future in futures.items():
try:
future.result()
print(f"Created {topic_name} successfully.")
except Exception as e :
print(f'Error in topic {topic_name} creation: ', e)Топик в списке появляется не сразу, а где-то через 30 секунд.
Список тем:
def list_topics():
metadata = admin_client.list_topics(timeout=10)
print("Существующие топики:", list(metadata.topics.keys()))consumer.py
from confluent_kafka import Consumer, KafkaException
config = {
'bootstrap.servers': '192.168.1.195:29092',
'group.id' : 'my_group'
}
consumer = Consumer(config)
if __name__ == '__main__':
topic = 'test_topic'
timeout_s = 1
consumer.subscribe([topic])
try:
while True:
msg = consumer.poll(timeout=timeout_s)
if msg is None:
continue
if msg.error():
raise KafkaException(msg.error())
else:
print(f'Input: {msg.value().decode()}')
finally:
consumer.close()Если consumer в одной группе и количество partitions не меньше количества consumers, то сообщения распределяются между ними. Если partitions будет меньше consumers (которые в одной группе), то оставшаяся часть будет простаивать.
Функция consumer.subscribe получает список топиков, то есть одним consumer можно подписаться на несколько.
Если при двух и более partitions для одного из partitions отваливается consumer то Kafka автоматически проводит перебалансировку в пределах группы. После подключения опять произойдет перебалансировка.
producer.py
import time
from confluent_kafka import Producer
def delivery_report(err, msg):
if err is not None:
print(f"Message failed: {err}")
raise err
else:
print(f'Message {msg.value()} in topic {msg.topic()} delivered')
config = {
'bootstrap.servers': '192.168.1.195:29092',
'message.timeout.ms' : 5000
}
producer = Producer(config)
def send_message(topic_: str, message_: str):
producer.produce(topic_, value=message_, callback=delivery_report)
producer.flush()
if __name__ == '__main__':
topic = 'test_topic'
message = 'Producer 1'
i = 0
while True:
send_message(topic, f'{message} # {i}')
i += 1
time.sleep(1)Распределение сообщений по ключам
Для распределения по консюмерам используется хэш функция от ключа и количества партиций. При коротких ключах может быть коллизия. Довольно неприятная и сложнообходимая ситуация. Проще указать номер partition (int).
import time
from confluent_kafka import Producer
def delivery_report(err, msg):
if err is not None:
print(f"Message failed: {err}")
raise err
else:
print(f'Message {msg.value()} in topic {msg.topic()} delivered')
config = {
'bootstrap.servers': '192.168.1.195:29092',
'message.timeout.ms' : 5000
}
producer = Producer(config)
def send_message(topic_: str, message_: str, part_: int | None = None):
producer.produce(topic_, partition=part_, value=message_, callback=delivery_report)
producer.flush()
if __name__ == '__main__':
topic = 'test_topic'
message = 'Producer 1'
i = 0
while True:
part1 = 0
part2 = 1
send_message(topic, message_=f'key={key1} {message} # {i}', part_=part1)
send_message(topic, message_=f'key={key2} {message} # {i}', part_=part2)
i += 1
time.sleep(1)
#============================================ consumer =========================================
from confluent_kafka import Consumer, KafkaException
config = {
'bootstrap.servers': '192.168.1.195:29092',
'group.id' : 'my_group'
}
consumer = Consumer(config)
if __name__ == '__main__':
topic = 'test_topic'
timeout_s = 1
consumer.subscribe([topic])
try:
while True:
msg = consumer.poll(timeout=timeout_s)
if msg is None:
continue
if msg.error():
raise KafkaException(msg.error())
else:
print(f'Input: {msg.value().decode()}')
finally:
consumer.close()Транзакции
В config для consumer добавляется элемент 'enable.auto.commit': False
И затем consumer.commit()
Сериализация данных
Через json
while True:
obj = {
'id': i,
'value': randomword(8),
'timestamp': datetime.now().isoformat(timespec='seconds'),
}
send_message(topic, json.dumps(obj))
i += 1
time.sleep(1)
============================================================================
str_msg = msg.value().decode('utf-8')
obj = json.loads(str_msg)
print(f'id={obj.get("id")}, timestamp={obj.get("timestamp")}')
aiokafka
consumer.py
from aiokafka import AIOKafkaConsumer
import asyncio
async def consume():
consumer = AIOKafkaConsumer(
'test_first_one',
bootstrap_servers='192.168.1.195:29092'
)
await consumer.start()
try:
async for msg in consumer:
print(f"Consumed message {msg.value.decode()}")
finally:
await consumer.stop()
if __name__ == "__main__":
asyncio.run(consume())producer.py
from aiokafka import AIOKafkaProducer
import asyncio
async def send():
producer = AIOKafkaProducer(
bootstrap_servers="192.168.1.195:29092"
)
topik = 'test_first_one'
message = 'Hello from python!'
await producer.start()
i = 0
try:
while True:
bytes_msg = f'{message} {i}'.encode('utf-8') # convert string into byte
await producer.send_and_wait(topik, bytes_msg)
i += 1
print(f'Message {i} sent.')
await asyncio.sleep(1)
finally:
await producer.stop()
if __name__ == '__main__':
asyncio.run(send())Yandex DataLens
Установка и настройка
Установка
Склонировать git
git clone https://github.com/datalens-tech/datalensНастраиваем файл настройки .env
DATALENS_BASE_URL=http://ipdockerserver:8080
FORCE_REBUILD_STATIC=1