Движок таблицы Kafka

• Публикуйте потоки данных или подписывайтесь на них.
• Организуйте отказоустойчивое хранилище.
• Обрабатывайте потоки по мере их поступления.

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [ALIAS expr1],
    name2 [type2] [ALIAS expr2],
    ...
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'host:port',
    kafka_topic_list = 'topic1,topic2,...',
    kafka_group_name = 'group_name',
    kafka_format = 'data_format'[,]
    [kafka_security_protocol = '',]
    [kafka_sasl_mechanism = '',]
    [kafka_sasl_username = '',]
    [kafka_sasl_password = '',]
    [kafka_autodetect_client_rack = '',]
    [kafka_schema = '',]
    [kafka_num_consumers = N,]
    [kafka_max_block_size = 0,]
    [kafka_skip_broken_messages = N,]
    [kafka_commit_every_batch = 0,]
    [kafka_client_id = '',]
    [kafka_poll_timeout_ms = 0,]
    [kafka_poll_max_batch_size = 0,]
    [kafka_flush_interval_ms = 0,]
    [kafka_consumer_reschedule_ms = 0,]
    [kafka_thread_per_consumer = 0,]
    [kafka_handle_error_mode = 'default',]
    [kafka_commit_on_select = false,]
    [kafka_consumer_acquire_timeout_ms = 30000,]
    [kafka_max_rows_per_message = 1,]
    [kafka_compression_codec = '',]
    [kafka_compression_level = -1,]
    [kafka_partition_shard_num = '',]
    [kafka_shard_count = 0];

Обязательные параметры:

• kafka_broker_list — Список брокеров, разделённых запятыми (например, localhost:9092).
• kafka_topic_list — Список топиков Kafka.
• kafka_group_name — Группа потребителей Kafka. Смещения чтения отслеживаются отдельно для каждой группы. Если вы не хотите, чтобы сообщения дублировались в кластере, используйте везде одно и то же имя группы.
• kafka_format — Формат сообщений. Используется та же нотация, что и для SQL-функции FORMAT, например JSONEachRow. Дополнительные сведения см. в разделе Форматы.

Необязательные параметры:

