Интеграция ClickHouse с Kafka с помощью именованных коллекций

Введение

В этом руководстве мы рассмотрим, как подключить ClickHouse к Kafka с помощью именованных коллекций. Использование файла конфигурации для именованных коллекций дает несколько преимуществ:

Централизованное и более удобное управление параметрами конфигурации.
Параметры можно изменять без правок в SQL-определениях таблиц.
Конфигурации проще проверять и устранять неполадки, анализируя один файл конфигурации.

Это руководство было протестировано на Apache Kafka 3.4.1 и ClickHouse 24.5.1.

Предварительные условия

В этом документе предполагается, что у вас есть:

Рабочий кластер Kafka.
Настроенный и работающий кластер ClickHouse.
Базовые знания SQL и понимание конфигурации ClickHouse и Kafka.

Предварительные требования

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

<access_management>1</access_management>
<named_collection_control>1</named_collection_control>
<show_named_collections_secrets>1</show_named_collections_secrets>

Подробнее о том, как включить управление доступом, см. в руководстве по управлению пользователями.

Конфигурация

Добавьте следующий раздел в файл ClickHouse config.xml:

<!-- Именованные коллекции для интеграции с Kafka -->
<named_collections>
    <cluster_1>
        <!-- Параметры движка ClickHouse Kafka -->
        <kafka_broker_list>c1-kafka-1:9094,c1-kafka-2:9094,c1-kafka-3:9094</kafka_broker_list>
        <kafka_topic_list>cluster_1_clickhouse_topic</kafka_topic_list>
        <kafka_group_name>cluster_1_clickhouse_consumer</kafka_group_name>
        <kafka_format>JSONEachRow</kafka_format>
        <kafka_commit_every_batch>0</kafka_commit_every_batch>
        <kafka_num_consumers>1</kafka_num_consumers>
        <kafka_thread_per_consumer>1</kafka_thread_per_consumer>

        <!-- Расширенная конфигурация Kafka -->
        <kafka>
            <security_protocol>SASL_SSL</security_protocol>
            <enable_ssl_certificate_verification>false</enable_ssl_certificate_verification>
            <sasl_mechanism>PLAIN</sasl_mechanism>
            <sasl_username>kafka-client</sasl_username>
            <sasl_password>kafkapassword1</sasl_password>
            <debug>all</debug>
            <auto_offset_reset>latest</auto_offset_reset>
        </kafka>
    </cluster_1>

    <cluster_2>
        <!-- Параметры движка ClickHouse Kafka -->
        <kafka_broker_list>c2-kafka-1:29094,c2-kafka-2:29094,c2-kafka-3:29094</kafka_broker_list>
        <kafka_topic_list>cluster_2_clickhouse_topic</kafka_topic_list>
        <kafka_group_name>cluster_2_clickhouse_consumer</kafka_group_name>
        <kafka_format>JSONEachRow</kafka_format>
        <kafka_commit_every_batch>0</kafka_commit_every_batch>
        <kafka_num_consumers>1</kafka_num_consumers>
        <kafka_thread_per_consumer>1</kafka_thread_per_consumer>

        <!-- Расширенная конфигурация Kafka -->
        <kafka>
            <security_protocol>SASL_SSL</security_protocol>
            <enable_ssl_certificate_verification>false</enable_ssl_certificate_verification>
            <sasl_mechanism>PLAIN</sasl_mechanism>
            <sasl_username>kafka-client</sasl_username>
            <sasl_password>kafkapassword2</sasl_password>
            <debug>all</debug>
            <auto_offset_reset>latest</auto_offset_reset>
        </kafka>
    </cluster_2>
</named_collections>

Примечания по конфигурации

Скорректируйте адреса Kafka и связанные параметры в соответствии с конфигурацией вашего кластера Kafka.
Раздел перед <kafka> содержит параметры движка ClickHouse Kafka. Полный список параметров см. в разделе параметров движка ClickHouse Kafka.
Раздел внутри <kafka> содержит расширенные параметры конфигурации Kafka. Дополнительные параметры см. в конфигурации librdkafka.
В этом примере используются протокол безопасности SASL_SSL и механизм PLAIN. Скорректируйте эти параметры в соответствии с конфигурацией вашего кластера Kafka.

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

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

