Выполнение запросов к Distributed таблицам Clickhouse

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

Настройка Distributed таблицы

Путь у нас уже есть три шарда с таблицей pageviews. Теперь нам необходимо на отдельном сервере:

• Настроить кластер в конфиге Clickhouse.
• Объявить distributed таблицу.
• Запилить пару запросов для проверки.

Детально процесс описан в статье по распределенному хранению. На сервере, на котором будем делать выборки, добавляем в конфиг (/etc/clickhouse-server/config.xml) список IP адресов шардов:

<**analytics**>

**10.62.195.100**

9000

**10.62.194.180**

9000

**10.62.194.148**

9000

</**analytics**>

## Три сервера шардов, с которых будет читать данные распределенная таблица

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

CREATE TABLE pageviews_analytics

(date Date, time DateTime, …, post_id String, session_id String)

ENGINE = Distributed(**analytics**, **mediabox**, **pageviews**)

## Распределенная таблица кластера analytics и базы данных mediabox

Запросы к распределенным данным

Теперь наша таблица готова к запросам:

SELECT count(*) FROM pageviews_analytics

Посчитаем количество записей на всех трех шардах:

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

1 rows in set. Elapsed: 3.290 sec. Processed 5.56 billion rows, 5.56 GB (1.69 billion rows/s., 1.69 GB/s.) 

Ну и что-то посложнее – посчитаем, например, количество устройств и просмотров, которые собирает [https://iotechnologies.com наш счетчик] по странам за 30 дней:

SELECT

uniq(user_id) AS devices,

count(*) AS pageviews,

country

FROM pageviews_analytics

WHERE date >= today() – 30

GROUP BY country

ORDER BY devices DESC

LIMIT 75

В результате увидим следующую таблицу:

┌──devices─┬─pageviews─┬─country────────┐
│ 39570941 │ 103196486 │ United States  │
│ 23291622 │ 172966086 │ Netherlands    │
│ 22805292 │ 117434407 │ Germany        │
│ 21708853 │ 155991547 │ Belgium        │
│ 19607740 │  90915385 │ United Kingdom │
...
│  5717729 │  28592470 │ Singapore      │
│  5151877 │  44360084 │ Chile          │
│  5119712 │  22587348 │ Turkey         │
└──────────┴───────────┴────────────────┘

75 rows in set. Elapsed: 284.743 sec. Processed 5.57 billion rows, 257.16 GB (19.55 million rows/s., 903.15 MB/s.) 

Ограничение по памяти для обработки

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

SELECT 
    count(*) AS total, 
    user_id
FROM pageviews_analytics 
GROUP BY user_id
ORDER BY total DESC
LIMIT 10

↗ Progress: 411.68 million rows, 15.22 GB (39.81 million rows/s., 1.47 GB/s.)
Received exception from server (version 18.14.19):
***Code: 241. DB::Exception: Received from localhost:9000,
::1. DB::Exception: Received from 10.62.194.180:9000.
DB::Exception: Memory limit (for query) exceeded:
would use 9.31 GiB (attempt to allocate chunk of 4194304 bytes),
maximum: 9.31 GiB. ***

0 rows in set. Elapsed: 10.700 sec. Processed 411.68 million rows, 15.22 GB (38.47 million rows/s., 1.42 GB/s.) 

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

<h2>TL;DR

Distributed движок в Clickhouse позволяет делать выборки из таблиц с разных серверов. При этом вам не нужно беспокоиться о склеивании результата, т.к. это как раз эффективно решает сама база данных.