• kafka_security_protocol — протокол, используемый для связи с брокерами. Возможные значения: plaintext, ssl, sasl_plaintext, sasl_ssl.
• kafka_sasl_mechanism - механизм SASL, используемый для аутентификации. Возможные значения: GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER, AWS_MSK_IAM.
• kafka_aws_region - регион AWS для аутентификации MSK IAM. Если не указан, автоматически определяется по адресу брокера. Указывайте его явно при использовании псевдонимов PrivateLink или пользовательских DNS-имен хостов, не содержащих информации о регионе. По умолчанию: пусто (автоопределение).
• kafka_sasl_username - имя пользователя для SASL-аутентификации при использовании механизмов PLAIN и SASL-SCRAM-...
• kafka_sasl_password — пароль SASL для механизмов PLAIN и SASL-SCRAM-...
• kafka_schema — параметр, который необходимо использовать, если для формата требуется определение схемы. Например, Cap'n Proto требует указать путь к файлу схемы и имя корневого объекта schema.capnp:Message.
• kafka_schema_registry_skip_bytes — Количество байтов, которое нужно пропустить в начале каждого сообщения при использовании schema registry с заголовками обёртки (например, AWS Glue Schema Registry, который добавляет 19-байтную обёртку). Диапазон: [0, 255]. По умолчанию: 0.
• kafka_num_consumers — Количество consumers на таблицу. Укажите больше consumers, если пропускной способности одного consumer недостаточно. Общее количество consumers не должно превышать количество партиций в topic, поскольку на одну партицию может быть назначен только один consumer, и не должно быть больше количества физических ядер на сервере, где развернут ClickHouse. По умолчанию: 1.
• kafka_max_block_size — Максимальный размер батча (в сообщениях) при выполнении poll. По умолчанию: max_insert_block_size.
• kafka_skip_broken_messages — допустимое для парсера сообщений Kafka число несовместимых со схемой сообщений на блок. Если kafka_skip_broken_messages = N, то движок пропускает N сообщений Kafka, которые не удаётся разобрать (одно сообщение соответствует одной строке данных). Значение по умолчанию: 0.
• kafka_commit_every_batch — Выполнять коммит для каждого потреблённого и обработанного батча, а не один коммит после записи всего блока. По умолчанию: 0.
• kafka_client_id — идентификатор клиента. По умолчанию пустой.
• kafka_poll_timeout_ms — тайм-аут для одного опроса из Kafka. По умолчанию: stream_poll_timeout_ms.
• kafka_poll_max_batch_size — Максимальное количество сообщений, получаемых за один опрос Kafka. По умолчанию: max_block_size.
• kafka_flush_interval_ms — Тайм-аут для сброса данных из Kafka. По умолчанию: stream_flush_interval_ms.
• kafka_consumer_reschedule_ms — интервал перепланирования при остановке потоковой обработки Kafka (например, когда нет доступных для чтения сообщений). Эта настройка определяет задержку перед тем, как consumer повторит опрос. Не должно превышать kafka_consumers_pool_ttl_ms. По умолчанию: 500 миллисекунд.
• kafka_thread_per_consumer — Выделяет отдельный поток для каждого консьюмера. Если включен, каждый консьюмер независимо сбрасывает данные, параллельно (в противном случае строки от нескольких консьюмеров объединяются в один блок). По умолчанию: 0.
• kafka_handle_error_mode — как обрабатывать ошибки в движке Kafka. Возможные значения: default (если не удастся разобрать сообщение, будет сгенерировано исключение), stream (сообщение об исключении и исходное сообщение будут сохранены в виртуальных столбцах _error и _raw_message), dead_letter_queue (данные, связанные с ошибкой, будут сохранены в system.dead_letter_queue).
• kafka_commit_on_select — Выполнять коммит сообщений при выполнении запроса SELECT. Значение по умолчанию: false.
• kafka_consumer_acquire_timeout_ms — тайм-аут в миллисекундах на получение потребителя Kafka при выполнении прямых SELECT-запросов к таблице Kafka2 (с хранением смещений в Keeper). Когда к одной и той же таблице одновременно выполняются несколько прямых SELECT-запросов, каждый из них должен ждать, пока потребители не станут доступны. Тайм-аут предотвращает взаимные блокировки в случаях, когда запросы удерживают разные подмножества потребителей. Значение по умолчанию: 30000.
• kafka_max_rows_per_message — Максимальное количество строк, записываемых в одно сообщение Kafka для построчных форматов. По умолчанию: 1.
• kafka_autodetect_client_rack — автоматически задаёт параметр client.rack для librdkafka, чтобы отдавать предпочтение ближайшим репликам Kafka. Поддерживаемые источники: AWS_ZONE_ID — идентификатор зоны доступности AWS IMDSv2, например euc1-az1; AWS_ZONE_NAME — имя зоны доступности AWS IMDSv2, например eu-central-1a; GCP_ZONE — зона сервиса метаданных GCP, например europe-central2-a; CLICKHOUSE — использовать внутреннее определение ClickHouse, которое может опираться на метаданные облака или конфигурацию; AWS_ZONE_NAME_THEN_GCP_ZONE — сначала попытаться использовать AWS_ZONE_NAME, а затем GCP_ZONE. По умолчанию: пустая строка, отключено. Совет: в разных средах используются разные форматы зон доступности. Amazon MSK обычно использует идентификаторы зон, поэтому предпочтителен AWS_ZONE_ID. Confluent Cloud обычно использует имена зон, поэтому предпочтителен AWS_ZONE_NAME. Если вы не уверены, используйте AWS_ZONE_NAME_THEN_GCP_ZONE или проверьте значение broker.rack в вашем кластере. Примечание: брокеры Kafka должны быть настроены с broker.rack и replica.selector.class=org.apache.kafka.common.replica.RackAwareReplicaSelector.
• kafka_compression_codec — кодек сжатия, используемый при отправке сообщений. Поддерживаются: пустая строка, none, gzip, snappy, lz4, zstd. Если указана пустая строка, кодек сжатия таблицей не задаётся, поэтому будут использоваться значения из файлов конфигурации или значение по умолчанию из librdkafka. По умолчанию: пустая строка.
• kafka_compression_level — параметр уровня сжатия для алгоритма, выбранного с помощью kafka_compression_codec. Более высокие значения обеспечивают лучшее сжатие, но требуют больше ресурсов CPU. Допустимый диапазон зависит от алгоритма: [0-9] для gzip; [0-12] для lz4; только 0 для snappy; [0-12] для zstd; -1 = уровень сжатия по умолчанию, зависящий от кодека. Значение по умолчанию: -1.
• kafka_map_virtual_columns_on_write — Если включено, столбцы со специальными именами _key, _timestamp, _headers.name и _headers.value в схеме таблицы сопоставляются с соответствующими метаданными сообщений Kafka при INSERT и исключаются из полезной нагрузки сообщения. См. Сопоставление столбцов с метаданными сообщений Kafka. По умолчанию: false.
• kafka_partition_shard_num — Текущий номер сегмента для статической привязки партиций к сегментам. Должен быть от 1 до kafka_shard_count включительно. Партиции назначаются по формуле partition_id % kafka_shard_count == kafka_partition_shard_num - 1. Поддерживает раскрытие макросов (например, '{shard}'). Должен использоваться вместе с kafka_shard_count. Поддерживается только в StorageKafka2 (требуются kafka_keeper_path и kafka_replica_name). По умолчанию: '' (отключено).
• kafka_shard_count — Общее количество сегментов, участвующих в потреблении. Используется вместе с kafka_partition_shard_num для статического назначения партиций. Должен использоваться вместе с kafka_partition_shard_num. Поддерживается только в StorageKafka2. По умолчанию: 0 (отключено).

