Список наиболее распространённых проблем и как их избегать: Проблема: Неподходящий формат сообщений (JSON vs Avro). Решение: стандартизируйте формат, используйте Schema Registry и формат Avro/Protobuf. Проблема: Повторная обработка сообщений при падении consumer. Решение: используйте контролируемые группы (kafka_group_name) и управление оффсетами; при необходимости используйте exactly-once upstream. Проблема: Переполнение MergeTree при большом потоке записей. Решение: настройте merge_max_size, партиционирование и регулярно запускайте OPTIMIZE для маленьких партиций. Проблема: Неправильный выбор движка для агрегатов (SummingMergeTree vs AggregatingMergeTree). Решение: выбирайте AggregatingMergeTree для сложных state-агрегатов. Совет: используйте kafka_num_consumers и количество партиций в топике кратное количеству consumer'ов, чтобы обеспечить параллельное потребление и равномерную нагрузку по партициям.

Как масштабировать?

Масштабирование предполагает масштабирование трёх компонент: Kafka (ингест), ClickHouse (хранение/вычисления) и сети между ними. Привожу рекомендации с конкретикой. Kafka: увеличьте число партиций топика. На 10k сообщений/сек и 5k уникальных ключей рекомендуется 24–64 партиций. Увеличение партиций даёт большую параллельность consumer'ов, но повышает нагрузку на контроллеры. ClickHouse: горизонтальное масштабирование через кластер (Distributed + ReplicatedMergeTree). Для 50k событий/сек начните с 3 шардов x 3 реплики, каждая нода с 16 CPU и 64 GB RAM. Используйте Distributed-таблицы и балансировку запросов через remote и cluster конфигурации. Сетевой слой: используйте 10 Gbps между брокерами Kafka и ClickHouse нодами при потоках >100 MB/s. Практический пример: масштабирование агрегатов. Если один MergeTree не успевает выполнять merges, добавьте шардинг по hash(user_id) и распределите агрегаты по шардовой схеме; используйте Distributed view для запросов. Мониторинг: соберите метрики...

как избежать дублирования при перезапуске materialized view?

Дублирование происходит, если materialized view повторно читает одни и те же сообщения: это случается при неправильной конфигурации групп или при ручном реимпорте. Решение — использовать контролируемые kafka_group_name и idempotent inserts (например, уникальный ключ в MergeTree и INSERT с ON DUPLICATE SKIP или использовать ReplacingMergeTree с ключом). Также полезно хранить offset-позиции вне ClickHouse, если нужна сложная логика повторной доставки.

что лучше для schema evolution: JSON или Avro?

Avro с Schema Registry предпочтительнее в продах из-за строгой версии схем и совместимости (backward/forward). JSON проще для быстрых прототипов, но при изменении полей требуется дополнительная обработка на стороне парсера. Для ClickHouse+Kafka production-рекомендуется Avro/Protobuf + Schema Registry (Confluent), особенно если несколько сервисов публикуют события.

почему materialized view не вставляет данные в MergeTree?

Проверьте несколько вещей: совпадает ли database у Kafka-таблицы и materialized view; не произошло ли исключение при парсинге; есть ли ошибки в логах ClickHouse (обычно в /var/log/clickhouse-server/clickhouse-server.log). Также убедитесь, что Kafka consumer group корректна и нет зависимости от security (SASL/SSL). Частая причина — mismatch полей/типов между Kafka-сообщением и схемой таблицы.

сколько памяти потребуется для AggregatingMergeTree на 100 млн уникальных пользователей?

При использовании AggregatingMergeTree для state-агрегатов оценка зависит от типа state. При хранении approx-агрегатов (HyperLogLog, approx count) требуется ~1–5 KB на state, итого 100 млн * 1 KB = ~100 GB. Для точных state (uniqExactState) требуются десятки раз больше. Планируйте 2×–3× для overhead и индексов, используйте внешнее хранение state или approximate-агрегаты, если RAM ограничена.

какой latency achievable с ClickHouse + Kafka?

При локальной сетевой среде и оптимальной конфигурации достигается end-to-end latency 0.5–2 секунд: Kafka доставляет событие обычно Дополнительные материалы доступны в разделах DevOps и Базы данных на сайте, где есть примеры Docker Compose и production-настроек для ClickHouse и Kafka.

Streaming аналитика на ClickHouse + Kafka

-- Создать внешний Kafka engine используя HTTP-интерфейс ClickHouse
curl -sS -X POST 'http://localhost:8123/' --data-binary $'CREATE TABLE default.events_kafka (
    event_time DateTime64(3),
    user_id UInt64,
    action String,
    value Float64
) ENGINE = Kafka()
SETTINGS kafka_broker_list = 'kafka:9092',
         kafka_topic_list = 'events',
         kafka_group_name = 'clickhouse-consumer-1',
         kafka_format = 'JSONEachRow',
         kafka_num_consumers = 1;'

Ok.

