23.11.2025 08:15
flow_data 0 359 Источник

В этой статье мы поговорим о том, как реализовать поведение атомарной вставки в ClickHouse. Рассмотрим несколько вариантов, подсветим их сильные и слабые стороны, а также, когда каждый из них применять.

Задача

Мы хотим добиться, чтобы не было случаев, когда мы начали вставку, а пользователь прочитал данные до её завершения и получил неактуальный (неполный) набор данных.

Неатомарная вставка = риск чтения некорректного набора данных.

Сценарии, когда такое может произойти:

  1. Удалили партицию и хотели начать вставлять данные взамен удалённой, но пользователь обратился к этому интервалу.

  2. Вставка данных оборвалась (из-за проблем с сетью, например).

  3. Вставляем данные батчами, и пользователь проверил наличие данных, но не проверил, что все батчи на месте — прочитал часть данных.

Перейдём к реализации атомарной вставки в ClickHouse.

Вариант 1. Настройка min_insert_block_size_rows

Подход описан в статье от Altinity.


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

Метод сработает, если при вставке мы создадим только 1 парт (какие критерии этого можно прочитать в статье)

Скрипт для реализации подхода 
-- Создаем целевую таблицу
drop table if exists core.data;
create table if not exists core.data 
(
	num_int Int64, 
	partiton_num String
) 
engine = MergeTree
order by num_int;


-- Вставляем данные
insert into core.data 
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	intDiv(number, 1_000_000) as partiton_num
from
	numbers(100_000_000);


-- Видим что партов больше чем 1
SELECT 
    count(),
    min(rows),
    max(rows),
    sum(rows)
FROM system.parts
WHERE (level = 0) AND (table = 'data');

-- count()|min(rows)|max(rows)|sum(rows)|
-- -------+---------+---------+---------+
--      96|   397345|  1048449|100000000|




-- Пересоздаем целевую таблицу
drop table if exists core.data;
create table if not exists core.data 
(
	num_int Int64, 
	partiton_num String
) 
engine = MergeTree
order by num_int;


-- Вставляем данные но уже с настройкамии
  -- input_format_parallel_parsing=0,
  -- min_insert_block_size_bytes=0,
  -- min_insert_block_size_rows=1000000000
-- так мы задаем что если строк меньше чем 1_000_000_000 
-- то данные будут вставлены единым партом
insert into core.data 
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	intDiv(number, 1_000_000) as partiton_num
from
	numbers(100_000_000)
settings 
	input_format_parallel_parsing=0,
    min_insert_block_size_bytes=0,
    min_insert_block_size_rows=1_000_000_000;


-- Видим что создался только 1 парт  
SELECT 
    count(),
    min(rows),
    max(rows),
    sum(rows)
FROM system.parts
WHERE (level = 0) AND (table = 'data');

-- count()|min(rows)|max(rows)|sum(rows)|
-- -------+---------+---------+---------+
--       1|100000000|100000000|100000000|

Кажется что можно остановиться на этом варианте, но как писал выше у него есть минусы и зоны применения.

Плюсы:

  • Лаконичная и простая реализация

Минусы:

  • Высокое потребление памяти при вставке, ниже видем что даже на "игрушечном" примере разница значительная (в 10 раз). Вызвано это тем что весь парт предварительно пишется в оперативную память

    event_time         |memory_usage_gb|query             |
-------------------+---------------+------------------+
2025-11-06 17:02:17|            3.5|with settings     |
2025-11-06 17:04:13|          0.034|with out settings |

  • Если захотим вставлять данные батчами, то вариант не подойдет

Когда применять:

  • Небольшие таблицы

  • Единичная вствка в одну партицию

  • Если скорость разработки у вас в приоритете, и вы готовы принять, то что будет использоваться оперативная память на вставку

Вариант 2. Временная таблица + переименование

Вариант заключается в том, что мы вставляем новые данные во временную таблицу, после чего меняем местами названия таблиц (временная становится основной, а основная — временной).

Возможно это благодаря операции exchange в ClickHouse.

exchange tables table1 and table2
Скрипт для реализации подхода
-- Создаем целевую таблицу
drop table if exists core.data;
create table if not exists core.data 
(
	num_int Int64, 
	partiton_num String
) 
engine = MergeTree
order by num_int;


-- Наполняем целевую таблицу данными
insert into core.data 
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	intDiv(number, 1_000_000) as partiton_num
from
	numbers(100_000_000);


-- Видим что в ней числа от 0 до 99_999_999
select 
	min(num_int),
	max(num_int),
from core.data;


-- Создаем временную таблицу
drop table if exists stage.data_tmp;
create table if not exists stage.data_tmp
(
	num_int Int64, 
	partiton_num String
) 
engine = MergeTree
order by num_int;


-- Вставляем в нее числа в диапазоне 100_000_000 - 200_000_000
insert into stage.data_tmp
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	intDiv(number, 1_000_000) as partiton_num
from
	numbers(100_000_000, 100_000_000);


-- Меняем местами временную и целевую таблицы
exchange tables core.data and stage.data_tmp;


-- Видим что в целевой обновились данные
-- и произошло это абсолютно атомарно
select 
	min(num_int),
	max(num_int)
from core.data;

Плюсы:

  • Достаточно простая и понятная реализация.

  • Можно откатиться, если снова переименовать таблицы (восстановить предыдущее состояние).

  • Нет затрат на дополнительную оперативную память.

Минусы:

  • Нужна дополнительная таблица, в которой необходимо поддерживать консистентность полей с основной.

  • Дополнительное место на диске (можно очищать tmp после переименования, и тогда этот минус почти незначительный).

