> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 使用命名集合集成 ClickHouse 与 Kafka

> 如何使用命名集合将 ClickHouse 连接到 Kafka

<div id="introduction">
  ## 简介
</div>

本指南将介绍如何使用 命名集合 将 ClickHouse 连接到 Kafka。使用 命名集合 的配置文件有以下几个优势：

* 可集中管理配置项，且管理起来更方便。
* 无需修改 SQL 表定义即可调整设置。
* 只需检查单个配置文件，即可更轻松地审查配置并排查问题。

本指南已在 Apache Kafka 3.4.1 和 ClickHouse 24.5.1 上测试验证。

<div id="assumptions">
  ## 前提假设
</div>

本文档假定你已具备：

1. 一个正常运行的 Kafka 集群。
2. 一个已设置完成并正在运行的 ClickHouse 集群。
3. 具备基本的 SQL 知识，并熟悉 ClickHouse 和 Kafka 的配置。

<div id="prerequisites">
  ## 前置条件
</div>

请确保创建 命名集合 的用户具有所需的访问权限：

```xml theme={null}
<access_management>1</access_management>
<named_collection_control>1</named_collection_control>
<show_named_collections>1</show_named_collections>
<show_named_collections_secrets>1</show_named_collections_secrets>
```

有关如何启用访问控制的更多信息，请参阅[用户管理指南](/docs/zh/concepts/features/security/access-rights)。

<div id="configuration">
  ## 配置
</div>

将以下内容添加到 ClickHouse 的 `config.xml` 文件中：

```xml theme={null}
<!-- 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>
```

<div id="configuration-notes">
  ### 配置说明
</div>

1. 请调整 Kafka 地址及相关配置，使其与您的 Kafka 集群设置相匹配。
2. `<kafka>` 前面的部分包含 ClickHouse Kafka 引擎参数。有关完整参数列表，请参阅 [Kafka 引擎参数 ](/docs/zh/reference/engines/table-engines/integrations/kafka)。
3. `<kafka>` 内的部分包含扩展的 Kafka 配置选项。更多选项请参阅 [librdkafka 配置](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md)。
4. 本示例使用 `SASL_SSL` 安全协议和 `PLAIN` 机制。请根据您的 Kafka 集群配置调整这些设置。

<div id="creating-tables-and-databases">
  ## 创建表和数据库
</div>

在 ClickHouse 集群上创建所需的数据库和表。如果你以单节点方式运行 ClickHouse，请省略 SQL 命令中的集群相关部分，并使用其他任意引擎替代 `ReplicatedMergeTree`。

<div id="create-the-database">
  ### 创建数据库
</div>

```sql theme={null}
CREATE DATABASE kafka_testing ON CLUSTER LAB_CLICKHOUSE_CLUSTER;
```

<div id="create-kafka-tables">
  ### 创建 Kafka 表
</div>

为第一个 Kafka 集群创建第一张 Kafka 表：

```sql theme={null}
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 表：

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

<div id="create-replicated-tables">
  ### 创建复制表
</div>

为第一个 Kafka 表创建对应的表：

```sql theme={null}
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 表创建对应的表：

```sql theme={null}
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;
```

<div id="create-materialized-views">
  ### 创建 materialized views
</div>

创建一个 materialized view，将第一个 Kafka 表中的数据插入到第一个 复制表 中：

```sql theme={null}
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 表中的数据插入到第二个复制表中：

```sql theme={null}
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;
```

<div id="verifying-the-setup">
  ## 验证设置
</div>

现在，你应该能在 Kafka 集群中看到对应的消费者组：

* `cluster_1_clickhouse_consumer` 位于 `cluster_1`
* `cluster_2_clickhouse_consumer` 位于 `cluster_2`

在任意一个 ClickHouse 节点上运行以下查询，即可查看这两个表中的数据：

```sql theme={null}
SELECT * FROM first_replicated_table LIMIT 10;
```

```sql theme={null}
SELECT * FROM second_replicated_table LIMIT 10;
```

<div id="note">
  ### 注意
</div>

在本指南中，摄取到两个 Kafka topic 的数据是相同的；而在您的实际环境中，它们会有所不同。您可以根据需要添加任意数量的 Kafka 集群。

示例输出：

```response theme={null}
┌─id─┬─first_name─┬─last_name─┐
│  0 │ FirstName0 │ LastName0 │
│  1 │ FirstName1 │ LastName1 │
│  2 │ FirstName2 │ LastName2 │
└────┴────────────┴───────────┘
```

至此，已完成使用命名集合将 ClickHouse 与 Kafka 集成的设置。将 Kafka 配置集中放在 ClickHouse 的 `config.xml` 文件中后，您可以更轻松地管理和调整这些设置，从而使集成过程更简洁高效。