Геораспределенный кластер

Елена Дворядкина

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

В этой статье приведены примеры возможных топологий геораспределенного кластера ADQM. Наиболее подходящая топология зависит от необходимого баланса в каждом конкретном случае между такими параметрами, как надежность, низкая задержка, дисковое хранилище и ремонтопригодность. Например, часто используются такие топологии, как Stretched-кластер, Stretched-кластер с локальным чтением и Stretched-кластер с локальной записью, обеспечивающие работоспособность кластера в условиях высокой латентности.

В каждом из приведенных ниже примеров серверы кластера расположены в двух дата-центрах — West и East.

Stretched-кластер

Особенности топологии:

  • Запросы на чтение и вставку данных могут быть направлены в дата-центры West или East, при этом не гарантируется, что все операции чтения будут выполняться локально (возможно, запрос будет выполняться репликами в разных дата-центрах).

  • Серверы ZooKeeper размещаются в дата-центрах West и East, а также в третьем дополнительном дата-центре Nord, так как ZooKeeper требует нечетного количества серверов.

Stretched-кластер
Stretched-кластер
Stretched-кластер
Stretched-кластер

Конфигурация логического кластера

В настройках каждого сервера (config.xml) описывается кластер stretched_cluster, содержащий два шарда, каждый из которых состоит из двух реплик, находящихся в разных дата-центрах. Параметр шардов internal_replication устанавливается в true — в этом случае репликацию данных будет выполнять реплицируемая таблица, в которую данные записываются (то есть данные записываются на одну любую реплику шарда, вторая реплика получает данные автоматически).

config.xml
<remote_servers>
  <stretched_cluster>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>shard1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>shard2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster>
</remote_servers>

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

На всех серверах кластера создается реплицируемая таблица replicated_table на базе движка ReplicatedMergeTree.

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

CREATE TABLE table_distributed ON CLUSTER stretched_cluster AS replicated_table
ENGINE = Distributed(stretched_cluster, default, replicated_table);

Stretched-кластер с локальным чтением

Особенности топологии:

  • Запросы на чтение и вставку данных могут быть направлены в дата-центры West или East, при этом гарантируется, что все операции чтения будут локальными.

  • Серверы ZooKeeper размещаются в дата-центрах West и East, а также в третьем дополнительном дата-центре Nord, так как ZooKeeper требует нечетного количества серверов.

Stretched-кластер с локальным чтением
Stretched-кластер с локальным чтением
Stretched-кластер с локальным чтением
Stretched-кластер с локальным чтением

Конфигурация логических кластеров

В настройках каждого сервера (config.xml) описываются следующие кластеры:

  • stretched_cluster_writes — включает серверы дата-центров West и East (два шарда, по две реплики в каждом), на которых будут выполняться операции записи данных;

  • stretched_cluster_west_reads — включает cерверы дата-центра West (два шарда, по одной реплике в каждом), на которых будут выполняться запросы на чтение;

  • stretched_cluster_east_reads — включает cерверы дата-центра East (два шарда, по одной реплике в каждом), на которых будут выполняться запросы на чтение.

Параметр шардов internal_replication устанавливается в true — в этом случае репликацию данных будет выполнять реплицируемая таблица, в которую данные записываются (то есть данные записываются на одну любую реплику шарда, вторая реплика получает данные автоматически).

config.xml
<remote_servers>
  <stretched_cluster_writes>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>shard1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>shard2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster_writes>
  <stretched_cluster_west_reads>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster_west_reads>
  <stretched_cluster_east_reads>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster_east_reads>
</remote_servers>

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

На всех серверах кластера создается реплицируемая таблица replicated_table на базе движка ReplicatedMergeTree.

Распределенные таблицы для чтения создаются отдельно в каждом дата-центре:

  • West:

    CREATE TABLE table_distributed_west_reads ON CLUSTER stretched_cluster_west_reads AS replicated_table
ENGINE = Distributed(stretched_cluster_west_reads, default, replicated_table);

  • East:

    CREATE TABLE table_distributed_east_reads ON CLUSTER stretched_cluster_east_reads AS replicated_table
ENGINE = Distributed(stretched_cluster_east_reads, default, replicated_table);

Для записи создаются идентичные распределенные таблицы в логическом кластере, объединяющем оба дата-центра:

CREATE TABLE table_distributed_writes ON CLUSTER stretched_cluster_writes AS replicated_table
ENGINE = Distributed(stretched_cluster_writes, default, replicated_table);

Stretched-кластер с локальной записью

Особенности топологии:

  • Запросы на чтение и вставку данных могут быть направлены в дата-центры West или East, при этом гарантируется, что все операции записи данных будут выполняться локально.

  • Серверы ZooKeeper размещаются в дата-центрах West и East, а также в третьем дополнительном дата-центре Nord, так как ZooKeeper требует нечетного количества серверов.

Stretched-кластер с локальной записью
Stretched-кластер с локальной записью
Stretched-кластер с локальной записью
Stretched-кластер с локальной записью

Конфигурация логических кластеров

В настройках каждого сервера (config.xml) описываются следующие кластеры:

  • stretched_cluster_reads — включает серверы дата-центров West и East (два шарда, по две реплики в каждом), на которых будут выполняться запросы на чтение;

  • stretched_cluster_west_writes — включает cерверы дата-центра West (два шарда, по одной реплике в каждом), на которых будут выполняться операции записи данных;

  • stretched_cluster_east_writes — включает cерверы дата-центра East (два шарда, по одной реплике в каждом), на которых будут выполняться операции записи данных.

Параметр шардов internal_replication устанавливается в true — в этом случае репликацию данных будет выполнять реплицируемая таблица, в которую данные записываются (то есть данные записываются на одну любую реплику шарда, вторая реплика получает данные автоматически).

config.xml
<remote_servers>
  <stretched_cluster_reads>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>shard1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>shard2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster_reads>
  <stretched_cluster_west_writes>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster_west_writes>
  <stretched_cluster_east_writes>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </stretched_cluster_east_writes>
</remote_servers>

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

На всех серверах кластера создается реплицируемая таблица replicated_table на базе движка ReplicatedMergeTree.

Распределенные таблицы для записи создаются отдельно в каждом дата-центре:

  • West:

    CREATE TABLE table_distributed_west_writes ON CLUSTER stretched_cluster_west_writes AS replicated_table
ENGINE = Distributed(stretched_cluster_west_writes, default, replicated_table);

  • East:

    CREATE TABLE table_distributed_east_writes ON CLUSTER stretched_cluster_east_writes AS replicated_table
ENGINE = Distributed(stretched_cluster_east_writes, default, replicated_table);

Для чтения создаются идентичные распределенные таблицы в логическом кластере, объединяющем оба дата-центра:

CREATE TABLE table_distributed_reads ON CLUSTER stretched_cluster_reads AS replicated_table
ENGINE = Distributed(stretched_cluster_reads, default, replicated_table);

Асимметричный кластер

Особенности топологии:

  • Чтение данных возможно как в дата-центре West, так и в East, а запись — только в West.

  • Серверы одного ансамбля ZooKeeper расположены в двух дата-центрах (третий дата-центр не нужен), при этом в кворуме участвуют только серверы дата-центра West. Реплики в дата-центре East не могут быть лидерами (это управляется настройкой replicated_can_become_leader=0 в секции merge_tree файла config.xml).

  • Параметр insert_quorum установлен в 1, так как число серверов в кворуме должно быть равно или меньше, чем количество реплик в основном дата-центре (в данном примере это дата-центр West, в который записываются данные).

Асимметричный кластер
Асимметричный кластер
Асимметричный кластер
Асимметричный кластер

Конфигурация логических кластеров

В настройках каждого сервера (config.xml) описываются следующие кластеры:

  • asymmetric_cluster_west — включает серверы дата-центра West, которые распределены в два шарда, по одной реплике в каждом;

  • asymmetric_cluster_east — включает серверы дата-центра East, которые распределены в два шарда, по одной реплике в каждом.

Параметр шардов internal_replication устанавливается в true — в этом случае репликацию данных будет выполнять реплицируемая таблица, в которую данные записываются (то есть данные записываются на одну любую реплику шарда, вторая реплика получает данные автоматически).

config.xml
<remote_servers>
  <asymmetric_cluster_west>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
    </shard>
  </asymmetric_cluster_west>
  <asymmetric_cluster_east>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </asymmetric_cluster_east>
