ClickHouse не тормозит, но теряет данные. Часть 2 — от буферных таблиц к Kafka Engine

Продолжаем тему потери данных в ClickHouse. Рекомендую прочитать первую статью цикла, в которой раскрывается проблема атомарности INSERT и дедупликация (и, конечно же, решение проблемы потери данных). А тема текущего обсуждения — буферные таблицы и то, как они теряют данные.

P.S. читать каждую ссылку из статьи совсем не обязательно, основные тезисы из ссылок изложены в статье.

1. Введение

Как известно, ClickHouse не любит, когда в него делают 1 000 000 вставок по одной записи — он любит, когда делают одну вставку на 1 000 000 записей. Об этом сказано в официальной документации, а еще об этом рассказывал лично Алексей Миловидов (основатель ClickHouse) на одной из конференций, стенограмму которой можно почитать в статье на Хабре. Краткая выжимка из статьи: все кейсы (кроме Kafka), которые в ней описаны — ненадежные способы решения проблемы батчинга данных перед вставкой. Как правило, все эти методы сводятся к сбору информации от 10 000 операций INSERT в некий буфер, который будет раз в секунду делать INSERT в ClickHouse. Нас же из всего этого интересует именно то, как сам ClickHouse может потерять данные и как этого избежать, поэтому рассматривать будем именно способ с буферными таблицами.

Итак, описываем кейс: у вас есть стримы, генерирующие большое количество вставок данных в ClickHouse в секунду (условно 10 000 операций INSERT в секунду). Вы натравили стрим на обычную MergeTree и получаете следующее (цитата из выступления Алексея Миловидова): «Что будет, если делать плохо? Вставляем по одной строке в таблицу MergeTree и получается 59 строк в секунду. Это в 10 000 раз медленнее. В ReplicatedMergeTree – 6 строк в секунду. А если еще кворум включится, то получается 2 строки в секунду. По-моему, это какой-то кромешный отстой. Как можно так тормозить? У меня даже на футболке написано, что ClickHouse не должен тормозить. Но тем не менее бывает иногда.»

Помимо невероятного торможения, конечно же, получим еще и перегруженность железа: под капотом отрабатывает слияние кусков данных (собственно поэтому движки таблиц и имеют в названии Merge, где есть Merge — там присутствует процесс слияния данных). Проще говоря: каждая операция INSERT в таблицу, содержащую в названии движка *Merge*, вставляет данные не в файл целевой таблицы, а в небольшие кусочки данных, хранящиеся рядом с основным файлом таблицы. И под капотом с определенным интервалом, задающимся настройкой max_merge_selecting_sleep_ms, запускается процесс слияния всех кусков данных в один файл данных (как правило, есть крупный файл таблицы/партиции и мелкие файлики от кучи инсертов, которые мержатся с этим крупным файлом и удаляются). Именно процесс слияния и будет убивать производительность вашего ClickHouse.

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

2. Буферные таблицы

Идея буферной таблицы очень простая — она пишет данные для записи в оперативную память, периодически сбрасывая их в целевую таблицу. Условия для сброса данных вы пишите сами, например: либо по истечении 100 секунд с момента предыдущего сброса данных на диск, либо по накоплению миллиона записей, либо по накоплению 1ГБ данных. Очень удобно + из коробки. 10 000 операций INSERT попадут не в файлы на диске (которые потом еще и слить надо с основной таблицей), а в оперативную память, и затем в один-единственный файл на диске вместо кучи файлов, что существенно снизит нагрузку при операции слияния.

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

Как же не потерять данные? Самый правильный, на мой взгляд, способ — использовать Kafka (как бы не хотелось добавлять еще одну распределенную систему) в связке с Kafka Engine. Данный способ хорошо зарекомендовал себя на рынке.

3. Kafka Engine

В рамках статьи, конечно, не получится рассказать обо всем. Поэтому оставлю ссылки на полезные статьи, которые позволят теоретически освоить как Kafka, так и Kafka Engine.

• Изучаем Kafka — Уровень 1 / Хабр

• Kafka | ClickHouse Docs

• ClickHouse® Kafka Engine Tutorial — Altinity | Run open source ClickHouse® better

• ClickHouse® Kafka Engine FAQ — Altinity | Run open source ClickHouse® better

А мы же сконцентрируемся именно на решении проблемы потери данных. Нам нужно доказательство атомарности технологии Kafka Engine. Для этого подготовил демонстрационный кейс, рассмотрим его.

Конфигурация максимально простая: докер, одна нода Kafka, одна нода Zookeper, одна нода ClickHouse

В Kafka создан тестовый топик с названием test_topic_json, в котором сообщение в формате json с одним полем message и строковым значением (на изображении плохо видно, продублирую текстом)

