Как мы переносили базу Clickhouse между географически удаленными дата-центрами

В конце прошлого года мы писали о сложном переезде нашего собственного сервиса в новый дата-центр в Детройте. Среди прочих задач мы переносили Clickhouse. Напомню, что речь идет о нагруженном сервисе, который обслуживает десятки серверов, принимающих сотни тысяч запросов с низкой latency в секунду.

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

Объем данных для Clickhouse у нас не такой уж и большой - процесс получился не столько объемный, сколько ресурсоемкий. Но в открытых источниках информации по использованным механизмам минимум, поэтому считайте это руководством к clickhouse-copier утилите (https://github.com/ClickHouse/copier) на конкретном примере со скриптами и командами для запуска.

Самый простой подход

Наш сервис функционирует 24/7, т.е. просто выключить и перенести все копированием нельзя. Мы спланировали переезд, но из-за некоторых ограничений самого Clickhouse (а точнее его схемы репликации) не смогли реализовать его в соответствии с планом.

Изначально мы предполагали, что сможем к каждой шарде в старом дата-центре подключить реплику из нового, подождем синхронизации, ну а потом просто отключим реплики в старом дата-центре. Но из-за географической удаленности мы получили слишком большую задержку передачи данных между дата-центрами, несмотря на скорость в 2-2,5 Гбита. В результате в ZooKeeper, который координировал репликацию, объем данных вырос на порядок. Так что процесс пришлось остановить, иначе он грозил затормозить продакшн. Пришлось подыскивать другие способы переезда.

В базе сохраняются все запросы, приходящие в наш сервис. В Clickhouse мы храним два типа данных:

• “Сырые данные” за две недели. За это время накапливается около 6 Тб.

• “Агрегаты” - важные для нас результаты обработки сырых данных. Агрегаты занимают порядка 300 Гб.

Нашим первым решением стало дождаться процесса агрегации данных и перенести в новый дата-центр только их; ну а пока это происходит, запускать новые шарды и переносить сервисы. Так что задача свелась к тому, что необходимо было найти способ перенести данные и не потерять ни байта из тех 300 Гб агрегатов.

Перечень возможных подходов мы нашли в статье от Альтинити: https://kb.altinity.com/altinity-kb-setup-and-maintenance/altinity-kb-data-migration/. В нашем случае было два варианта дальнейших действий: физический перенос или копирование с помощью clickhouse-copier. 

Далее поговорим про каждый из вариантов.

Физический перенос данных

Первый возможный вариант - низкоуровневый (физический) перенос данных.

Clickhouse хранит данные в виде Parts, которые можно физически скопировать с одного сервера на другой в виде файлов. Для этого на исходном сервере надо выполнить подобный запрос и отцепить все 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 ""

Далее нам нужно создать схему копирования - файл schema.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 (с учетом того, что параллельно мы переносили другие части сервиса). Надеемся, что эта статья поможет вам сэкономить время на поиске подхода к решению подобных задач.

Авторы статьи: Игорь Иванов и Денис Палагута, Максилект.

P.S. Мы публикуем наши статьи на нескольких площадках Рунета. Подписывайтесь на нашу страницу в VK или на Telegram-канал, чтобы узнавать обо всех публикациях и других новостях компании Maxilect.