Когда у вас появляется продукт с активными процессами или пользователями, данные начинают расти быстрее, чем ожидалось. Сначала всё просто: есть PostgreSQL, в котором хранятся основные сущности, есть Kafka с событиями, и вроде бы этого достаточно. Но в какой-то момент возникает типичная ситуация: команде нужна аналитика. Не просто «посчитать количество записей», а ответить на вопросы уровня:

  • что происходило с объектом во времени;

  • какие действия совершила система;

  • как менялось состояние после определённых событий;

  • как определить кто виноват в той или иной проблеме.

И тут начинаются проблемы. Основные сущности хранятся в Kafka. Так зачем изобретать что то? Ведь есть ksqlDB (KSQL) и еще есть KafkaUI. Может попробовать строить аналитику прямо на событиях в Kafka? Спойлер — вообще нет. Kafka пусть лучше остается источником событий, а не аналитической базой.

На практике это оказалось нежизнеспособно:

  • запросы выполнялись медленно;

  • сложную аналитику писать неудобно;

  • поиск по топикам через UI превращался в отдельный квест;

  • сложные запросы хрупкие, могут ломаться и требуют пересборки с нуля;

  • Kafka отлично подходит как транспорт и источник событий, но не как основа для аналитики.

Часть результирующей информации лежит в PostgreSQL. Он отлично справляется с транзакционной нагрузкой. Но аналитику делать в OLTP идея прискорбная. Сложные JOIN’ы, агрегации по большим объёмам, попытки собрать историю изменений — все это создает избыточную нагрузку. Проблема даже не в том, что все начинает работать медленно, а в том, что со временем они начинают мешать основным процессам системы.

Наша задача сформулировалась так:

  • объединить PostgreSQL и Kafka в единую аналитическую модель;

  • не нагружать продовую базу;

  • обеспечить быстрые и предсказуемые запросы;

  • реализовать инструмент анализа вне существующих систем c быстрой скоростью обработки.

Программа минимум — предоставить техническим специалистам быстрый, предсказуемый и удобный инструмент для поиска и анализа данных по всей системе. На тот момент существующие решения, такие как Kafka UI и ksqlDB, не удовлетворяли нашим требованиям по удобству и возможностям аналитики вне Kafka.

Впоследствии сформировалась и программа максимум — бизнесу также потребовался доступ к аналитике и данным в удобном формате. Для этого был внедрён полноценный BI-инструмент на базе Apache Superset, который позволил предоставить пользователям самостоятельный доступ к аналитике и визуализации данных. В итоге мы собрали пайплайн:

PostgreSQL→ PeerDB → ClickHouse→ Superset

Kafka→ ClickHouse→ Superset

Но не будем забегать вперед. К этому варианту мы пришли не сразу. Он стал результатом серии экспериментов, проб и отказов от более сложных или менее подходящих решений.

Эта статья посвящена технической стороне решения построению аналитического контура и практическим аспектам его реализации. Но у любого большого проекта есть своя история: неожиданные ограничения и неудачные эксперименты. О том, как проходил этот путь, рассказал наш лид в статье Per aspera ad astra. Как построить космолет, не привлекая внимания санитаров.

1. Почему выбрали ClickHouse?

Перед тем как прийти к ClickHouse, мы попробовали использовать OpenSearch как основное хранилище для аналитики. Важно отметить, что изначально полноценная BI-система не планировалась — задача была проще: получить возможность просматривать состояние системы, логи и события в удобном виде.

На первый взгляд решение выглядело разумным: OpenSearch часто используют для работы с логами, событиями и быстрым поиском. Однако на практике мы достаточно быстро столкнулись с ограничениями.

Объём данных. Даже при работе всего с одним топиком Kafka стало видно, что данные занимают существенно больше места, чем ожидалось. Сжатие есть, но оно не решает проблему радикально. При этом основные процессы системы проходят только на 5 день. Для анализа нужно держать хотя бы 7-10 дней истории. И таких топиков десятки. В такой конфигурации объём хранилища начинает расти очень быстро и становится трудно контролируемым.

Скорость загрузки данных. Вторая проблема это производительность. Данные грузили через Kafka Connector. Именно его интеграция с OpenSearch оказался узким местом. Процесс трансформации в ЕTL для OpenSearch оказался не очень удобным. Сложную логику преобразований реализовывать тяжело, пайплайн становится хрупким. А любые изменения требуют дополнительных усилий и времени. Скорость загрузки была ниже ожидаемой, а тюнинг коннектора оказался нетривиальной задачей. По итогу добиться стабильной и высокой производительности не удалось.