Примеры:

CREATE TABLE queue (
    timestamp UInt64,
    level String,
    message String
  ) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');

  SELECT * FROM queue LIMIT 5;

CREATE TABLE queue2 (
    timestamp UInt64,
    level String,
    message String
  ) ENGINE = Kafka SETTINGS kafka_broker_list = 'localhost:9092',
                            kafka_topic_list = 'topic',
                            kafka_group_name = 'group1',
                            kafka_format = 'JSONEachRow',
                            kafka_num_consumers = 4;

CREATE TABLE queue3 (
    timestamp UInt64,
    level String,
    message String
  ) ENGINE = Kafka('localhost:9092', 'topic', 'group1')
              SETTINGS kafka_format = 'JSONEachRow',
                       kafka_num_consumers = 4;

Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format
      [, kafka_row_delimiter, kafka_schema, kafka_num_consumers, kafka_max_block_size, kafka_skip_broken_messages, kafka_commit_every_batch, kafka_client_id, kafka_poll_timeout_ms, kafka_poll_max_batch_size, kafka_flush_interval_ms, kafka_consumer_reschedule_ms, kafka_thread_per_consumer, kafka_handle_error_mode, kafka_commit_on_select, kafka_max_rows_per_message]);

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

Группы гибко настраиваются и синхронизируются в кластере. Например, если у вас есть 10 топиков и 5 копий таблицы в кластере, то каждая копия получает по 2 топика. Если количество копий меняется, топики автоматически перераспределяются между копиями. Подробнее об этом читайте на http://kafka.apache.org/intro.

Рекомендуется, чтобы у каждого топика Kafka была собственная группа потребителей, обеспечивающая эксклюзивное соответствие между топиком и группой, особенно в средах, где топики могут динамически создаваться и удаляться (например, при тестировании или в промежуточном хранилище).

SELECT не особенно полезен для чтения сообщений (кроме отладки), потому что каждое сообщение можно прочитать только один раз. Гораздо практичнее создавать потоки в реальном времени с помощью materialized view. Для этого:

1. Используйте движок, чтобы создать потребитель Kafka, и рассматривайте его как поток данных.
2. Создайте таблицу с нужной структурой.
3. Создайте materialized view, которое преобразует данные из движка и помещает их в ранее созданную таблицу.

