Всем привет!
Меня зовут Валерий Локтаев, я backend-разработчик сервиса биллинга в CloudMTS.
В этой статье я расскажу, как насовсем убрать дублирующие записи в ClickHouse (CH). Логичный вопрос — откуда вообще взялась проблема? Можно взять движок таблицы ReplacingMergeTree, указать ORDER BY в качестве ключа дедупликации, и CH чудесным образом удалит все дубли в базе.
ReplacingMergeTree, безусловно, отличное решение. Но представьте, что ваша задача — сделать так, чтобы в таблице дубли никогда не появлялись, даже на несколько секунд.
Далее я расскажу, в каких случаях это необходимо и какое решение удалось подобрать.
Контекст: когда дубли становятся проблемой
CloudMTS использует CH в сервисе отчетов и предоставления клиентской детализации потребления ресурсов и их стоимости. Пользователь может зайти в личный кабинет и на графике увидеть все подробности потребления облачных ресурсов: по проектам, услугам, за конкретный период времени.
В личном кабинете потребление облачных мощностей подробно детализируется:
График формируется за счет агрегации метрик потребления.
Метрика потребления — уникальная сущность, которую биллинговая система получает от интегрированных с ней продуктов. В метрике находится информация о времени использования ресурса, количестве потребления, метаданные для идентификации пользователя и уникальный идентификатор.
Когда биллинговая система получает метрику, она процессит ее по пайплайну, где на каждом этапе над метрикой производятся какие-то действия, вычисления и т. д. В конечном итоге метрика попадает в сервис детализации.
Для общения между сервисами мы используем Kafka at-least-once, что может привести к появлению дублей. В свою очередь дублирующие записи ведут к искажению стоимости потребляемых услуг: если появится дубль, в личном кабинете сумма станет больше, чем пользователь потратил на самом деле. Такую ситуацию мы допустить не можем.
Поиск методов борьбы с дублированием данных
Изначально задача звучала так, что в теории у нее должен быть миллион решений. На практике для наших вводных ничего готового не существовало.
Обратившись к документации CH, отмечу следующие особенности движка ReplacingMergeTree:
- гарантируется отсутствие дублей во вставке (т. е. если в батче данных, которые летят на вставку, есть дубли — движок их удалит, в таблицу заедут уникальные значения);
- не гарантируется отсутствия дублей в таблице в определенный момент времени. Задача очистки от дублей происходит в фоне, в неопределенный неизвестный нам момент времени (CH лишь «обещает», что когда-нибудь дубли удалит);
- есть ручной вызов задачи по дедупликации с помощью OPTIMIZE (движок не начнет дедупликацию сразу, а лишь запланирует процесс, который когда-нибудь начнет);
- НЕТ точных гарантий, что дубли вообще будут удалены.
Такое решение нам точно не подходит.
Проведя ресерч, я нашел несколько реализаций, которые… почти работали, — всё равно так или иначе просачивались дубли.
Очевидный вариант — использовать движок Kafka, который напрямую может подключаться к брокеру, вычитывать сообщения и складывать в таблицу. К сожалению, движок не обеспечивает стратегию exactly once, что в итоге приводит к появлению дублей.
Следующая идея — развязочная таблица. Решение заключается в следующем:
- Создаем таблицу Table 1, куда будем вставлять записи из «кафки». Мы не доверяем предыдущим в цепочке сервисам и проводим дедупликацию на своей стороне.
- Создаем развязочную таблицу Table 2, в которой будут храниться дедуплицированные ID уже вставленных в основную таблицу записей (в виде cityHash64).
- Создаем конечную таблицу Table 3, куда будут вставлены итоговые дедуплицированные записи.
- Создаем Materialized View (MW1), задача которой — при вставке новых записей в Table 1 проверить, нет ли записей с таким же ID в таблице Table 2. Если есть дублирующиеся записи, то их отбросим. Все остальные уникальные записи вставляются в таблицу Table 3.
Звучит хорошо, но давайте разберемся, какие в схеме есть подводные камни.
Мы должны хранить ID всех уже вставленных записей в таблице Table 2. По мере роста количества записей замедляется время каждой вставки, так как каждый раз мы итеративно сверяем все записи друг с другом. Подразумевая в перспективе бесконечное количество записей, получаем бесконечное время на вставку.
Почему бы не сохранить только записи, которые пришли за последний расчетный период (день, неделю, месяц)? В таком случае существенно сократится количество проверок из каждой новой вставки. И действительно, это предположение валидно, мы могли так сделать. Вот только дополнительно мы закладывались на обработку аварийных случаев, когда нужно одним разом переотправить все метрики за неизвестный временной отрезок, чтобы иметь возможность что-то исправить, перерассчитать и т. д. В таком случае невозможно спрогнозировать, какое количество метрик будет в таблице и сколько времени займет вставка.
Вторая проблема заключается в параллельной вставке в таблицу Table 1 двух батчей, имеющих одинаковые записи. Для первой и второй вставки будет справедливо состояние таблицы Table 2, в котором нет повторяющегося в этих вставках ID. В таком кейсе мы все равно получаем дубли.
Неожиданное решение: две СУБД
Потратив достаточно времени на генерацию идей и танцев с бубном, я пришел к выводу, что нужно вынести таблицу Table 2 в инструмент, который хорошо умеет работать с удалением дублей за счет возможности строить уникальные индексы. Этот инструмент — MongoDB.
А зачем нужна еще какая-то зависимость, если можно все вставленные записи хранить в памяти сервиса, в кеше, и дедуплицировать их из кеша? Не забывайте, что наша задача — иметь возможность дедупликации за весь период. Мы не знаем конечное количество записей и, соответственно, не можем предположить конечное количество памяти.
В результате получилась следующая схема:
- Создаем коллекцию в MongoDB, где сохраняем только ID всех вставленных метрик. По этому же полю строим уникальный индекс.
- Создаем конечную таблицу в CH, где будут храниться метрики в дедуплицированном виде.
- При новой вставке в рамках транзакции вставляем записи в коллекцию MongoDB. Все записи, которые удалось вставить, считаем уникальными и инсертим их в основную таблицу CH. Всё, что не удалось вставить из-за отсутствия уникальности, — это дубли, которые мы игнорируем.
Такое решение хранит абсолютно все когда-либо вставленные ID записей в CH и позволяет дедуплицировать по ним за любой период времени. При этом затрачиваемое на дедупликацию время растет логарифмически, а не линейно.
В решении нет конкретной привязки к MongoDB. Команда выбрала ее, потому что имеет хороший опыт взаимодействия с этой СУБД и она подходит для решения подобной задачи. Если вы знаете другую технологию быстрой дедупликации, расскажите в комментариях.