Ошибка: Code: 1001. DB::Exception: Cannot parse input: bad JSON in row
Фикс: Проверьте формат сообщений в Kafka (ключ/значение). Для JSONEachRow каждое сообщение должно быть валидным JSON с полями, соответствующими схеме таблицы. Запустите producer с примером:

cat sample.json | kafka-console-producer --broker-list localhost:9092 --topic events

-- Создаём целевую таблицу
curl -sS -X POST 'http://localhost:8123/' --data-binary $'CREATE TABLE default.events_mt (
    event_date Date,
    event_time DateTime64(3),
    user_id UInt64,
    action String,
    value Float64
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id);

-- Создаём materialized view, который читает из Kafka-таблицы и вставляет в MergeTree
curl -sS -X POST 'http://localhost:8123/' --data-binary $'CREATE MATERIALIZED VIEW default.events_mv TO default.events_mt AS
SELECT
    toDate(event_time) AS event_date,
    event_time,
    user_id,
    action,
    value
FROM default.events_kafka;'

Ok.
Ok.

Ошибка: Code: 57. DB::Exception: Unknown table default.events_kafka
Фикс: Проверьте, что таблица Kafka создана в том же database и что ClickHouse-server запущен. Также убедитесь, что вы не создаёте materialized view до Kafka-таблицы.

-- Создаём таблицу с pre-aggregates (1-minute buckets)
curl -sS -X POST 'http://localhost:8123/' --data-binary $'CREATE TABLE default.events_1m_agg (
    bucket_start DateTime64(3),
    action String,
    users_count UInt64,
    value_sum Float64
) ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(bucket_start)
ORDER BY (bucket_start, action);

-- Materialized view обновляет агрегат в режиме near-real-time
curl -sS -X POST 'http://localhost:8123/' --data-binary $'CREATE MATERIALIZED VIEW default.events_1m_mv TO default.events_1m_agg AS
SELECT
    toStartOfMinute(event_time) AS bucket_start,
    action,
    uniqExactState(user_id) AS users_count_state,
    sumState(value) AS value_sum_state
FROM default.events_mt
GROUP BY bucket_start, action;

-- Для SummingMergeTree нужно развернуть state, используем функцию finalizeAggregates при чтении
'

Ok.
Ok.

Ошибка: Code: 49. DB::Exception: Aggregate function uniqExactState is not supported in this context
Фикс: Убедитесь, что используете AggregatingMergeTree как движок или храните агрегатные state-колонки. Альтернатива: заменить на approxAggregate(uniqExactState -> uniqState) или использовать simple counts для приемлемой погрешности.

# Генерация 1000 событий за 5 секунд (пример для bash + kafka-console-producer)
for i in $(seq 1 1000); do
  ts=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ")
  echo "{\"event_time\": \"$ts\", \"user_id\": $((RANDOM%5000+1)), \"action\": \"click\", \"value\": $(awk "BEGIN{srand(); printf \"%.2f\", rand()*10}") }" \
    | kafka-console-producer --broker-list localhost:9092 --topic events > /dev/null
done

# Проверить вставку в ClickHouse
curl -sS 'http://localhost:8123/?query=SELECT+count%281%29+FROM+default.events_mt' | sed -n '1p'

Ошибка при продьюсе: ERROR: Failed to connect to kafka:9092
Фикс: Проверьте, что broker доступен по названию/хосту из контейнера. При Docker Compose используйте алиасы сервисов или пробросьте порты. Тестируйте изнутри контейнера ClickHouse: 

docker exec -it clickhouse-server ping -c 3 kafka

# Пример systemd unit для ClickHouse (файл /etc/systemd/system/clickhouse.service)
[Unit]
Description=ClickHouse Server
After=network.target

[Service]
Type=simple
User=clickhouse
ExecStart=/usr/bin/clickhouse-server --config-file=/etc/clickhouse-server/config.xml
LimitNOFILE=262144
Restart=on-failure

[Install]
WantedBy=multi-user.target

# После добавления:
systemctl daemon-reload
systemctl enable --now clickhouse
systemctl status clickhouse --no-pager

● clickhouse.service - ClickHouse Server
   Loaded: loaded (/etc/systemd/system/clickhouse.service; enabled)
   Active: active (running) since Thu 2026-03-12 12:34:56 UTC; 2min 10s ago

Ошибка: Failed at step USER spawning /usr/bin/clickhouse-server: No such user
Фикс: Создайте пользователя clickhouse: useradd -r -s /sbin/nologin clickhouse, проверьте права на каталоги данных и логов.

Streaming аналитика на ClickHouse + Kafka | KtoHto

Streaming аналитика на ClickHouse + Kafka

Комментарии (0)

Что вы изучите

Требования

Архитектура real-time аналитики

Шаг 1: Kafka engine таблица

Шаг 2: materialized view

Шаг 3: агрегации

Шаг 4: загрузка данных и проверка

Шаг 5: деплой, сервисы и мониторинг

Какие pitfalls?