# Инфраструктура Apache / BI

# Apache Superset

# Установка

**Сборка образа**

```
FROM apache/superset:latest

USER root

RUN pip install --no-cache-dir \
    psycopg2-binary \
    redis

USER superset
```

Дополнительные драйвера:

<table id="bkmrk-%D0%91%D0%94-%D0%94%D1%80%D0%B0%D0%B9%D0%B2%D0%B5%D1%80-mysql-mys" style="width: 99.4048%;"><thead><tr><th style="width: 38.4615%;">БД</th><th style="width: 61.5385%;">Драйвер</th></tr></thead><tbody><tr><td style="width: 38.4615%;">MySQL</td><td style="width: 61.5385%;">mysqlclient</td></tr><tr><td style="width: 38.4615%;">MSSQL</td><td style="width: 61.5385%;">pymssql</td></tr><tr><td style="width: 38.4615%;">ClickHouse</td><td style="width: 61.5385%;">clickhouse-connect</td></tr><tr><td style="width: 38.4615%;">Oracle</td><td style="width: 61.5385%;">cx\_Oracle</td></tr></tbody></table>

```bash
docker build -t test/superset6_1 .
```

Но это не поехало. Нужно использовать инструкцию по сборке образа из git.

Делаем Dockerfile и добавляем нужные драйвера. В инструкции предлагается тег master однако с ним не поехало. Указал конкретную цифру версии.

```yaml
# 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.git
```

URI:

[postgresql://testuser:testpass@pgdataout:5432/testdb](postgresql://testuser:testpass@pgdataout:5432/testdb)

python -m pip install psycopg2-binary

**Агент**

```python
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 обертки**

Обертка для прокручивания дашбордов

```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

```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

```yaml
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
```

Описание сервисов

<table border="1" id="bkmrk-superset-%D0%9E%D1%81%D0%BD%D0%BE%D0%B2%D0%BD%D0%BE%D0%B9-%D1%81%D0%B5" style="border-collapse: collapse; width: 100%;"><colgroup><col style="width: 24.4644%;"></col><col style="width: 75.6548%;"></col></colgroup><tbody><tr><td>superset</td><td>Основной сервис</td></tr><tr><td>superset\_db</td><td>Сервис хранения метаданных.

Если не будет, то метаданные будут храниться в /var/lib/superset/superset.db При работе через docker после удаления контейнера все настройки будут сброшены.

</td></tr><tr><td>redis</td><td>Кэширующий сервер.</td></tr><tr><td>nginx</td><td>Распределяющий прокси.</td></tr><tr><td>pgdataout</td><td>Не нужен для работы superset. Это для хранения данных дашбордов, указывается в виде dataset.</td></tr></tbody></table>

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

```python
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 для старта в режиме разработки:

```yaml
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
```

Но в стандартном не оказалось драйверов, странно...

</body></html>

# Kafka

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

Apache Kafka — opensource платформа хранения и потоковой передачи событий в режиме реального времени.

Событийная архитектура (Event driven architecture, EDA): шаблон, в котором <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="37:10">производство</span>, <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="49:9">обнаружение</span>, <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="60:11">потребление</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="74:3">и</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="78:8">реакция</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="87:2">на</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="90:6">события</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="97:4">играют</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="104:7">центральную</span> <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="112:4">роль</span>. <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="136:1">Преимущества:</span>

- <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="136:1">Асинхронное взаимодействие, позволяющее функционировать системе при неустойчивой работе части компонентов</span>
- <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="136:1">Увеличение гибкости </span>
- <span class="YPkS7KbdpWfGdYKd3QB9" data-src-align="136:1">Снижение риска каскадных ошибок</span>

Недостатки:

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

**Примеры задач EDA:**

Объединяет создание значительных объемов данных, которые асинхронно получаются из различных источников и анализируются позже. Примеры:

- анализ активности пользователей
- сбор логов в задачах администрирования
- детекция мошенничества
- предсказание аварии железа

**Ключевые компоненты**

Кластерная структура стандартная:

<div drawio-diagram="294"><img src="http://bobrobotirk.ru/uploads/images/drawio/2026-09/drawing-1-1788792053.png" alt=""/></div>

Производители создают и отправляют данные, брокеры распределяют данные по потребителям, потребители получают и обрабатывают данные.

Процессы внутри одного брокера:

<div drawio-diagram="296"><img src="http://bobrobotirk.ru/uploads/images/drawio/2026-09/drawing-1-1790087058.png" alt=""/></div>

<table border="1" id="bkmrk-%D0%A2%D0%BE%D1%87%D0%BA%D0%B0-%D0%9A%D1%80%D0%B8%D1%82%D0%B5%D1%80%D0%B8%D0%B9-%D0%B2%D1%8B%D0%B1%D0%BE%D1%80" style="border-collapse: collapse; width: 100%;"><colgroup><col style="width: 12.2765%;"></col><col style="width: 34.2314%;"></col><col style="width: 53.4921%;"></col></colgroup><thead><tr><td class="align-center">Точка</td><td class="align-center">Критерий выбора</td><td class="align-center">Дополнительно</td></tr></thead><tbody><tr><td>Topic</td><td>Topic name</td><td>Задается Producer при отправке сообщения. Обязательный параметр</td></tr><tr><td>Partition</td><td>  
</td><td>Может задаваться Producer при отправке сообщения для определения Partition. Необязательный параметр.</td></tr><tr><td>Key</td><td>  
</td><td>Выбор раздела по hash функции от ключа. Необязательный параметр. Но не гарантируется и зависит от количества разделов и количества ключей.

Если не указан ни Key, ни Partition, то сообщения отправляются всем Consumer по кругу

</td></tr><tr><td>Group</td><td>  
</td><td>  
</td></tr><tr><td>Consumer</td><td>  
</td><td>  
</td></tr></tbody></table>

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 - репликация между кластерами.

<table border="1" id="bkmrk-broker.id-%D0%A6%D0%B5%D0%BB%D0%BE%D1%87%D0%B8%D1%81%D0%BB%D0%B5%D0%BD" style="border-collapse: collapse; width: 100%;"><colgroup><col style="width: 15.5046%;"></col><col style="width: 84.4954%;"></col></colgroup><tbody><tr><td>broker.id</td><td>Целочисленный идентификатор, с 0, уникальный.</td></tr><tr><td>listeners</td><td>протокол://имя\_хоста:порт Перечисляются через запятую. Имя хоста: конкретный ip - соответствующий интерфейс, 0.0.0.0 - все интерфейсы, не указан - интерфейс по умолчанию. Протокол: PLAINTEXT, SSL

Если порт менее 1024 то Kafka от имени root,

</td></tr><tr><td>log.dirs</td><td>Директории размещения логов. </td></tr><tr><td>  
</td><td>  
</td></tr></tbody></table>

Python: kafka-python

# Установка

Kafka с web интерфейсом:

```yaml
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/](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)**

[![image.png](http://bobrobotirk.ru/uploads/images/gallery/2026-09/scaled-1680-/image.png)](http://bobrobotirk.ru/uploads/images/gallery/2026-09/image.png)

<table border="1" id="bkmrk-%D0%9F%D0%BE%D0%BB%D0%B5-%D0%A2%D0%B8%D0%BF-%D0%B4%D0%B0%D0%BD%D0%BD%D1%8B%D1%85-%D0%9E%D0%BF%D0%B8%D1%81" style="border-collapse: collapse; width: 100%; height: 1897.08px;"><colgroup><col style="width: 24.3147%;"></col><col style="width: 16.9337%;"></col><col style="width: 58.7516%;"></col></colgroup><thead><tr style="height: 29.7969px;"><td class="align-center" style="height: 29.7969px;">Поле</td><td class="align-center" style="height: 29.7969px;">Тип данных</td><td class="align-center" style="height: 29.7969px;">Описание</td></tr></thead><tbody><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Topic Name </td><td style="height: 29.7969px;">Строка</td><td style="height: 29.7969px;">Название темы.</td></tr><tr style="height: 855.969px;"><td style="height: 855.969px;">Number of partitions</td><td style="height: 855.969px;">Целое число</td><td style="height: 855.969px;">Количество разделов — управление производительностью и масштабируемостью кластера. Единицы параллелизма в Kafka. Разделы определяют скорость и масштабирование.

- Для производителей (Producers): несколько разделов позволяет отправлять данные параллельно. Больше разделов — выше общая пропускная способность на запись.
- Для потребителей (Consumers): ключевой момент. В одной группе потребителей (consumer group) каждый раздел может обрабатываться только одним потребителем. Количество разделов определяет максимальное число потребителей в группе, которые могут работать параллельно.

**Пример**: для 10 разделов можно запустить до 10 потребителей, и каждый будет обрабатывать свой раздел. Если потребителей больше, чем разделов, то лишние потребители будут простаивать и тратить ресурсы впустую .

**Дополнительные факторы**

- Гарантируется порядок сообщений только внутри одного раздела. Если критичен порядок событий (например, все действия одного пользователя должны обрабатываться последовательно), используется ключ сообщения (key). Все сообщения с одним ключом всегда попадают в один и тот же раздел, сохраняя порядок.
- Каждый раздел это доп. нагрузка (больше файлов, больше метаданных, больше времени на выборы лидера). Слишком большое количество разделов может привести к нестабильности и деградации производительности .
- Количество разделов можно только увеличить, но не уменьшить. Увеличение числа разделов приводит к перебалансировке потребителей и может изменить способ распределения сообщений по ключам, что нарушит порядок.

**Практические рекомендации**

- Для большинства новых тем начинать с 6–12 разделов. Это хороший баланс между производительностью и управляемостью.
- Расчет от потребностей. Если один раздел обрабатывает 10 МБ/с, а нужно 100 МБ/с, то потребуется как минимум 10 разделов .
- Учет роста. Число разделов должно быть с запасом на 2-3 года роста.
- Согласование с потребителями. Количество разделов кратно (или равно) максимальному числу потребителей, которое планируется запускать для параллельной обработки .

Выбор "Number of partitions" — это всегда компромисс между желаемой производительностью, требованиями к порядку данных и операционными затратами на поддержку кластера.

</td></tr><tr style="height: 832.531px;"><td style="height: 832.531px;">Cleanup policy</td><td style="height: 832.531px;">Список.

- Delete
- Compact
- Compact, Delete

</td><td style="height: 832.531px;">Политика управления сроком жизни данных. По умолчанию delete.

**Политика delete (Удаление)**

Удаление старых данных по истечении заданного времени или при превышении лимита размера.

**Как это работает**: удаление когда старше retention.ms (по умолчанию 7 дней), или когда общий размер раздела превышает retention.bytes. Происходит в фоновом режиме с заданным интервалом.

**Когда применять**: потоки событий и логов, где важна вся история. Например, логи приложений, метрики, трекинговые события. Здесь каждое сообщение самоценно, и старые данные со временем теряют актуальность .

**Политика compact (Сжатие)**

Эта политика работает по принципу "сохранить только последнее известное состояние" для каждого ключа сообщения .

**Как это работает**: В фоновом режиме для каждого уникального ключа (key) в разделе остается только последнее по времени сообщение, а все промежуточные удаляет. Сообщения обязаны иметь ключ (key), без ключа будут удалены при сжатии.

**Когда применять**: для хранения состояние или конфигурации.

Например:

- Таблица с актуальными данными пользователей (адрес, почта)
- Информация о текущем статусе заказа.
- Список доступных товаров на складе.
- Внутренний топик Kafka \_\_consumer\_offsets, где хранятся текущие смещения потребителей, по умолчанию использует именно эту политику .

**Комбинированная политика delete,compact**

**Как это работает**: сначала удаляются старые данные по времени/размеру, а затем сжимается оставшийся лог. Это полезно, когда нужно ограничить общий объем данных и гарантировать, что для каждого ключа в пределах этого срока хранения сохранится только последнее значение.  
**Важные нюансы**

- Изменяемость: политику можно изменить для уже существующего топика без перезапуска кластера.
- Смена политики: переключение с delete на compact возможно, но требует осторожности. При включении сжатия сообщения без ключа будут потеряны.
- Производительность: сжатие требует дополнительных ресурсов ЦП и ввода-вывода, так как в фоне работает специальный "чистильщик" (Log Cleaner). При неправильной настройке он может повлиять на производительность брокера.

**Как выбрать?**

Важна ли полная история событий или только их текущее состояние? Если история — выбирайте delete. Если состояние — выбирайте compact.

</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Min in sync replicas</td><td style="height: 29.7969px;">Целое число</td><td style="height: 29.7969px;">Минимальное количество брокеров-подписчиков, которые должны подтвердить запись, чтобы она считалась успешной.

Иначе — это порог, защищающий от потери данных при сбое брокеров.  
**Как это работает (в связке с ack)**

Работает в паре с настройкой продюсера acks (подтверждения). У топика replication.factor = 3 (3 копии данных на разных брокерах), а min.insync.replicas = 2.

- Продюсер отправляет сообщение с настройкой acks=all (или acks=-1), что означает: "Жди подтверждения от всех реплик, которые сейчас в синхронизации".
- Брокер-лидер принимает сообщение и ждет, пока его сохранят у себя как минимум 2 брокера (лидер + 1 реплика-подписчик).
- Только когда эти 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).

</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Replication factor</td><td style="height: 29.7969px;">Целое число</td><td style="height: 29.7969px;">Фундамент отказоустойчивости. Определяет, сколько всего копий каждого раздела будет храниться в кластере Kafka.  
**Что это означает на практике**

Есть топик с replication.factor = 3. Это значит, что каждый раздел этого топика будет физически скопирован на 3 разных брокера.

Для каждого раздела Kafka выбирает:

- 1 Лидера (Leader) — брокер, который обрабатывает все чтения и записи для этого раздела.
- N-1 Подписчиков (Followers) — брокеры, которые просто копируют данные у лидера и ждут своего часа.

Репликация происходит на уровне разделов, а не топиков целиком. Если в топике 10 разделов и replication.factor=3, то в кластере будет храниться 30 копий этих разделов (10 разделов × 3 реплики).  
**Как это работает**:

- Продюсер пишет данные → они попадают к лидеру раздела.
- Лидер сохраняет данные у себя и реплицирует их на всех подписчиков.
- Подписчики копируют данные и отправляют лидеру подтверждение.
- Лидер ждет нужное количество подтверждений (то самое min.insync.replicas) и отвечает продюсеру.

**Что происходит при сбое**:

- Если лидер падает, Kafka выбирает нового лидера из числа синхронизированных подписчиков (In-Sync Replicas, ISR).
- Если падает подписчик, он просто выпадает из ISR, но данные не теряются, пока жив лидер и хотя бы min.insync.replicas реплик.

На что влияет replication.factor

<table border="1" style="border-collapse: collapse; width: 100.224%; height: 199.375px;"><colgroup><col style="width: 34.9853%;"></col><col style="width: 64.8173%;"></col></colgroup><tbody><tr style="height: 80.1875px;"><td style="height: 80.1875px;">Отказоустойчивость</td><td style="height: 80.1875px;">Чем выше фактор, тем больше брокеров может выйти из строя без потери данных. RF=3 переживает падение 2 брокеров (при min.insync.replicas=2).</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Доступность на чтение</td><td style="height: 29.7969px;">При падении лидера новый выбирается из ISR. Чем больше реплик, тем выше шанс, что хотя бы одна синхронизирована.</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Дисковое пространство</td><td style="height: 29.7969px;">RF=3 означает, что каждый байт данных хранится трижды. Это напрямую увеличивает требования к диску в 3 раза.</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Сетевой трафик</td><td style="height: 29.7969px;">Каждая репликация — это дополнительный трафик между брокерами. Чем выше RF, тем больше нагрузка на сеть.</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Скорость записи</td><td style="height: 29.7969px;">Запись успешна после подтверждения от min.insync.replicas. Больше реплик — дольше ожидание, выше latency.</td></tr></tbody></table>

**Правила настройки**

- replication.factor должен быть ≥ 3 для production

 RF=1 — любая потеря брокера = потеря данных. Только для dev/test.

 RF=2 — переживает падение 1 брокера. Но при обслуживании кластера (rolling restart) вы остаетесь без запаса.

 RF=3 — стандарт индустрии. Переживает падение 2 брокеров, позволяет обслуживать кластер без простоя.

- replication.factor ≤ количество брокеров в кластере

Если у вас 3 брокера, вы не можете поставить RF=5 — Kafka не сможет разместить все реплики. Максимум = число брокеров.

- Связка с min.insync.replicas

Классическая production-конфигурация:  
text

replication.factor = 3  
min.insync.replicas = 2  
acks = all

Это дает: данные записаны минимум на 2 брокера, и кластер переживает падение 1 брокера без потери доступности на запись.  
**Важные нюансы**

- replication.factor задается при создании топика и может быть увеличен позже (через kafka-reassign-partitions.sh), но не может быть уменьшен без пересоздания топика.
- Реплики распределяются автоматически по брокерам с учетом rack awareness (если настроено). Kafka старается разместить реплики на разных стойках/зонах доступности.
- ISR (In-Sync Replicas) — это динамический список реплик, которые полностью синхронизированы с лидером. Только они могут стать новым лидером. Если подписчик отстает (например, из-за сети), он исключается из ISR.
- replication.factor не влияет на количество партиций — это независимые параметры. RF — про копии, partitions — про параллелизм.

</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Time to retain data</td><td style="height: 29.7969px;">Целое число, ms</td><td style="height: 29.7969px;">определяет время хранения сообщения в топике перед удалением. По умолчанию 7 дней (168 часов). Есть два уровня настройки: на уровне брокера и на уровне конкретного топика.

Удаление происходит по сегментам, а не по сообщениям

Kafka не удаляет отдельные сообщения. Лог раздела разбит на сегменты (segments). Параметр retention.ms определяет, когда закрытый (неактивный) сегмент становится кандидатом на удаление.

**Что это значит на практике:**

 Активный сегмент (тот, в который сейчас пишутся данные) никогда не удаляется, даже если его сообщения «просрочены».

 Проверка на удаление запускается периодически (по умолчанию каждые 5 минут), а не мгновенно.

 Дополнительная задержка в 1 минуту (file.delete.delay.ms) перед физическим удалением файла.

**Пример из практики (низкая нагрузка)**

Вы хотите хранить данные 1 день, но у вас мало сообщений. Активный сегмент не «закрывается» (не достигает лимита по размеру segment.bytes или времени segment.ms), поэтому он живет долго.

Результат: сообщения могут храниться до 2 дней (1 активный сегмент + 1 неактивный), хотя вы задали retention.ms=1 день.  
**Сравнение с retention.bytes**

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

<table border="1" style="border-collapse: collapse; width: 100.224%;"><colgroup><col style="width: 33.1933%;"></col><col style="width: 33.1933%;"></col><col style="width: 33.1933%;"></col></colgroup><tbody><tr><td>Параметр</td><td>Ограничивает</td><td>Когда срабатывает</td></tr><tr><td>retention.ms</td><td>время хранения</td><td>Сегмент старше указанного времени</td></tr><tr><td>retention.bytes</td><td>размер раздела</td><td>Суммарный размер сегментов превышает лимит</td></tr></tbody></table>

**Практические рекомендации**  
Логи приложений 3–7 дней (достаточно для отладки)  
Метрики и мониторинг 1–3 дня  
События для аналитики 7–30 дней  
Комплаенс/аудит 90+ дней

</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Maximum message size in bytes</td><td style="height: 29.7969px;">Целое число, bytes</td><td style="height: 29.7969px;">Потолок одного пакета. Но пакеты могут объединяться, так что потолок может быть превышен.</td></tr><tr style="height: 29.7969px;"><td style="height: 29.7969px;">Custom parameters</td><td style="height: 29.7969px;">Разные</td><td style="height: 29.7969px;">Дополнительные параметры для тонкой настройки.</td></tr></tbody></table>

# Python + Kafka

Библиотеки

- confluent\_kafka от разработчиков
- aiokafka асинхронная библиотека

**confluent\_kafka**

**AdminClient**

Модуль автоматизации административных задач.

Настройки:

```python
from confluent_kafka.admin import AdminClient, NewTopic

config = {
    'bootstrap.servers': '192.168.1.195:29092'
}

admin_client=AdminClient(config)
```

Создание темы:

```python
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 секунд.

Список тем:

```python
def list_topics():
    metadata = admin_client.list_topics(timeout=10)
    print("Существующие топики:", list(metadata.topics.keys()))
```

**consumer.py**

```python
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**

```python
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).

```python
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

```python
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

```python
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

```python
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

```bash
git clone https://github.com/datalens-tech/datalens
```

Настраиваем файл настройки .env

```
DATALENS_BASE_URL=http://ipdockerserver:8080
FORCE_REBUILD_STATIC=1
```