Ограничение трансформации данных. Отдельная проблема это необходимость преобразовывать данные еще на этапе загрузки. OpenSearch не ограничивал нас по следующим моментам:

  • фильтровать лишние поля и не хранить ненужные данные;

  • раскладывать вложенные объекты и массивы в плоскую структуру;

  • поддерживать сложные схемы сообщений;

  • полноценно работать с AVRO форматом.

В связке Kafka Connector + OpenSearch такие задачи реализовывались сложно и неудобно. Любая нетривиальная трансформация быстро усложняла пайплайн и снижала его надёжность. В результате стало понятно, что мы используем инструмент не совсем по назначению. OpenSearch хорошо подходит для поиска, логирования или простых фильтраций по событиям. Но в нашем случае требовалось:

  • хранить большие объёмы данных;

  • быстро их забирать;

  • простые запросы не достаточны, нужно осуществлять сложные агрегации.

Почему же все таки ClickHouse? Мы начали смотреть его в разрезе наших сценариев и он показал себя значительно лучше: за счёт предсказуемого поведения при росте объёмов данных, стабильной и быстрой работы при постоянной потоковой загрузке, удобной агрегации и структурирования данных.

Дальше — подробнее по каждому из аспектов.

Хранение данных: когда требуется добавить «ещё один топик», а их уже десятки.

Одним из ключевых факторов при выборе ClickHouse стало эффективное хранение данных. В нашей системе:

  • 22 микросервиса генерируют события;

  • 20 млн сущностей в день. Но данные приходят в разрозненном виде и их нужно объединить. Фактическое количество от 300 до 500 млн записей в день;

  • для аналитики требуется хранить в основном по 10 дней истории;

  • источники данных:

    • события из Apache Kafka;

    • объекты из PostgreSQL.

Это сотни миллионов записей ежедневно и милиарды на горизонте хранения. За счёт колоночного формата и эффективного сжатия в ClickHouse удалось существенно сократить объём данных:

  • из Apache Kafka сжатие в 6 раз;

  • из PostgreSQL сжатие в 2.5 раза.

Загрузка данных: почти real-time на практике, а не в теории

Данные из Apache Kafka попадают в ClickHouse практически в реальном времени. От нескольких секунд до минуты после публикации события. Для оценки задержки мы считаем разницу между временем события и временем его вставки в ClickHouse.

На примере одной таблицы распределение получилось следующим:

  • около 140 млн событий (около 50%) обрабатываются менее чем за 5 секунд;

  • более 80% данных доезжают в систему в пределах 10 секунд;

  • задержки свыше минуты насчитывает несколько тыс. случаи (доли процента).

Фактически пайплайн обеспечивает близкую к реальному времени (NRT) загрузку данных. Учитывая, что это аналитическая система и данные не требуют мгновенной консистентности (допустима задержка даже до часов), полученные показатели можно считать более чем достаточными для задач BI и расследования инцидентов.

Агрегация: когда данные нужно не фильтровать, а приводить в порядок структуру

ClickHouse дал нам возможность использовать сложные аналитические запросы. Они выполнялись быстро и предсказуемо. СУБД дают возможность в реальном времени разделять данные по параметрам в разные таблицы и наоборот. Достаточно подробно процесс описан в этой статье.

Наши данные, с которыми мы работали, требовали несколько различных сценариев обработки:

  • Разделение

В одном топике могут храниться сразу несколько сущностей (например, основные данные, детали по объекту и дополнительные атрибуты). Каждая из них приходит отдельными сообщениями. Несмотря на общий поток, эти сущности имеют разную семантику, частоту обновлений и требования к обработке. В ClickHouse мы это решили через разбиение на отдельные витрины и таблицы состояний, каждая из которых отвечает за свою сущность. Фильтрация при этом задаётся явно в WHERE: из общего потока выбираются только те события, которые относятся к конкретной бизнес-сущности. Если данные внутри сообщения представлены в виде массива, он извлекается из payload и разворачивается в строки с помощью ARRAY JOIN, после чего можно фильтровать уже отдельные элементы этого массива.