Создайте базу данных

CREATE DATABASE kafka_testing ON CLUSTER LAB_CLICKHOUSE_CLUSTER;

Создайте таблицы Kafka

Создайте первую таблицу Kafka для первого кластера Kafka:

CREATE TABLE kafka_testing.first_kafka_table ON CLUSTER LAB_CLICKHOUSE_CLUSTER
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
)
ENGINE = Kafka(cluster_1);

Создайте вторую таблицу Kafka для второго кластера Kafka:

CREATE TABLE kafka_testing.second_kafka_table ON CLUSTER STAGE_CLICKHOUSE_CLUSTER
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
)
ENGINE = Kafka(cluster_2);

Создайте реплицированные таблицы

Создайте таблицу для первой таблицы Kafka:

CREATE TABLE kafka_testing.first_replicated_table ON CLUSTER STAGE_CLICKHOUSE_CLUSTER
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
) ENGINE = ReplicatedMergeTree()
ORDER BY id;

Создайте таблицу для второй таблицы Kafka:

CREATE TABLE kafka_testing.second_replicated_table ON CLUSTER STAGE_CLICKHOUSE_CLUSTER
(
    `id` UInt32,
    `first_name` String,
    `last_name` String
) ENGINE = ReplicatedMergeTree()
ORDER BY id;

Создайте materialized views

Создайте materialized view для вставки данных из первой таблицы Kafka в первую реплицированную таблицу:

CREATE MATERIALIZED VIEW kafka_testing.cluster_1_mv ON CLUSTER STAGE_CLICKHOUSE_CLUSTER TO first_replicated_table AS
SELECT 
    id,
    first_name,
    last_name
FROM first_kafka_table;

Создайте materialized view для вставки данных из второй таблицы Kafka во вторую реплицированную таблицу:

CREATE MATERIALIZED VIEW kafka_testing.cluster_2_mv ON CLUSTER STAGE_CLICKHOUSE_CLUSTER TO second_replicated_table AS
SELECT 
    id,
    first_name,
    last_name
FROM second_kafka_table;

Проверка настройки

Теперь вы должны увидеть соответствующие группы consumer group в ваших кластерах Kafka:

cluster_1_clickhouse_consumer в cluster_1
cluster_2_clickhouse_consumer в cluster_2

Выполните следующие запросы на любом из узлов ClickHouse, чтобы увидеть данные в обеих таблицах:

SELECT * FROM first_replicated_table LIMIT 10;

SELECT * FROM second_replicated_table LIMIT 10;

Примечание

В этом руководстве в оба топика Kafka поступают одни и те же данные. В вашем случае они будут различаться. Вы можете добавить сколько угодно кластеров Kafka. Пример вывода:

┌─id─┬─first_name─┬─last_name─┐
│  0 │ FirstName0 │ LastName0 │
│  1 │ FirstName1 │ LastName1 │
│  2 │ FirstName2 │ LastName2 │
└────┴────────────┴───────────┘

На этом настройка интеграции ClickHouse с Kafka с использованием именованных коллекций завершена. Централизовав конфигурацию Kafka в файле ClickHouse config.xml, вы сможете проще управлять настройками и изменять их, обеспечивая более удобную и эффективную интеграцию.

Источники данных

Визуализация данных

Ингестия данных

Интеграции данных

SQL-клиенты

Инструменты

Интеграция ClickHouse с Kafka с помощью именованных коллекций

Введение

Предварительные условия

Предварительные требования

Конфигурация

Примечания по конфигурации

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

Создайте базу данных

Создайте таблицы Kafka

Создайте реплицированные таблицы

Создайте materialized views

Проверка настройки

Примечание

​Введение

​Предварительные условия

​Предварительные требования

​Конфигурация

​Примечания по конфигурации

​Создание таблиц и баз данных

​Создайте базу данных

​Создайте таблицы Kafka

​Создайте реплицированные таблицы

​Создайте materialized views

​Проверка настройки

​Примечание

Введение

Предварительные условия

Предварительные требования

Конфигурация

Примечания по конфигурации

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

Создайте базу данных

Создайте таблицы Kafka

Создайте реплицированные таблицы

Создайте materialized views

Проверка настройки

Примечание