</remote_servers>

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

На всех серверах кластера создается реплицируемая таблица replicated_table на базе движка ReplicatedMergeTree.

Распределенные таблицы для чтения создаются в каждом дата-центре:

  • West:

    CREATE TABLE table_distributed_west_reads ON CLUSTER asymmetric_cluster_west AS replicated_table
ENGINE = Distributed(asymmetric_cluster_west, default, replicated_table);

  • East:

    CREATE TABLE table_distributed_east_reads ON CLUSTER asymmetric_cluster_east AS replicated_table
ENGINE = Distributed(asymmetric_cluster_east, default, replicated_table);

Распределенная таблица для записи создается только в дата-центре West:

CREATE TABLE table_distributed_west_writes ON CLUSTER asymmetric_cluster_west AS replicated_table
ENGINE = Distributed(asymmetric_cluster_west, default, replicated_table);

Кластер с локальным чтением без собственной георепликации

Особенности топологии:

  • Не поддерживается нативная репликация ClickHouse между дата-центрами. Репликация между дата-центрами обеспечивается через распределенные таблицы.

  • Параметр insert_quorum устанавливается равным 2, чтобы при выполнении запроса INSERT распределенная таблица гарантированно записывала данные на две реплики, расположенные в разных дата-центрах.

  • Запросы на чтение и вставку данных могут быть направлены в дата-центры East или West, при этом гарантируется, что все операции чтения будут локальными.

  • Допускается большая задержка между West и East.

  • ZooKeeper доступен только в одном дата-центре.

Кластер с локальным чтением без собственной георепликации
Кластер с локальным чтением без собственной георепликации
Кластер с локальным чтением без собственной георепликации
Кластер с локальным чтением без собственной георепликации

Конфигурация логических кластеров

В настройках каждого сервера (config.xml) описываются следующие кластеры:

  • cluster_writes — включает серверы дата-центров West и East (два шарда, по две реплики в каждом), на которых будут выполняться операции записи данных;

  • cluster_west_reads — включает серверы дата-центра West (два шарда, по одной реплике в каждом), на которых будут выполняться запросы на чтение данных;

  • cluster_east_reads — включает серверы дата-центра East (два шарда, по одной реплике в каждом), на которых будут выполняться запросы на чтение данных.

Параметр internal_replication для шардов кластера cluster_writes, в который будут записываться данные, устанавливается в false, чтобы данные реплицировались через распределенную таблицу.

config.xml
<remote_servers>
  <cluster_writes>
    <shard>
      <internal_replication>false</internal_replication>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>sh1r1.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>false</internal_replication>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
      <replica>
        <host>sh2r1.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </cluster_writes>
  <cluster_west_reads>
    <shard>
      <replica>
        <host>shard1.west</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <replica>
        <host>shard2.west</host>
        <port>9000</port>
      </replica>
    </shard>
  </cluster_west_reads>
  <cluster_east_reads>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard1r2.east</host>
        <port>9000</port>
      </replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica>
        <host>shard2r2.east</host>
        <port>9000</port>
      </replica>
    </shard>
  </cluster_east_reads>
</remote_servers>

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

На всех серверах кластера создается реплицируемая таблица replicated_table на базе движка ReplicatedMergeTree.

Распределенная таблица для чтения создается в дата-центре West на каждом сервере отдельно, так как чтобы работал механизм ON CLUSTER, требуется ZooKeeper. Для этого необходимо выполнить следующий запрос на всех серверах дата-центра West:

CREATE TABLE table_distributed_reads AS replicated_table
ENGINE = Distributed(cluster_west_reads, default, replicated_table);

Чтобы создать распределенную таблицу для чтения в дата-центре East, достаточно выполнить следующий запрос на любом сервере:

CREATE TABLE table_distributed_reads ON CLUSTER cluster_east_reads AS replicated_table
ENGINE = Distributed(cluster_east_reads, default, replicated_table);

Распределенная таблица для записи должна быть создана на серверах shard1.west, shard2.west, shard1r1.east, shard2r1.east:

CREATE TABLE table_distributed_writes AS replicated_table
ENGINE = Distributed(cluster_writes, default, replicated_table);
Нашли ошибку? Выделите текст и нажмите Ctrl+Enter чтобы сообщить о ней