--тут указываем условия создания представления (источник таблицы, название представления, поля)
...
FROM kafka.table
ARRAY JOIN JSONExtractArrayRaw(payload, '<array>') AS obj
WHERE
-- фильтрация верхнеуровневых атрибутов
    <top_field_1> = '<value>' -- фильтрация верхнеуровневых атрибутов
    AND <top_field_2> = '<value>'
-- фильтрация вложенных объектов массива по значению атрибута.
    AND JSONExtractString(obj, '<attribute>') IN (<value1>, <value2>)

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

  • Соединение

Обратная ситуация — когда данные одной сущности распределены по нескольким топикам. Например, основная информация приходит в одном потоке, а обогащение или дополнительные атрибуты уже в другом. В этом случае на уровне ClickHouse выполняется объединение нескольких топиков, сборка через входную таблицу. Для этого при создании таблицы в параметре kafka_topic_list указывается список источников.

--тут указываем условия создания входной таблицы (название таблицы, поля) 
...
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'host:port',
    kafka_topic_list = 'name_topic_1, name_topic_2',
    kafka_group_name = 'group',
    kafka_format = 'AvroConfluent',
    format_avro_schema_registry_url = 'format'

Это позволяет получить целостный объект, пригодный для аналитики, несмотря на физическое разделение источников.

  • Агрегация в потоке

События в топике могут приходить в разрозненном виде: часть полей отсутствует и досылается позже в сообщениях с тем же ID. При этом в системе есть ограничение на время сборки объекта, после которого он считается финальным. Это создаёт риск неполноты данных, поэтому важно не только собирать события, но и наблюдать за качеством их формирования. Для этого в ClickHouse мы реализовали отдельную витрину, которая агрегирует статистику по отсутствующим атрибутам. Представление(view) строится поверх таблицы состояний и работает уже с финализированными данными за счёт использования оператора FINAL. Агрегации считаются через условные функции countIf, а фильтрация явно задаётся в WHERE. В выборку попадают только те записи, для которых процесс сборки завершён, что позволяет анализировать именно итоговое состояние объектов.

--тут указываем условия создания представления (источник таблицы, название представления, время обновления, поля)
...
DEFINER = clickhouse SQL SECURITY DEFINER
AS SELECT <attribute>,
          countIf(<attribute1> = 0),
          countIf(<attribute2> = 0),
          countIf(<attribute3> = 0)
FROM kafka.table FINAL
WHERE (<attribute4> = 1) AND (<attribute5> = 1) --признак что сборка завершена;

В результате ClickHouse используется не только как инструмент агрегации и подготовки данных, но и как слой нормализации структуры: он помогает как разделять разнородные данные, так и собирать их обратно в удобный для анализа вид, оставаясь частью потоковой архитектуры. В задачах подготовки данных для BI ClickHouse показал себя как удобный и гибкий инструмент. При этом его возможности значительно шире. Описанные сценарии являются лишь частью того, что мы использовали на практике.

Гибкость обработки: данные не обновляются — они пересобираются

ClickHouse не является реляционной СУБД, но работает с таблицами и имеет SQL-подобный язык запросов. Он не ориентирован на частые обновления и удаления строк. Вместо этого данные только добавляются, а изменения достигаются за счёт переобработки и слияния. Эта модель реализуется через движки семейства MergeTree, которые спроектированы для работы с очень большими объёмами данных и высокой скоростью записи. Данные записываются не «на место», а в виде отдельных частей. Каждая вставка создаёт новую часть на диске, уже отсортированную по ключу. Далее фоновый процесс выполняет слияние (merge) этих частей, постепенно объединяя их и оптимизируя хранение. Вот несколько варианты и их назначение:

MergeTree — Базовый движок. Даёт полный контроль: сам определяешь ключ сортировки, партиции, TTL. Подходит, если тебе не нужны дополнительные механики вроде дедупликации или агрегации.

ReplacingMergeTree — Автоматически заменяет дубликаты строк при мерджах. Используется, когда возможны повторные записи. Можно указать колонку версии, чтобы выбирать самую актуальную запись. Далее использовать оператор FINAL.

SummingMergeTree — Суммирует числовые значения при слиянии частей. Работает по ключу сортировки. Подходит для предагрегированных данных.