В ClickHouse создан классический пайплайн Kafka Engine:

drop table if exists test_kafka_json_parse; create table test_kafka_json_parse (     message String )     engine = Kafka     SETTINGS kafka_broker_list = 'kafka:29092'         , kafka_topic_list = 'test_topic_json'         , kafka_group_name = 'my-group4'         , kafka_format = 'JSONEachRow'         , kafka_max_block_size = 2097152 ;  drop table if exists message_json_parse; create table message_json_parse (     message String ) engine = MergeTree order by message ;   drop view if exists test_kafka_json_parse_view; create materialized view test_kafka_json_parse_view TO default.message_json_parse             (              `message`  String                 ) AS SELECT message FROM test_kafka_json_parse ;

Что этот пайплайн делает: читает свежие данные из Kafka из топика test_topic_json . Коннект осуществляется благодаря таблице test_kafka_json_parse, материализованное представление test_kafka_json_parse_view является консьюмером, message_json_parse — конечная таблица, в которую мат вьюха (консьюмер) пишет данные.

Реализовав все выше изложенное, получаем следующий результат: в конечной таблице видим оба сообщения, присутствующие в Kafka

Также видим следующе значение offset для консьюмера my-group4 (продублирую текстом — всего сообщений 2, offset тоже равен 2):

Получается, что консьюмер my-group4 вычитал полностью топик test_topic_json, что абсолютно верно.

Теперь необходимо доказать атомарность. Для этого пересоздадим конечную таблицу message_json_parse таким образом:

drop table if exists message_json_parse; create table message_json_parse (     message UInt64 ) engine = MergeTree order by message ;

message UInt64 — ключевой момент (UInt64 — строго положительное число). Мы специально ломаем конечную таблицу, чтобы мат вьюха не могла записать в нее данные. Затем закидываем новое сообщение в топик test_topic_json, пускай будет

При всем при этом мы НЕ трогаем ни Kafka-таблицу, ни мат вьюху. И вот что получаем в таком случае:

Видим, что теперь для консьюмера my-group4 Offset не равен End. Сообщений в топике 3, но смещение у консьюмера равно 2. Очевидно, что мат вьюха не может «gg» записать в колонку с типом UInt64. Посмотрим на лог ошибок клика, который даст ответы на все вопросы:

Продублирую текстом самое важное и очевидное: Cannot parse string 'gg' as UInt64. А о чем же суммарно с несмещенным offset нам это говорит? — это говорит о том, что offset коммитится только при успешной отработке материализованного представления, что и гарантирует атомарность.

Из кейса выше очевидно, что ClickHouse прочитал сообщение «gg» из топика. Это было важно доказать в контексте потери данных. Мы понимаем, что операция чтения из Kafka была осуществлена, данные физически залетели в ClickHouse (в RAM). Затем происходят бесконечные попытки записать сообщение в таблицу. К новым сообщениям консьюмер, конечно же, не перейдет (хотя и есть настройки, конфигурирующие это — на практике не встречал и не советую).

Далее можно делать что угодно: положить ClickHouse, рестартнуть докер/машину, да хоть удалить ClickHouse и проделать все манипуляции заново, главное — не трогать Kafka, так как она бережно хранит в себе offset для нашего консьюмера.

Ну а мы же поправим ошибку, чтобы убедиться, что после этого сообщение будет успешно прочитано:

drop table if exists message_json_parse; create table message_json_parse (     message String ) engine = MergeTree order by message ;

И закономерный результат:

Также видим закономерный результат и в Kafka (End=3, Offset=3):

Получается, что после решения проблемы чтение продолжилось, очевидно, с последнего offset, и корректная работа была восстановлена БЕЗ потерь данных. Тоже самое будет и в любой другой ситуации: падение сервера, ошибки коллег, отклонение от контракта данных в Kafka и т.д.

P.S. Если есть желание, то можно посмотреть исходный код консьюмера.

4. Заключение

Далеко не факт, что вы столкнетесь с ситуацией «10000 INSERT Per Second». Но всегда помните следующее:

• Автор большой фанат ClickHouse и хочет, чтобы ClickHouse не только не тормозил, но и не терял данные.

• ClickHouse не любит, когда в него делают 1 000 000 вставок по одной записи — он любит, когда делают одну вставку на 1 000 000 записей.

• Все способы буферизации (включая асинхронные вставки) чреваты потерей данных.

• Kafka Engine — проверенная временем технология, не теряющая данные.

• Если есть возможность формировать крупные пачки данных надежным способом, отличным от Kafka Engine — то это тоже решение проблемы.

Пользуйтесь на здоровье!