Когда применять:

  • Таблицы ещё относительно небольшие, их можно полностью пересобрать.

Вариант 3. Временная таблица + replace партиции

Если мы не готовы полностью перезаписывать всю таблицу, мы можем доработать подход из варианта 2 и перезаписывать только нужную партицию.

Операция replace partition в ClickHouse работает атомарно, то есть партиция обновится сразу, а не отдельными блоками.

Скрипт для реализации подхода
-- Создаем целевую таблицу
-- Партиционируем таблицу (в 1й партиции миллион строк)
drop table if exists core.data;
create table if not exists core.data 
(
	num_int Int64, 
	partiton_num String
) 
engine = MergeTree
partition by partiton_num
order by num_int;


-- Наполняем целевую таблицу данными
-- от 1 до 100М
insert into core.data 
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	intDiv(number, 1_000_000) as partiton_num
from
	numbers(100_000_000);


-- Видим что в партиции с id '99' 
-- дежит диапазон 99М - 100М
select 
	min(num_int),
	max(num_int)
from core.data
where partiton_num = '99';


-- Создаем временную таблицу
drop table if exists stage.data_tmp;
create table if not exists stage.data_tmp
(
	num_int Int64, 
	partiton_num String
) 
engine = MergeTree
partition by partiton_num
order by num_int;


-- Для временной таблицы заполняем только партицию с id '99' 
-- Причем делаем диамазон 100М - 101М
insert into stage.data_tmp
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	'99' as partiton_num
from
	numbers(100_000_000, 1_000_000);


-- Проверяем что вставился именно диапазон 100М - 101М
select 
	min(num_int),
	max(num_int)
from stage.data_tmp
where partiton_num = '99';


-- Делаем атомарную операцию replace partition
-- Партиция '99' копируется из stage.data_tmp и заменяется в core.data
alter table core.data replace partition '99' from stage.data_tmp;


-- Видим что в целевой таблице данные изменены
-- В партиции '99' диапазон 100М - 101М
select 
	min(num_int),
	max(num_int)
from core.data
where partiton_num = '99';

Плюсы:

  • Перезапись короткого интервала, что чаще используется, чем обновление всей таблицы.

  • Нет затрат на дополнительную оперативную память.

Минусы:

  • Нужна дополнительная таблица, в которой необходимо поддерживать консистентность полей с основной.

  • Дополнительное место на диске (можно очищать tmp после переименования, что уменьшает минус).

Когда применять:

  • В случаях, когда используется стратегия обновления через перезапись партиций.

Вариант 4. View с фильтрами

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

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

Подход:

  1. Имеем таблицу и view поверх неё.

  2. Во view используем фильтр, который выбирает только актуальные данные. Например, фильтр может быть таким: partiton_num < 100.

  3. Добавляем новую партицию данных insert'ом в целевую таблицу.

  4. Пересоздаём view с обновлённым фильтром.

Скрипт для реализации подхода
-- Создаем целевую таблицу
drop table if exists core.data;
create table if not exists core.data 
(
	num_int Int64, 
	partiton_num Int16
) 
engine = MergeTree
partition by partiton_num
order by num_int;


-- Наполняем целевую таблицу данными
-- от 1 до 100М
insert into core.data 
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	intDiv(number, 1_000_000) as partiton_num
from
	numbers(100_000_000);


-- Создаем view для целевой таблицы
create or replace view core.v_data as
select
	num_int, 
	partiton_num
from core.data
where partiton_num < 100;


-- Видим что доступны данные в интервале 
-- от 1 до 100М
select 
	min(num_int),
	max(num_int)
from core.v_data;


-- Вставляем еще блок данных в диапазоне 100М - 101М
-- Для наглядности делаем это в 2 подхода


-- 1й батч
insert into core.data
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	'100' as partiton_num
from
	numbers(100_000_000, 500_000);


-- 2й батч
insert into core.data
(
	num_int, 
	partiton_num
)
select
	toInt64(number) num_int,
	'100' as partiton_num
from
	numbers(100_500_000, 500_000);


-- Видим что пока нам все еще доступен интервал 
-- от 1 до 100М
select 
	min(num_int),
	max(num_int)
from core.v_data;


-- Пересоздаем view и тем самым предоставляя пользователю новые данные
create or replace view core.v_data as
select
	num_int, 
	partiton_num
from core.data
where partiton_num < 101;


-- Видим что доступны данные до 101М
select 
	min(num_int),
	max(num_int)
from core.v_data;

Плюсы:

  • Нет дополнительных затрат на дисковое пространство.

  • Легко реализовать, если view уже используется.

Минусы:

  • Не самый гибкий подход. Если требуется перезапись существующих данных, придётся дважды обновлять view для одной вставки (до и после), что приведёт к большим объёмам DDL-операций на кластере.

Когда применять:

  • Если view уже используется для работы с данными.

  • Если фильтр выбора актуальных данных не слишком сложный (подходит для таблиц, где данные в основном только добавляются и редко перезаписываются).

Примечания и советы:

  • Используйте insert_quorum (https://clickhouse.com/docs/operations/settings/settings#insert\_quorum). Без этой настройки есть риск, что данные запишутся только на одну реплику, а затем чтение будет происходить с другой, на которую данные ещё не успели реплицироваться.

Спасибо что прочитали до конца

Приглашаю вступить в мой Telegram-канал - flow_data, там я буду публиковать материалы и заметки по Data Engineering.

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