AggregatingMergeTree— Хранит агрегатные состояния (не просто числа, а состояния функций). Использует AggregateFunction. Позволяет делать гибкую агрегацию уже после записи.

CollapsingMergeTree — Позволяет схлопывать строки с противоположными знаками. Использует специальную колонку sign (+1 / -1). При мерджах удаляет «взаимоуничтожающиеся» записи Это дает возможность:

  • эффективно обрабатывать потоковые вставки без блокировок;

  • поддерживать высокую скорость записи даже при больших нагрузках;

  • избегать дорогостоящих операций UPDATE/DELETE на уровне отдельных строк;

  • автоматически оптимизировать структуру данных по мере накопления;

  • eстанавливать TTL для автоматического удаления данных.

За счёт этого ClickHouse хорошо подходит для потоковых сценариев (например, загрузки из Kafka), где данные поступают непрерывно, а требования к скорости записи и объёму значительно важнее, чем необходимость точечных изменений.

2. Выбор хранилища для аналитики

В компании уже использовался Superset и в рамках задачи по реализации BI решили использовать его. По событиям из Kafka описал выше, но это только часть данных. Вторая же часть данных хранились в PostgreSQL. В рамках задачи визуализации данных в BI основной целью было обеспечить его данными без негативного влияния на OLTP. Попытки строить аналитику напрямую поверх неё приводили бы к росту нагрузки и потенциально влияли на бизнес-логику. Поэтому очевидно, что данные необходимо выносить в отдельное аналитическое хранилище. На этапе проработки рассматривалось несколько вариантов.

Независимая реплика PostgreSQL. Быстрый, но ограниченный путь

В качестве временного решения или просто обсудить все возможные варианты был предложен вариант разгрузить основную базу и направить BI-запросы на реплику. Да, так мы по быстрому реализуем сбор данных, однако это не решает ключевую проблему: PostgreSQL остаётся OLTP-системой и плохо подходит для тяжёлой аналитики. Да, Superset дает возможность реализовывать ускорение данных через Redis. Но это касается только старых данных. Для этого нужно первому пользователю зайти в интерфейс BI. Данные перераспределяются, но концептуально нагрузка на кластер остается. В целом природа транзакционной базы данных не подходит для аналитики. В долгосрочной перспективы это потенциальные проблемы.

OLAP-кубы: мечта или лишний шаг?

Реализовывая BI систему невольно задумываешься о масштабирования. Модель данных, которую сейчас реализуем с помощью какого то узла решит наши проблемы. В памяти всплыли OLAP-кубы. Представьте себе: один раз вы моделируете данные, строите куб, и дальше все пользователи, как по волшебству, получают быстрые и удобные отчёты. При наших поисках нашли мощные инструменты — Apache Kylin, Apache Druid. Они позволяют агрегировать данные заранее, и, по идее, это снимает нагрузку с BI-инструмента. Но возникает вопрос: а готовы ли мы сейчас создавать такую сложную инфраструктуру? Из того что описано в интернете и внутренний опыт подсказывает, что OLAP-кубы это не просто. Они требуют своих процессов, своих команд, своей экспертизы. Важно понимать, что это не просто куб, который ты развернул и забыл. Это система, в которой ты должен жить, обновлять кубы, поддерживать их актуальность. А в нашей команде не было нужных компетенций, да и скажем честно нет в этом целей у проекта. Из того перечня источников, которых у нас уже есть (PostgreSQL, данные из Kafka в ClickHouse) получается очень сложная архитектура. Такое явно будет не надежно работать. Мы понимали, что это только добавит сложности, лишних узлов.

Поэтому мы отказались от OLAP-кубов. Мы поняли, что нам нужен инструмент гибкий, знакомый и менее требующий дополнительных узлов. Так мы пришли к ClickHouse. Он дал нам возможность строить витрины, агрегировать данные и строить аналитику.

ClickHouse. Старый друг лучше новых двух

Когда мы были на этом этапе ClickHouse уже был нам знаком. До этого мы использовали его для потоковой загрузки. Настройка заняла минимальное время, данные начали поступать стабильно и быстро, а сам процесс интеграции оказался неожиданно простым и предсказуемым. Это создало ощущение, что инструмент хорошо ложится на наши задачи уже «из коробки».