Когда MATERIALIZED VIEW подключается к движку, оно начинает собирать данные в фоновом режиме. Это позволяет непрерывно получать сообщения из Kafka и преобразовывать их в нужный формат с помощью SELECT. У одной таблицы Kafka может быть сколько угодно materialized view; они не читают данные из таблицы Kafka напрямую, а получают новые записи (блоками), благодаря чему можно записывать данные в несколько таблиц с разным уровнем детализации (с группировкой — агрегацией — и без нее).

Пример с использованием именованных коллекций для хранения параметров connection:

  CREATE NAMED COLLECTION kafka_creds AS
    kafka_broker_list = 'localhost:9092',
    kafka_topic_list = 'topic',
    kafka_group_name = 'group1',
    kafka_format = 'JSONEachRow';

  CREATE TABLE queue (
    timestamp UInt64,
    level String,
    message String
  ) ENGINE = Kafka(kafka_creds);

  CREATE TABLE daily (
    day Date,
    level String,
    total UInt64
  ) ENGINE = SummingMergeTree
  PARTITION BY toYYYYMM(day)
  ORDER BY (day, level);

  CREATE MATERIALIZED VIEW consumer TO daily
    AS SELECT toDate(toDateTime(timestamp)) AS day, level, count() AS total
    FROM queue GROUP BY day, level;

  SELECT level, sum(total) FROM daily GROUP BY level;

Чтобы повысить производительность, полученные сообщения группируются в блоки размером max_insert_block_size. Если блок не был сформирован в течение stream_flush_interval_ms миллисекунд, данные будут записаны в таблицу независимо от полноты блока.

Чтобы прекратить получение данных из топика или изменить логику преобразования, отсоедините materialized view:

DETACH TABLE consumer;
ATTACH TABLE consumer;

Если вы хотите изменить целевую таблицу с помощью ALTER, мы рекомендуем отключить materialized view, чтобы избежать расхождений между целевой таблицей и данными, поступающими из представления.

Как и GraphiteMergeTree, движок Kafka поддерживает расширенную настройку через файл конфигурации ClickHouse. Можно использовать два ключа конфигурации: глобальный (внутри <kafka>) и на уровне топика (внутри <kafka><kafka_topic>). Сначала применяется глобальная конфигурация, затем — конфигурация на уровне топика (если она задана).

<kafka>
    <!-- Global configuration options for all tables of Kafka engine type -->
    <debug>cgrp</debug>
    <statistics_interval_ms>3000</statistics_interval_ms>

    <kafka_topic>
        <name>logs</name>
        <statistics_interval_ms>4000</statistics_interval_ms>
    </kafka_topic>

    <!-- Settings for consumer -->
    <consumer>
        <auto_offset_reset>smallest</auto_offset_reset>
        <kafka_topic>
            <name>logs</name>
            <fetch_min_bytes>100000</fetch_min_bytes>
        </kafka_topic>

        <kafka_topic>
            <name>stats</name>
            <fetch_min_bytes>50000</fetch_min_bytes>
        </kafka_topic>
    </consumer>

    <!-- Settings for producer -->
    <producer>
        <kafka_topic>
            <name>logs</name>
            <retry_backoff_ms>250</retry_backoff_ms>
        </kafka_topic>

        <kafka_topic>
            <name>stats</name>
            <retry_backoff_ms>400</retry_backoff_ms>
        </kafka_topic>
    </producer>
</kafka>

Список возможных параметров конфигурации см. в справочнике по конфигурации librdkafka. В конфигурации ClickHouse используйте символ подчёркивания (_) вместо точки. Например, check.crcs=true записывается как <check_crcs>true</check_crcs>.

AWS MSK поддерживает аутентификацию на основе IAM, что позволяет подключаться к кластерам Kafka с помощью учетных данных AWS вместо использования отдельных имен пользователей и паролей.

Базовая настройка:

Задайте kafka_sasl_mechanism = 'AWS_MSK_IAM' в настройках таблицы:

CREATE TABLE msk_queue (
    timestamp UInt64,
    level String,
    message String
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'b-1.mycluster.kafka.us-east-1.amazonaws.com:9098',
    kafka_topic_list = 'my-topic',
    kafka_group_name = 'my-group',
    kafka_format = 'JSONEachRow',
    kafka_sasl_mechanism = 'AWS_MSK_IAM';

Регион AWS автоматически извлекается из конечной точки брокера путем сопоставления с шаблоном:

• Provisioned MSK: b-X.cluster.kafka.<region>.amazonaws.com:9098
• Serverless MSK: boot-X.kafka-serverless.<region>.amazonaws.com:9098
• VPC Endpoint: vpce-X.kafka.<region>.vpce.amazonaws.com:9098

Учетные данные AWS:

Учетные данные всегда загружаются из ~/.aws/credentials и ~/.aws/config (файлов профиля AWS), если они присутствуют. Чтобы также включить профили экземпляров EC2, переменные среды (AWS_ACCESS_KEY_ID и т. д.), роли задач ECS и другие автоматические источники учетных данных, добавьте в конфигурацию сервера:

<kafka>
  <use_environment_credentials>true</use_environment_credentials>
</kafka>

Этот параметр могут настраивать только администраторы сервера. Значение по умолчанию: false.

PrivateLink и пользовательский DNS:

Если вы используете псевдонимы PrivateLink или пользовательские DNS-имена хостов, не содержащие сведений о регионе, явно укажите регион AWS:

CREATE TABLE msk_privatelink_queue (
    timestamp UInt64,
    level String,
    message String
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'my-privatelink-alias.internal.example.com:9098',
    kafka_topic_list = 'my-topic',
    kafka_group_name = 'my-group',
    kafka_format = 'JSONEachRow',
    kafka_sasl_mechanism = 'AWS_MSK_IAM',
    kafka_aws_region = 'us-east-1';

Разрешения IAM:

Разрешения для consumer (для чтения сообщений):

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "kafka-cluster:Connect",
      "kafka-cluster:DescribeTopic",
      "kafka-cluster:ReadData",
      "kafka-cluster:AlterGroup",
      "kafka-cluster:DescribeGroup"
    ],
    "Resource": [
      "arn:aws:kafka:REGION:ACCOUNT:cluster/CLUSTER_NAME/*",
      "arn:aws:kafka:REGION:ACCOUNT:topic/CLUSTER_NAME/TOPIC_NAME/*",
      "arn:aws:kafka:REGION:ACCOUNT:group/CLUSTER_NAME/CONSUMER_GROUP/*"
    ]
  }]
}

Разрешения для producer (на запись сообщений):

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "kafka-cluster:Connect",
      "kafka-cluster:DescribeTopic",
      "kafka-cluster:WriteData"
    ],
    "Resource": [
      "arn:aws:kafka:REGION:ACCOUNT:cluster/CLUSTER_NAME/*",
      "arn:aws:kafka:REGION:ACCOUNT:topic/CLUSTER_NAME/TOPIC_NAME/*"
    ]
  }]
}

Для работы с Kafka с поддержкой Kerberos добавьте дочерний элемент security_protocol со значением sasl_plaintext. Этого достаточно, если средствами ОС получен и кэширован Kerberos ticket-granting ticket. ClickHouse может использовать учетные данные Kerberos с помощью файла keytab. Используйте дочерние элементы sasl_kerberos_service_name, sasl_kerberos_keytab и sasl_kerberos_principal.

Пример:

<!-- Kerberos-aware Kafka -->
<kafka>
<security_protocol>SASL_PLAINTEXT</security_protocol>
<sasl_kerberos_keytab>/home/kafkauser/kafkauser.keytab</sasl_kerberos_keytab>
<sasl_kerberos_principal>kafkauser/kafkahost@EXAMPLE.COM</sasl_kerberos_principal>
</kafka>

