Загрузка данных из Kafka в Cloud ClickHouse

Для чтения сообщений из топиков Kafka в Cloud ClickHouse используются таблица с движком Kafka и материализованное представление. Описание движка Kafka приведено в официальной документации ClickHouse.

1. Добавьте подключение к Kafka в экземпляр Cloud ClickHouse. Запишите название подключения: оно понадобится при создании таблицы с движком Kafka.
2. Подключитесь к экземпляру Cloud ClickHouse.

1. Выполните SQL-запрос, указав столбцы и типы данных, соответствующие структуре сообщений и выбранному формату:

    
   CREATE TABLE <ИМЯ_БД>.kafka_table ON CLUSTER '{cluster}'(    `id` UInt32,    `first_name` String,    `last_name` String)ENGINE = Kafka(<НАЗВАНИЕ_ПОДКЛЮЧЕНИЯ>);

   Здесь:

   • <ИМЯ_БД> — имя базы данных, заданное при создании экземпляра Cloud ClickHouse;
   • <НАЗВАНИЕ_ПОДКЛЮЧЕНИЯ> — название подключения к Kafka, созданного на подготовительном шаге.

2. Проверьте результат выполнения SQL-запроса: для каждого узла кластера Cloud ClickHouse в столбце status должно быть указано 0, а столбец error должен быть пуст. Это означает, что таблица создана на всех узлах кластера.

1. Выполните SQL-запрос:

    
   CREATE TABLE <ИМЯ_БД>.kafka_target_table ON CLUSTER '{cluster}'(    `id` UInt32,    `first_name` String,    `last_name` String)ENGINE = ReplicatedMergeTree()ORDER BY id;

2. Убедитесь, что целевая таблица создана на всех узлах кластера: в столбце status должно быть указано 0, а столбец error должен быть пуст.

Целевая таблица будет хранить сообщения, перенесенные материализованным представлением.

1. Выполните SQL-запрос:

    
   CREATE MATERIALIZED VIEW <ИМЯ_БД>.kafka_mv ON CLUSTER '{cluster}'TO <ИМЯ_БД>.kafka_target_table ASSELECT    id,    first_name,    last_nameFROM <ИМЯ_БД>.kafka_table;

2. Убедитесь, что материализованное представление создано на всех узлах кластера: в столбце status должно быть указано 0, а столбец error должен быть пуст.

После создания представления движок Kafka начнет читать сообщения из топиков, а представление — переносить их в целевую таблицу.

1. Дождитесь поступления новых сообщений в выбранный топик.

2. Выполните SQL-запрос к целевой таблице:

    
   SELECT * FROM <ИМЯ_БД>.kafka_target_table;

3. Убедитесь, что значения полей в полученных строках соответствуют исходным сообщениям из Kafka.

Если материализованное представление и таблица с движком Kafka больше не нужны, выполните SQL-запросы:

 
DROP VIEW IF EXISTS <ИМЯ_БД>.kafka_mv ON CLUSTER '{cluster}';DROP TABLE IF EXISTS <ИМЯ_БД>.kafka_table ON CLUSTER '{cluster}';

После удаления материализованного представления новые сообщения поступать в целевую таблицу не будут. Если накопленные данные больше не нужны, выполните SQL-запрос:

 
DROP TABLE IF EXISTS <ИМЯ_БД>.kafka_target_table ON CLUSTER '{cluster}';