По мере погружения стало понятно, что ClickHouse имеет множество возможностей и это не просто быстрый приёмник данных. Его настоящая сила раскрывается в подготовке аналитических витрин. В связке с Superset мы получили полноценный контур для BI. Поверх сырых данных мы начали строить материализованные представления, упрощённые витрины и агрегированные наборы данных. В ход пошли стандартные, но очень практичные механизмы ClickHouse: MATERIALIZED VIEW, агрегирующие таблицы, GROUP BY, ROLLUP, CUBE, grouping sets, сложные оконные и предварительные агрегации и т.д.

По сути, мы формировали всю тяжёлую бизнес-логику не в BI-инструменте, а внутри базы. Это позволило:

  • упростить модели в Superset;

  • уменьшить время построения дашбордов;

  • cнизить нагрузку на слой визуализации.

Отдельно отмечу, что ClickHouse оказался хорошо описанной технологией. В сообществе, в документации и в открытых источниках уже накоплено большое количество практических кейсов: от простых витрин до сложных аналитических систем. И по мере изучения этих материалов становилось всё более очевидно, что ClickHouse часто используется как альтернатива OLAP-кубам, но с меньшей сложностью инфраструктуры и большей гибкостью. Там, где классические OLAP-решения требуют заранее жёстко зафиксированной модели, ClickHouse позволяет итеративно развивать структуру данных. Мы можем менять витрины, пересобирать агрегации и адаптировать модель под новые бизнес-требования без перестройки всей системы. Конечно все налету не создастся. Нужно ждать пока заполнятся данные, но никто не мешает добавить еще один дашборды и отображать из двух витрин данные на время перехода.

В итоге ClickHouse для нас стал не просто хранилищем или инструментом для потоковой обработки, а полноценным аналитической слоем данных.

3. Механизм миграции данных из PostgreSQL

После того как определились с аналитическим слоем в виде ClickHouse возник вопрос: А как и чем переносить данные из PostgreSQL?

Была сформирована потребность в следующем виде:

  • часть данных имеет изменяемые значения (статусы, поля и т.д.). Поэтому для анализа часть данных требует отслеживания в реальном времени. Фактически нужен Change Data Capture, но без жёсткого real-time;

  • миграция не должна оказывать заметного влияния на нагрузку PostgreSQL (никаких тяжёлых SELECT’ов к Prod’ой базе);

  • увеличение инфраструктуры должно быть минимальным. Без раздувания зоопарка сервисов;

  • cложная бизнес-логика в пайплайне не требуется. Задача просто доставить данные в ClickHouse;

  • дополнительных потребителей данных (помимо аналитики) нет.

Командой было вынесено на обсуждение несколько вариантов.

Debezium: мощно, гибко, но сложно и излишне для нас

Debezium — мощный и гибкий инструмент для CDC. С помощью его можно отлеживает состояния во многих системах. Не только в PostgreSQL, но и в MySQL, MongoDB и т.д. Инструмент, пожалуй, является самым популярным.

По факту Debezium — это не самостоятельное решение, а часть полноценного CDC-пайплайна. В него входят Apache Kafka и Kafka Connect. Именно за счёт этого достигаются ключевые преимущества:

  • гарантированная доставка событий;

  • возможность переиспользования данных другими сервисами;

  • буферизация и устойчивость к сбоям.

Из явных плюсов — это любые изменения в БД отслеживаются без обращений к самой базе. Debezium читает Журнал предварительной записи (WAL), а не через SELECT’ы. Это означает, что миграция практически не создаёт нагрузки на транзакционную базу. При каждом изменении базы данных в результате транзакции, изменения сначала записываются в WAL, а затем применяются к фактическим файлам данных.

По факту несмотря на все преимущества, для нашей задачи решение выглядело слишком тяжёлым. Важно отметить, что в нашем случае Apache Kafka уже присутствовал в инфраструктуре. Тем не менее, мы сознательно не стали использовать его как транспорт для CDC с Debezium. Вовлечение Kafka в данный пайплайн означало бы расширение зоны ответственности и усложнение существующей архитектуры. Фактически, даже при наличии кластера, пришлось бы дополнительно поднимать и поддерживать Kafka Connect, настраивать отдельные коннекторы, заводить топики под CDC, следить за схемами сообщений и их эволюцией. Это автоматически тянет за собой всё то же сопровождение стримингового слоя: контроль лагов, обработку ретраев, возможные дубликаты и общее состояние консьюмеров.

