Распределенное хранение данных в Clickhouse

Рано или поздно объемы (аналитических) данных выходят за пределы возможностей одного сервера. В любом случае, если объем данных больше, чем 10Тб, пора думать о распределенном хранилище.

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

# Общий объем данных: 50Тб, 10 млрд записей
Сервер 1: хранит 5Тб, 1 млрд записей от 1 до 1 млрд
Сервер 2: хранит 5Тб, 1 млрд записей от 1 + 1 млрд до 2 млрд
Сервер 3: хранит 5Тб, 1 млрд записей от 1 + 2 млрд до 3 млрд
...

## Распределение данных между серверами

Т.е. на каждом сервере будет хранится кусок таблицы (еще называются шардами). Но в этом случае усложняется работа с таблицей. Ведь как сделать выборку, которая обработает сразу все данные?

Для этого в Clickhouse есть специальный движок таблицы – Distributed. Он как раз адресует обозначенную проблему. Вместо функций обычной таблицы, он отправляет запросы на разные сервера и собирает с них данные. Потом собранные данные дополнительно обрабатываются уже локально (например, группируются) и результаты отдаются клиенту.

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

Создание шардов

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

CREATE TABLE events (

date Date, # дата

time DateTime, # время

event String, # название события

client String, # уникальный идентификатор клиента

value UInt32 # цифровое значение события

) ENGINE = MergeTree(date, (event, client), 8192)

## Структура таблицы – одинаковая на каждом сервере

Мы распределим таблицу на 4 сервера:

10.135.77.56
10.135.77.80
10.135.77.58
10.135.77.66

## Сервера для распределенного хранения таблицы events

На каждом сервере необходимо добавить локальный IP адрес в конфигурацию хостов Clickhouse /etc/clickhouse-server/config.xml:

…

10.135.77.56

…

## для каждого сервера нужно указать его IP адрес

После этого нужно перезапустить сервер:

/etc/init.d/clickhouse-server restart

Распределенная таблица

На отдельном сервере (с которого будем слать запросы) настроим кластер, который указывает на сервера-шарды. Для этого в файле настроек (/etc/clickhouse-server/config.xml) нужно добавить в блок <remote_servers> следующее:

**

10.135.77.56

9000

10.135.77.80

9000

10.135.77.58

9000

10.135.77.66

9000

**

## Настройка кластера на сервере, с которого будем слать запросы

После перезапуска сервера можно создавать распределенную таблицу:

CREATE TABLE events (

date Date,

time DateTime,

event String,

client String,

value UInt32

) ENGINE = **Distributed(events, default, events)**

## Создание распределенной таблицы

Как видно, структура таблицы такая же, как и на шардах. Кроме определения движка, в котором:

Distributed(**events**, default, events) # название кластера из конфига

Distributed(events, **default**, events) # название БД на шардах

Distributed(events, default, **events**) # название таблицы на шардах

Вставка данных

Для вставки данных есть две опции. Первая – вставлять прямо в Distributed таблицу, но это довольно тупо, сталкиваемся со SPOF да и много ограничений в этом случае. Вторая – вставлять данные прямо на шарды. В этом случае будет и распределение нагрузки и ручной контроль логики распределения.

Попробуем вставить тестовые данные на любом шарде:

INSERT INTO events VALUES(today(), now(), ‘test’, ‘Den’, 2);

## выполняем на любом шарде

Теперь на основном сервере с распределенной таблицей выполним запрос:

**SELECT *
FROM events**

┌───────date─┬────────────────time─┬─event─┬─client─┬─value─┐
│ 2018-05-06 │ 2018-05-06 10:36:59 │ test  │ Den    │     2 │
└────────────┴─────────────────────┴───────┴────────┴───────┘

1 rows in set. Elapsed: 0.008 sec. 

## Данные собраны со всех шардов

Распределенные запросы

Для начала нагенерим побольше данных простым скриптом:

for ( $i = 0; $i < 10000; $i++ )

{

$tsv .= implode(“t”, [

date(‘Y-m-d’),

date(‘Y-m-d H:i:s’),

substr(str_shuffle(“qwertyuiopasdfghjklzxcvbnm”),0,5),

uniqid(),

mt_rand(1, 1000)

]) . “n”;

}

for ( $i = 0; $i < 1000; $i++ )

{

echo ‘.’;

file_put_contents(‘/tmp/events.tsv’, $tsv, FILE_APPEND);

}

for ( $i = 0; $i < 25; $i++ )

{

echo ‘+’;

exec(‘cat /tmp/events.tsv | clickhouse-client –query=”INSERT INTO events FORMAT TSV”‘);

}

## Вставим 250 млн тестовых записей на каждый шард

Теперь выберем суммарное количество записей с распределенной таблицы:

SELECT count(*)
FROM events 

┌────count()─┐
│ 1010000001 │
└────────────┘

1 rows in set. Elapsed: 0.099 sec. Processed 1.01 billion rows, 2.02 GB (**10.23 billion rows/s., 20.46 GB/s.**) 

## Запрос к Distributed таблице

Если такую же выборку сделать прямо на шарде, увидим что локальная скорость обработки меньше (что и понятно, ведь 4 сервера выполняют операции параллельно):

SELECT count(*)
FROM events 

┌───count()─┐
│ 250000000 │
└───────────┘

1 rows in set. Elapsed: 0.099 sec. Processed 250.00 million rows, 500.00 MB (***2.53 billion rows/s., 5.07 GB/s.***)

## Запрос прямо с шарда

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

SELECT 
    client, 
    count(*) AS total
FROM events 
WHERE event = 'yfxlb'
GROUP BY client
ORDER BY total ASC
LIMIT 10

┌─client────────┬─total─┐
│ 5aeedffa3d8ec │ 25000 │
│ 5aeedff0249a6 │ 25000 │
└───────────────┴───────┘

2 rows in set. Elapsed: **0.024 sec**. Processed 294.91 thousand rows, 10.62 MB (12.14 million rows/s., 437.13 MB/s.) 

## Выборки в Distributed таблицах работают очень быстро

Исключения могут составить некоторые ситуации с JOIN’ами, поэтому их лучше обходить.

<h2>TL;DR

Движок Distributed позволяет организовать распределенное хранение таблиц на основе Clickhouse, сохраняя возможность выборок, как в обычных таблицах. Используйте этот движок, если вам необходимо хранить огромные таблицы, которые не умещаются в один сервер.