将 Confluent 平台与 ClickHouse 集成

前提条件

假定您已经熟悉：

• ClickHouse Connector Sink
• Confluent Platform 和自定义连接器（Custom Connectors）。

ClickHouse 官方 Kafka 连接器（适用于 Confluent Platform）

在 Confluent Platform 上安装

本指南是一个快速入门，帮助你在 Confluent Platform 上开始使用 ClickHouse Sink 连接器。
更多详情请参考 Confluent 官方文档。

创建 Topic

在 Confluent Platform 上创建一个主题（topic）相当简单，详细步骤请参阅此文档。

重要说明

• Kafka topic 名称必须与 ClickHouse 表名相同。可以通过使用转换器（transformer）来调整这一点（例如 ExtractTopic）。
• 增加分区数量并不总能提升性能——请关注我们即将发布的指南以获取更多细节和性能优化建议。

安装连接器

你可以从我们的仓库下载连接器——也欢迎在此提交评论和问题！

在控制台中依次进入 "Connector Plugins" -> "Add plugin"，并使用以下设置：

'Connector Class' - 'com.clickhouse.kafka.connect.ClickHouseSinkConnector'
'Connector type' - Sink
'Sensitive properties' - 'password'。这将确保在配置期间对 ClickHouse 密码条目进行掩码处理。

Example:

收集连接信息

要通过 HTTP(S) 连接到 ClickHouse，您需要以下信息：

参数	说明
HOST 和 PORT	通常，在使用 TLS 时端口为 8443，不使用 TLS 时端口为 8123。
DATABASE NAME	默认提供一个名为 default 的数据库，请填写您要连接的目标数据库名称。
USERNAME 和 PASSWORD	默认用户名为 default。请使用适合您使用场景的用户名。

您的 ClickHouse Cloud 服务的详细信息可以在 ClickHouse Cloud 控制台中查看。 选择某个服务并点击 Connect：

选择 HTTPS。连接信息会显示在示例 curl 命令中。

如果您使用的是自托管 ClickHouse，则连接信息由您的 ClickHouse 管理员进行设置。

配置连接器

导航至 Connectors -> Add Connector，并使用以下设置（注意：以下值仅为示例）：

{
  "database": "<DATABASE_NAME>",
  "errors.retry.timeout": "30",
  "exactlyOnce": "false",
  "schemas.enable": "false",
  "hostname": "<CLICKHOUSE_HOSTNAME>",
  "password": "<SAMPLE_PASSWORD>",
  "port": "8443",
  "ssl": "true",
  "topics": "<TOPIC_NAME>",
  "username": "<SAMPLE_USERNAME>",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false"
}

指定连接端点

你需要配置一个允许连接器访问的端点列表。 在添加网络出站（egress）端点时，必须使用完全限定域名（FQDN）。 示例：u57swl97we.eu-west-1.aws.clickhouse.com:8443

注意

你必须指定 HTTP(S) 端口。该 Connector 目前尚不支持 Native 协议。

阅读文档。

现在一切已就绪！

已知限制

• Custom Connectors 必须使用公共互联网端点。不支持静态 IP 地址。
• 你可以覆盖部分 Custom Connector 属性。请参阅官方文档中的完整列表。
• Custom Connectors 仅在部分 AWS 区域可用。
• 请参阅官方文档中关于 Custom Connectors 的限制列表。