При этом сама задача не требовала всех преимуществ стриминговой платформы. Нам не нужно было строить событийную шину, у нас не было нескольких потребителей, отсутствовала необходимость в потоковых трансформациях. По сути, Kafka в данном сценарии использовалась бы исключительно как транспортный слой, что делало её применение избыточным.

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

AirFlow

Apache Airflow — это ещё один инструмент, с помощью которого можно реализовать перенос данных. Однако он является лишь частью общего ETL-процесса, а не готовым решением «из коробки».

Airflow выступает в роли оркестратора. Он управляет выполнением задач, но не решает саму задачу доставки изменений. По сути, он только координирует процесс выгрузки данных из PostgreSQL и их загрузки в ClickHouse. Для построения полноценного пайплайна необходимо дополнительно реализовать логику извлечения данных, их возможную трансформацию и загрузку, а также продумать механизм инкрементальности. Например, через поля insert_time или другие признаки изменения данных.

В типичном сценарии Airflow работает по batch-модели: с определённой периодичностью запускаются DAG’и, которые выполняют SELECT-запросы к PostgreSQL, извлекают данные и отправляют их в ClickHouse. Такой подход прост и прозрачен, но именно в этой простоте скрывается его основное ограничение. Ключевой минус заключается в том, что Airflow не умеет работать с WAL журналами. Он не может читать изменения напрямую из WAL, как это делает Debezium. Вместо этого вся работа строится на SQL-запросах, то есть фактически происходит регулярное опрашивание базы. Airflow в своей природе — это pull/оркестратор, а не event-driven система. Он либо работает по расписанию и сам не отслеживает изменения в БД.

На практике это привело бы к тому, что любая миграция через Airflow неизбежно создаёт дополнительную нагрузку на PostgreSQL. Даже если использовать инкрементальные выборки, ограничения по времени или читать данные с реплики, нагрузка никуда не исчезает — она лишь перераспределяется внутри кластера. При росте объёмов данных или увеличении частоты обновлений это начинает становиться заметным фактором. Кроме того, такой подход не даёт полноценного CDC. Мы не получаем поток изменений, а лишь периодические «срезы» данных. Это усложняет обработку часто изменяющихся сущностей, повышает риск пропуска или дублирования данных и требует дополнительной логики для обеспечения консистентности.

Ещё один момент, который стоит указать — это необходимость отдельной настройки и разработки пайплайна. В случае с Apache Airflow недостаточно просто “включить” перенос данных. Требуется создать и поддерживать DAG — Python-файл, в котором явно описывается вся логика: откуда забирать данные из PostgreSQL, по какому принципу определять изменения, как обрабатывать ошибки и каким образом загружать их в ClickHouse.

В итоге Airflow хорошо подходит для задач регулярной пакетной загрузки, где допустима некоторая задержка и нет строгих требований к отслеживанию каждого изменения. Однако в сценарии, где требуется аккуратный и малонагруженный перенос изменяющихся данных в аналитическое хранилище, его возможностей оказывается недостаточно без дополнительных усложнений.

PeerDB: новый, но практичный подход к CDC

PeerDB — это инструмент репликации данных из разными СУБД (Postgres, MySQL, MongoDB и др.) в другие системы (ClickHouse, Snowflake, BigQuery, S3, Kafka и др.) через CDC (Change Data Capture). Ключевая идея — отказаться от периодических выгрузок и перейти к доставке изменений почти в реальном времени (с задержкой в секунды).

Базовые сущности PeerDB

В PeerDB любая подключаемая система называется peer:

  • PostgreSQL-источник — это peer;

  • ClickHouse-приёмник — тоже peer.

Перед запуском репликации необходимо зарегистрировать оба peer’а — указать хост, порт и учетные данные. Mirror — это задача репликации. В ней задаются cписок таблиц, параметры партиционирования, политики применения данных. Именно mirror управляет всем жизненным циклом репликации.

Как работает репликация

Зеркало (mirror) запускается в два этапа:

1. Snapshot (только при первой загрузке)

PeerDB выполняет копирование текущих данных:

  • читает таблицы параллельными кусками;

  • работает в транзакции с уровнем REPEATABLE READ;

  • фиксирует LSN (Log Sequence Number) на момент старта.

Это гарантирует консистентный снимок данных.

