ClickHouse настройка репликации и шардирования с нуля

[lonelygoblin]

Поднимаем ClickHouse для аналитики, данные — события пользователей, около 5 млрд записей в год. Нужна отказоустойчивость и возможность горизонтально масштабироваться. Как правильно настроить репликацию и шардирование? Пока у нас один сервер, ZooKeeper/ClickHouse Keeper есть. Версия 23.8 LTS.

[roylrs]

Для репликации в ClickHouse используй ReplicatedMergeTree вместо обычного MergeTree. Таблица создаётся с параметрами пути в ZooKeeper и именем реплики: ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}'). Макросы {shard} и {replica} задаются в config.xml для каждого сервера — это позволяет использовать одинаковый DDL на всех узлах.

[tomcruz]

Шардирование настраивается через конфигурацию кластера в remote_servers в config.xml (или в .d/ директории). Определяешь кластер с шардами и репликами, потом создаёшь Distributed-таблицу поверх ReplicatedMergeTree. Запросы к Distributed-таблице автоматически раскидываются по шардам и агрегируются.

[seniorsamurai]

Обязательно настрой ClickHouse Keeper вместо ZooKeeper если ещё не сделал — он встроен в CH начиная с 22.x, меньше зависимостей, проще операционно. Три узла Keeper для кворума — стандартная схема. Конфиг в clickhouse-keeper.xml, порт по умолчанию 9181.

[Austkin]

Развёрнуто про выбор ключа шардирования: это критически важно. Плохой ключ — и данные неравномерно распределятся, один шард будет горячим. Для событий пользователей часто берут cityHash64(userId) % num_shards или просто rand() если равномерное распределение важнее локальности. Если часто делаешь JOIN или GROUP BY по userId — имеет смысл шардировать по userId чтобы данные одного юзера лежали на одном шарде, тогда локальные агрегации будут быстрее. Distributed таблица: CREATE TABLE events_dist ON CLUSTER my_cluster AS events ENGINE = Distributed(my_cluster, default, events, cityHash64(userId)). Важно: INSERT делай в Distributed-таблицу или напрямую на каждый шард — вставка через Distributed добавляет небольшой overhead на async_insert, но удобнее. Для отказоустойчивости: 2 реплики на каждый шард минимум. При падении одной реплики ReplicatedMergeTree автоматически восстановит данные с живой реплики после поднятия.

[matguyvr]

Один практический совет: сначала запусти на одном шарде с двумя репликами, проверь что репликация работает, данные сходятся. Потом добавляй шарды. Переход с нереплицированного MergeTree на ReplicatedMergeTree возможен через ATTACH PARTITION FROM, но это отдельная история.

[archlover]

Мониторь через system.replication_queue и system.replicas — там видно отставание реплик, ошибки синхронизации. SELECT * FROM system.replication_queue WHERE is_currently_executing = 1 покажет что сейчас реплицируется. Если очередь растёт и не уменьшается — смотри логи и связь с Keeper.

[juniorstack]

Цифра для калибровки: 5 млрд событий в год — это в среднем ~160 строк в секунду. Для ClickHouse это не нагрузка вообще, одна машина с NVMe такое жуёт не просыпаясь. Я бы не городил шарды на старте: один шард + две реплики, но в remote_servers сразу опиши кластер и заведи Distributed-таблицу — тогда добавление шардов потом не потребует менять схему и код. Шардироваться стоит, когда упрётесь в диск или latency тяжёлых запросов, а не заранее.

[Roost66]

@Austkin, плюсую про ключ и добавлю жирный нюанс: решардинг в ClickHouse — ручная боль, автоматического ребаланса нет, смена ключа потом означает перелив всех данных через INSERT SELECT. Поэтому cityHash64(userId) почти всегда лучше rand(): получаешь локальные JOIN и GROUP BY по юзеру, а равномерность у хеша и так приличная. rand() я бы брал только если запросов по userId точно не будет.

[tsav]

@seniorsamurai, про Keeper поддержу, но с оговоркой: не ставьте его на те же диски, что и нагруженные CH-ноды. Keeper чувствителен к latency fsync, и когда мержи забивают диск, начинаются таймауты сессий — реплики уходят в readonly, очередь репликации растёт. Либо три отдельные маленькие виртуалки, либо хотя бы выделенный диск под лог координации.