• _topic — топик Kafka. Тип данных: LowCardinality(String).
• _key — ключ сообщения. Тип данных: String.
• _offset — смещение сообщения. Тип данных: UInt64.
• _timestamp — временная метка сообщения. Тип данных: Nullable(DateTime).
• _timestamp_ms — временная метка сообщения в миллисекундах. Тип данных: Nullable(DateTime64(3)).
• _partition — партиция топика Kafka. Тип данных: UInt64.
• _headers.name — массив ключей заголовков сообщения. Тип данных: Array(String).
• _headers.value — массив значений заголовков сообщения. Тип данных: Array(String).

Дополнительные виртуальные столбцы, когда kafka_handle_error_mode='stream':

• _raw_message - необработанное сообщение, которое не удалось успешно разобрать. Тип данных: String.
• _error - сообщение об исключении, возникшем при неудачном разборе. Тип данных: String.

Примечание: виртуальные столбцы _raw_message и _error заполняются только при возникновении исключения во время разбора; если сообщение было успешно разобрано, они всегда пусты.

При отправке сообщений через INSERT INTO движок Kafka всегда использует столбец с именем _key (типа String) в качестве ключа сообщения Kafka и столбец с именем _timestamp (типа DateTime) в качестве временной метки сообщения Kafka — если такие столбцы есть в таблице. По умолчанию эти столбцы также включаются в полезную нагрузку отправляемого сообщения наряду с остальными столбцами.

При kafka_map_virtual_columns_on_write = 1 поведение меняется:

• _key (тип String) — сопоставляется с ключом сообщения Kafka.
• _timestamp (тип DateTime) — сопоставляется с временной меткой сообщения Kafka.
• _headers.name (тип Array(String)) и _headers.value (тип Array(String)) — сопоставляются с заголовками сообщений Kafka. Каждая пара (_headers.name[i], _headers.value[i]) становится одним заголовком Kafka. Поскольку _headers.name и _headers.value имеют общий вложенный префикс _headers, ClickHouse требует, чтобы оба массива имели одинаковый размер в каждой строке.

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

Пример:

