Как мы «переселяли» свою БД Clickhouse в удаленные центры обработки данных
В конце прошлого года мы писали о сложном процессе переезда в новый дата-центр в Детройте. Одной из самых нетривиальных задач стало «переселение» и Clickhouse. Напомню, что речь идет о высоконагруженном сервисе, который обслуживает десятки серверов и обрабатывает сотни тысяч запросов с максимально низкой задержкой.
В этой статье мы расскажем Вам о том, как именно мы переносили данные, не имея возможности отключить службу или использовать автоматическую репликацию.
Объем данных для нас не был проблемой, однако сам процесс переноса БД оказался очень ресурсоемким. В открытых источниках представлен минимум информации о механизмах, которые мы использовали. Основным нашим инструментом стала утилита clickhouse-copier (https://github.com/ClickHouse/copier), в этой статье мы хотим поговорить с Вами именно о ней (в том числе, о конкретных примерах ее использования, скриптах и командах).
Самый простой подход
Наш сервис работает круглосуточно, а это значит, что мы не можем просто взять и выключить его, а потом перенести все данные, используя функцию копирования. Мы составили план переезда БД, но ввиду некоторых ограничений самого Clickhouse (точнее, его схемы репликации) не смогли его реализовать полностью.
Изначально мы предполагали, что сможем подключить реплику из нового дата-центра к каждому шарду в старом, дождаться синхронизации, а затем просто отключить реплики в старом дата-центре. Однако из-за географической удаленности мы столкнулись со слишком большой задержкой при передаче данных между центрами обработки данных, несмотря на скорость 2-2,5 Гбит/с. В результате объем данных в ZooKeeper, который координировал репликацию, увеличился в разы. Поэтому нам пришлось остановить этот процесс, так как он грозил замедлить все производство. Пришлось искать другие способы переезда.
В базе данных хранятся все запросы, поступающие на наш сервис. В Clickhouse мы храним два типа данных:
- “Сырые данные” хранятся в течение 2 недель. За это время «набегает» около 6 Tб данных.
- “Агрегаты» - важные результаты обработки сырых данных. Агрегаты занимают около 300 ГБ.
Первое, что пришло нам в голову – это дождаться процесса агрегации данных и перенести в новый центр обработки данных только агрегаты, а тем временем запустить новые шарды и перенести сервисы. Таким образом, задача сводилась к поиску способа переноса данных без каких-либо потерь.
Список возможных вариантов решения нашей основной задачи мы нашли в статье Altinity: https://kb.altinity.com/altinity-kb-setup-and-maintenance/altinity-kb-data-migration/. В нашем случае было только два варианта дальнейших действий: физический перенос или копирование с помощью clickhouse-copier.
Физический перенос данных
Первый способ - низкоуровневая (физическая) передача данных.
Clickhouse хранит данные частями (Parts), которые могут быть физически скопированы с одного сервера на другой в виде файлов. Для этого на сервере-источнике необходимо выполнить аналогичный запрос и последовательно отсоединить все части для всех таблиц:
#DETACH PART/PARTITION ALTER TABLE <DATABASE_NAME>.<TABLE_NAME> DETACH PARTITION|PART <PARTITION_EXPRESSION>
Затем необходимо скопировать файлы из каталогов:
`ClickHouse_data_dir/<DATABASE_NAME>/<TABLE_NAME>/detached`
на новый сервер, а затем выполнить обратный запрос:
#ATTACH PART/PARTITION ALTER TABLE <DATABASE_NAME>.<TABLE_NAME> ATTACH PARTITION|PART <PARTITION_EXPRESSION>
Мы протестировали этот вариант, но в одном из тестов количество столбцов в таблицах на исходном и целевом серверах не совпало. Возможно, что-то было неправильно объединено. Мы не стали углубляться в истинные причины такого результата и решили использовать более безопасный метод.
Копирование данных с помощью clickhouse-copier
Второй вариант - копирование данных на более высоком уровне с помощью специальной утилиты от Clickhouse. В исходной базе данных выполняется SELECT, а затем на стороне назначения - INSERT. Утилита отслеживает статус выполнения всех задач в ZooKeeper, что позволяет избежать нежелательных проблем и ошибок.
В открытых источниках об этом методе написано мало, поэтому нам пришлось разбираться в нем практически с нуля.
Данный вариант показался нам более надежным в сравнении с физическим переносом , однако и здесь в итоге не обошлось без сюрпризов. Например, выяснилось, что если запустить утилиту на стороне источника, то после копирования данные не совпадут. Возможно, в этом виноваты сетевые проблемы - похоже, что некоторые фрагменты данных просто не дошли до нового центра обработки данных, а утилита это не зафиксировала. Однако, когда мы запустили ее на стороне назначения, все данные совпали на 100 %.
Процесс переноса данных для одного шарда занял 3-4 часа, еще полчаса потребовалось на различные сопутствующие манипуляции, в частности на копирование внутри сервера, поскольку изначально мы переносили данные во «временную» таблицу. Мы не могли выполнить копирование непосредственно в рабочую таблицу, поскольку кластер состоял из машин в двух дата-центрах, и в процессе копирования мы получили бы дублирующуюся статистику. Поэтому мы скопировали данные из Майами во временную таблицу в Детройте, а затем в детройтском центре обработки данных объединили ее с производственной, выполнив вставки объемом в 500-600 миллионов столбцов.
Несмотря на все меры предосторожности, мы все же столкнулись с парой случаев, когда клиенты превышали лимиты, установленные в панели администратора, поскольку в процессе копирования они видели неполную статистику. К счастью, общие потери составили всего 20 долларов или около того.
Копирование данных на практике
На практике процесс копирования выглядит следующим образом.
Чтобы начать работу, в первую очередь нужно поместить данные для утилиты в ZooKeeper:
zkCli.sh -server localhost:2181 create /clickhouse/copytasks ""
Далее нам нужно создать схему копирования - файл с именем `chema.xml`.
<clickhouse>
<!-- Configuration of clusters as in an ordinary server config -->
<remote_servers>
<source_cluster>
<shard>
<internal_replication>true</internal_replication>
<replica>
<host>IP</host>
<port>9000</port>
<user>default</user>
<password></password>
</replica>
</shard>
</source_cluster>
<destination_cluster>
<shard>
<internal_replication>true</internal_replication>
<replica>
<host>IP</host>
<port>9000</port>
<user>default</user>
<password></password>
</replica>
</shard>
</destination_cluster>
</remote_servers>
<!-- How many simultaneously active workers are possible. If you run more workers superfluous workers will sleep. -->
<max_workers>1</max_workers>
<!-- Setting used to fetch (pull) data from source cluster tables -->
<settings_pull>
<readonly>1</readonly>
</settings_pull>
<!-- Setting used to insert (push) data to destination cluster tables -->
<settings_push>
<readonly>0</readonly>
</settings_push>
<!-- Common setting for fetch (pull) and insert (push) operations. Also, copier process context uses it.
They are overlaid by <settings_pull/> and <settings_push/> respectively. -->
<settings>
<connect_timeout>3</connect_timeout>
<!-- Sync insert is set forcibly, leave it here just in case. -->
<insert_distributed_sync>1</insert_distributed_sync>
</settings>
<!-- Copying tasks description.
You could specify several table task in the same task description (in the same ZooKeeper node), they will be performed
sequentially.
-->
<tables>
<!-- A table task, copies one table. -->
<table_hits>
<cluster_pull>source_cluster</cluster_pull>
<database_pull>database</database_pull>
<table_pull>table_local</table_pull>
<cluster_push>destination_cluster</cluster_push>
<database_push>database</database_push>
<table_push>table_local1</table_push>
<engine>
ENGINE=ReplicatedSummingMergeTree('/clickhouse/tables/{shard}/ssp_report_common1', '{replica}')
partition by toYYYYMMDD(sspRequestDate)
order by (sspRequestDate, dspId, sspRequestCountry, endpointId)
</engine>
<sharding_key>rand()</sharding_key>
</table_hits>
</tables>
</clickhouse>
В данном случае:
- В разделе `clickhouse` описываются серверы - откуда и куда мы копируем данные;
- В разделе `tables` описывается то, какие таблицы мы копируем;
- В `table_hits` находится описание самого процесса копирования.
После этого мы отправляем этот файл в ZooKeeper:
zkCli.sh -server localhost:2181 create /clickhouse/copytasks/description "`cat schema.xml`"
Далее мы переходим на сервер Clickhouse и создаем файл `zookeeper.xml`, который необходим для успешной работы утилиты:
<clickhouse>
<logger>
<level>trace</level>
<size>100M</size>
<count>3</count>
</logger>
<zookeeper>
<node index="1">
<host>127.0.0.1</host>
<port>2181</port>
</node>
</zookeeper>
</clickhouse>
Выполняем:
clickhouse-copier --config-file=zookeeper.xml --task-path=/clickhouse/copytasks
После того, как все будет готово, очистите zookeeper:
zkCli.sh -server localhost:2181 deleteall /clickhouse/copytasks
Убедившись в том, что все данные совпали, выполните команду SQL:
INSERT INTO table_local SELECT * FROM table_local1;
В целом, процесс миграции БД у нас занял всего 10 дней (с учетом того, что параллельно мы переносили и другие части сервиса). Мы искренне надеемся, что эта статья поможет Вам сэкономить драгоценное время при решении подобных задач.