2. CDC (чтение WAL) После завершения snapshot начинается потоковая обработка изменений:

  • PeerDB читает WAL через replication slot;

  • изменения применяются начиная с зафиксированного LSN;

  • в поток попадают INSERT, UPDATE, DELETE.

Дальше процесс становится непрерывным.

Replication Slot — механизм PostgreSQL, который не позволяет удалить WAL, пока его не прочитал подписчик. PostgreSQL постоянно пишет изменения в WAL. При этом старые WAL-файлы обычно удаляются автоматически. Чтобы данные гарантированно не были удалены раньше времени используется Replication Slot. Слот хранит позицию чтения LSN (Log Sequence Number) и не даёт удалить WAL раньше этой точки.

Существует два типа слотов. Physical — для физической репликации (для read-реплик) и Logical — логическая репликация (декодирует события : INSERT/UPDATE/DELETE). Именно логический используется для CDC, Debezium, pglogical, Airbyte, PeerDB.

Publication: что именно реплицировать

Publication определяет, какие изменения попадут в репликацию. Если Replication Slot отвечает на вопрос “где мы остановились в WAL”, то Publication отвечает на вопрос “какие изменения нас интересуют”. Они работают в паре: WAL (все изменения) → Publication (фильтр: что публиковать) → Replication Slot → Подписчик

Роль S3: буфер между источником и приёмником

В архитектуре PeerDB используется промежуточный слой — S3. Поток данных выглядит так: PostgreSQL → Replication Slot → S3 → ClickHouse.

Главная причина такой схемы — защита PostgreSQL. Если replication slot долго не потребляется, WAL-файлы накапливаются и могут вывести из строя БД. S3 всегда принимает данные из слота, даже если ClickHouse временно недоступен — например, на время обновления. Данные в этом случае накапливаются в S3 и применяются после восстановления.

Как данные ложатся в ClickHouse

Ранее в статье, при разборе особенностей работы ClickHouse, мы уже затрагивали ограничения, связанные с обновлениями и удалениями данных. Движки семейства MergeTree не предполагают классических UPDATE и DELETE, а любые изменения реализуются через добавление новых версий строк с последующей дедупликацией. В этом контексте PeerDB оказался удачным решением. Он изначально ориентирован на CDC-подход и учитывает специфику ClickHouse: изменения из PostgreSQL доставляются в виде событий, а на стороне приёмника преобразуются в модель с версиями строк (_peerdb_version) и флагом удаления (_peerdb_is_deleted). Такой подход органично ложится на ReplacingMergeTree и позволяет сохранить консистентность данных без дополнительных костылей в пайплайне. Дедупликация происходит на фоне через merge-процессы, а на уровне запроса через оператор FINAL.

Главные риски и проблемы

Конечно в любом подходе есть особенности. В PeerDB тоже есть. Replication slot должен постоянно читаться. Если этого не происходит — WAL начинает накапливаться. Сначала читает данные в S3, а затем применяет их в ClickHouse. Однако при высоком потоке данных возможна ситуация, когда скорость генерации WAL больше скорости обработки. В этом случае слот продолжит расти Это можно регулировать через параметр в PostgreSQL max_slot_wal_keep_size. Важно, по умолчанию он не задан. При достижении этого значения слот аннулируется. Слот нужно будет создавать заново и включать ресинхронизацию. Подробнее здесь и здесь.

PostgreSQL выступает сервером, PeerDB — клиентом. Если соединение обрывается, срабатывает wal_sender_timeout (по умолчанию ~10 минут) и mirror может перейти в ошибку. Со стороны PeerDB есть параметр PEERDB_POSTGRES_WAL_SENDER_TIMEOUT. По умолчанию — 120 секунд. Это довольно низкое значение. На практике его часто отключают (NONE), чтобы избежать ложных сбоев.

Отдельная проблема связана с отказоустойчивостью самого PostgreSQL. При переключении мастера (failover) логические replication slot’ы не переносятся на новый мастер. В результате PeerDB теряет позицию чтения WAL, репликация останавливается и требует ручного восстановления (resync). Это ограничение на стороне PostgreSQL. В более новых версиях механизмы работы со слотами улучшены, однако в managed-окружениях ситуация зависит от реализации. Например, в Yandex Cloud (Managed Service for PostgreSQL) автоматический перенос логических слотов не поддерживается на время написания статьи. Как отмечает поддержка Yandex Cloud:

«Сейчас такого функционала нет. При переключении мастера логический слот репликации автоматически не переезжает на новый мастер, и эта настройка не поддерживается в MDB PostgreSQL».

При локальном развертывании PeerDB могут возникать ошибки на этапе начальной загрузки (snapshot). При тестировании так и не удавалось справиться с ошибкой. Схожая ошибка описана тут. По наблюдениям проблема не воспроизводится в облаке Yandex Cloud. Вероятная причина - это сетевые ограничения или нестабильность.

Мониторинг

PeerDB предоставляет интерфейс, в котором можно отслеживать:

  • Lag replication slot;

  • состояние mirror;

  • прогресс синхронизации.

Это критично важно иметь такую наблюдаемость, так как именно lag показывает, успевает ли система обрабатывать поток изменений.

4. Подключение Superset

После того как данные начали стабильно поступать в ClickHouse, следующим шагом стало подключение BI-инструмента. На процессе развёртывания останавливаться не буду — это не тема статьи. Скажу лишь, что были некоторые нюансы, но без серьёзных сложностей.

Подключение ClickHouse к Superset прошло без проблем. Superset поддерживает подключение к различным СУБД, а также работу с файлами, поэтому интеграция сводится к настройке соединения. Отдельно стоит отметить слой подготовки данных. В Superset он реализован через понятие датасета — логического представления данных, поверх которого строятся графики и дашборды.

Существует два типа датасетов:

  • Физический — прямая привязка к таблице в источнике (в нашем случае — ClickHouse),

  • Логический (виртуальный) — SQL-запрос, который позволяет добавлять фильтры, объединять таблицы, скрывать технические поля.

Однако при работе с большими объёмами данных (миллионы строк и более) и сложными JOIN`ами становится критично, где именно выполняется обработка. В таких случаях лучше переносить тяжёлую логику на сторону ClickHouse, а не выполнять её в каждом запросе Superset. Здесь как раз помогает модель хранения ClickHouse на базе движков семейства MergeTree. На практике оптимальным подходом оказалось создание представлений (VIEW) в ClickHouse, которые осуществлять в потоке агрегацию данных.

В результате Superset используется как тонкий слой визуализации, а вся тяжёлая обработка остаётся на стороне ClickHouse.

5. Результаты и выводы

Что получили?

Удалось объединить события из Apache Kafka и объекты из PostgreSQL. Достигли ключевые нефункциональные требования, которые преследовали:

  • почти real-time данные (задержка — секунды);

  • минимальная нагрузка на PostgreSQL (за счёт чтения WAL через PeerDB);

  • поддержка двух потоков данных: CDC из PostgreSQL и стриминга из Kafka;

  • отсутствие классических batch-процессов;

  • устойчивость к сбоям ClickHouse (за счёт буферизации через S3 в CDC-потоке);

  • единая точка аналитики в ClickHouse, объединяющая разные источники.

Как итог нам удалось собрать единый инструмент, который позволяет анализировать данные под разными углами и объединяет несколько источников в единую аналитическую систему. Наш проект является связующим звеном между разрозненными системами данных. При этом даже команды, которые эти данные поставляют, не обладали полной картиной на уровне всей системы и качества передаваемых данных. Использование такого подхода позволило не только выстроить целостную аналитику внутри проекта, но и выявить ряд проблем — как на уровне собственных данных, так и на уровне процессов в компании в целом.

Чему мы научились?

В процессе построения архитектуры стало понятно, что ключевая сложность подобных систем заключается не столько в выборе конкретного инструмента, сколько в правильном разделении потоков и ответственности между ними. Мы на практике увидели, что CDC и событийный стриминг — это два разных класса задач, которые стоит обрабатывать раздельно. В нашем случае PostgreSQL через PeerDB и Kafka-поток решают разные задачи и могут сосуществовать параллельно, не конкурируя друг с другом.

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

Когда такой подход НЕ подойдёт?

В начале статьи я выделил задачи, которые стояли перед командой. Их мы смогли решить. На мой взгляд, получился достаточно удобный и практичный инструмент. Однако, несмотря на все преимущества, решение не является универсальным. Подход не подойдёт, если:

  • требуется единая событийная шина для всех источников данных;

  • нужны потоковые трансформации и обогащение событий в одном контуре;

  • важна строгая консистентность без допустимой задержки.

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