CREATE TABLE kafka_out
(
    event_json String,
    `_key` String,
    `_timestamp` DateTime,
    `_headers.name` Array(String),
    `_headers.value` Array(String)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'broker:9092',
    kafka_topic_list = 'events',
    kafka_group_name = 'events-producer',
    kafka_format = 'JSONEachRow',
    kafka_map_virtual_columns_on_write = 1;

INSERT INTO kafka_out VALUES
    ('{"a":1}', 'session-42', now(), ['source', 'trace_id'], ['api', 'abc-123']);

Сформированное сообщение Kafka содержит полезную нагрузку {"event_json":"{\"a\":1}"}, ключ session-42, текущую временную метку и два заголовка source=api и trace_id=abc-123.

Движок Kafka поддерживает все форматы, доступные в ClickHouse. Количество строк в одном сообщении Kafka зависит от того, является ли формат построчным или блочным:

• Для построчных форматов количество строк в одном сообщении Kafka можно задать с помощью настройки kafka_max_rows_per_message.
• Для блочных форматов блок нельзя разбить на более мелкие части, но количество строк в одном блоке можно задать с помощью общей настройки max_block_size.

Если включен параметр allow_experimental_kafka_offsets_storage_in_keeper, то для движка таблицы Kafka можно задать еще две настройки:

• kafka_keeper_path задает путь к таблице в ClickHouse Keeper
• kafka_replica_name задает имя реплики в ClickHouse Keeper

Должны быть заданы либо обе настройки, либо ни одна из них. Если заданы обе, будет использоваться новый экспериментальный движок Kafka. Этот новый движок не зависит от хранения зафиксированного смещения в Kafka, а хранит их в ClickHouse Keeper. Он по-прежнему пытается коммитить смещения в Kafka, но опирается на них только при создании таблицы. Во всех остальных случаях (при перезапуске таблицы или восстановлении после какой-либо ошибки) для продолжения чтения сообщений будут использоваться смещения, сохраненные в ClickHouse Keeper. Помимо зафиксированного смещения, он также хранит количество сообщений, прочитанных в последнем батче, поэтому, если вставка завершится ошибкой, будет снова прочитано то же количество сообщений, что при необходимости позволяет выполнить дедупликацию.

При использовании StorageKafka2 можно включить статическую привязку партиций к сегментам, указав kafka_partition_shard_num и kafka_shard_count. Это позволяет нескольким экземплярам ClickHouse (сегментам) читать данные из одного топика Kafka, при этом каждый сегмент обрабатывает только детерминированное подмножество партиций согласно формуле:

partition_id % kafka_shard_count == kafka_partition_shard_num - 1

Оба параметра необходимо указывать вместе; указание только одного из них вызывает исключение. Значение kafka_partition_shard_num должно находиться в диапазоне от 1 до kafka_shard_count включительно. Поддерживается раскрытие макросов (например, '{shard}'), которое повторно выполняется при каждом запуске сервера. Это позволяет использовать общие метаданные таблицы для всех сегментов в базе данных Replicated, при этом каждый сегмент подставляет собственное значение. Проверка выполняется после раскрытия макросов.

Все сегменты должны использовать одно и то же значение kafka_keeper_path. Все реплики используют общие закоммиченные смещения и размеры намерений, но за блокировки партиций конкурируют только реплики с одинаковым номером сегмента (при условии, что у всех реплик одинаковое количество сегментов).

Пример для 3 сегментов, потребляющих топик с 12 партициями:

-- Shard 1: consumes partitions 0, 3, 6, 9
CREATE TABLE kafka_shard1 (key UInt64, value String)
ENGINE = Kafka('localhost:9092', 'my-topic', 'my-group', 'JSONEachRow')
SETTINGS
    kafka_keeper_path = '/clickhouse/kafka/{database}',
    kafka_replica_name = '{replica}',
    kafka_partition_shard_num = '1',
    kafka_shard_count = 3
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper = 1;

-- Shard 2: consumes partitions 1, 4, 7, 10
CREATE TABLE kafka_shard2 (key UInt64, value String)
ENGINE = Kafka('localhost:9092', 'my-topic', 'my-group', 'JSONEachRow')
SETTINGS
    kafka_keeper_path = '/clickhouse/kafka/{database}',
    kafka_replica_name = '{replica}',
    kafka_partition_shard_num = '2',
    kafka_shard_count = 3
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper = 1;

В сочетании с репликами (несколькими значениями kafka_replica_name, использующими один и тот же kafka_keeper_path) фильтр привязки сначала определяет доступные партиции, после чего блокировки ZooKeeper распределяют их между репликами.

Пример:

CREATE TABLE experimental_kafka (key UInt64, value UInt64)
ENGINE = Kafka('localhost:19092', 'my-topic', 'my-consumer', 'JSONEachRow')
SETTINGS
kafka_keeper_path = '/clickhouse/{database}/{uuid}',
kafka_replica_name = '{replica}'
SETTINGS allow_experimental_kafka_offsets_storage_in_keeper=1;

Поскольку новый движок пока является экспериментальным, он ещё не готов к использованию в продакшне. У этой реализации есть несколько известных ограничений:

• Быстрое удаление и повторное создание таблицы либо указание одного и того же пути ClickHouse Keeper для разных движков может привести к проблемам. В качестве рекомендации можно использовать {uuid} в kafka_keeper_path, чтобы избежать конфликтов путей.
• Чтобы обеспечить повторяемые чтения, нельзя потреблять сообщения из нескольких партиций в одном потоке. С другой стороны, потребителей Kafka нужно регулярно опрашивать, чтобы они оставались активными. Из-за этих двух требований мы решили разрешить создание нескольких потребителей только при включённом kafka_thread_per_consumer, иначе становится слишком сложно избежать проблем, связанных с их регулярным опросом.
• При использовании привязки к партициям все сегменты должны использовать одинаковое значение kafka_shard_count; в противном случае некоторые партиции могут потребляться несколькими сегментами или остаться непотреблёнными.

См. также

• Виртуальные столбцы
• background_message_broker_schedule_pool_size
• system.kafka_consumers