将 Amazon MSK 与 ClickHouse 集成

注意：视频中展示的策略较为宽松，仅用于快速上手。请参阅下文基于最小权限原则的 IAM 指南。

前提条件

我们假定：

• 你已经熟悉 ClickHouse Connector Sink、Amazon MSK 和 MSK Connectors。我们推荐阅读 Amazon MSK 的入门指南和 MSK Connect 指南。
• MSK broker 已配置为可通过公网访问。请参阅《开发者指南》中 Public Access 章节。

ClickHouse 官方 Kafka 连接器（适用于 Amazon MSK）

收集连接详细信息

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

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

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

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

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

步骤

1. 请确保已熟悉 ClickHouse Connector Sink。
2. 创建一个 MSK 实例。
3. 创建并分配 IAM 角色。
4. 从 ClickHouse Connect Sink 的 发布页面下载 jar 文件。
5. 在 Amazon MSK 控制台的 自定义插件页面中安装下载的 jar 文件。
6. 如果 Connector 需要与公网 ClickHouse 实例通信，请启用互联网访问。
7. 在配置中提供主题（topic）名称、ClickHouse 实例主机名和密码。

connector.class=com.clickhouse.kafka.connect.ClickHouseSinkConnector
tasks.max=1
topics=<topic_name>
ssl=true
security.protocol=SSL
hostname=<hostname>
database=<database_name>
password=<password>
ssl.truststore.location=/tmp/kafka.client.truststore.jks
port=8443
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
exactlyOnce=true
username=default
schemas.enable=false

推荐的 IAM 权限（最小权限原则）

仅使用部署所需的最小权限集。先从下面的基线配置开始，只有在实际使用相关服务时才添加对应的可选服务权限。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "MSK集群访问",
      "Effect": "Allow",
      "Action": [
        "kafka:DescribeCluster",
        "kafka:GetBootstrapBrokers",
        "kafka:DescribeClusterV2",
        "kafka:ListClusters",
        "kafka:ListClustersV2"
      ],
      "Resource": "*"
    },
    {
      "Sid": "Kafka授权",
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:Connect",
        "kafka-cluster:DescribeCluster",
        "kafka-cluster:DescribeGroup",
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:ReadData"
      ],
      "Resource": "*"
    },
    {
      "Sid": "可选Glue模式注册表",
      "Effect": "Allow",
      "Action": [
        "glue:GetSchema*",
        "glue:ListSchemas",
        "glue:ListSchemaVersions"
      ],
      "Resource": "*"
    },
    {
      "Sid": "可选Secrets Manager",
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": [
        "arn:aws:secretsmanager:<region>:<account-id>:secret:<your-secret-name>*"
      ]
    },
    {
      "Sid": "可选S3读取",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::<your-bucket>/<optional-prefix>/*"
    }
  ]
}

• 仅在使用 AWS Glue Schema Registry 时使用 Glue 块。
• 仅在从 Secrets Manager 获取凭证/信任库时使用 Secrets Manager 块。限定 ARN 的作用域。
• 仅在从 S3 加载工件（例如 truststore）时使用 S3 块。将权限限定到指定 bucket/前缀。

另请参阅：Kafka 最佳实践 – IAM。

性能调优

提高性能的一种方法是在 worker 配置中添加以下内容，以调整从 Kafka 拉取的批量大小和记录数量：

consumer.max.poll.records=[记录数]
consumer.max.partition.fetch.bytes=[记录数 * 单条记录字节数]

您使用的具体数值会因期望的记录数量和记录大小而有所不同。例如，默认值为：

consumer.max.poll.records=500
consumer.max.partition.fetch.bytes=1048576

你可以在官方的 Kafka 文档和 Amazon MSK 文档中查阅更多信息（包括实现细节和其他方面的考量）。

关于 MSK Connect 的网络注意事项

为了使 MSK Connect 能够连接到 ClickHouse，我们建议将 MSK 集群部署在带有 Private NAT、可访问互联网的私有子网中。下面提供了相关设置步骤。请注意，虽然也支持使用公共子网，但并不推荐，因为需要持续为 ENI 分配 Elastic IP 地址，AWS 在此提供了更多详细信息

1. 创建私有子网： 在 VPC 中创建一个新的子网，并将其指定为私有子网。该子网不应具有直接访问互联网的能力。
2. 创建 NAT Gateway： 在 VPC 的公共子网中创建一个 NAT Gateway。NAT Gateway 允许私有子网中的实例连接到互联网或其他 AWS 服务，但阻止互联网主动与这些实例建立连接。
3. 更新路由表： 添加一条路由，将发往互联网的流量指向 NAT Gateway。
4. 确保安全组和网络 ACL 配置： 配置 security groups（安全组） 和 network ACLs（网络访问控制列表），以允许相关流量。
   1. 从 MSK Connect worker ENI 到 MSK broker 的 TLS 端口（通常为 9094）。
   2. 从 MSK Connect worker ENI 到 ClickHouse 端点：9440（原生 TLS）或 8443（HTTPS）。
   3. 在 broker 的安全组上允许来自 MSK Connect worker 安全组的入站流量。
   4. 对于自托管的 ClickHouse，开放服务器中配置的端口（HTTP 默认为 8123）。
5. 将安全组附加到 MSK： 确保这些安全组已附加到 MSK 集群和 MSK Connect workers。
6. 与 ClickHouse Cloud 的连通性：
   1. 公共端点 + IP 允许名单：需要从私有子网经由 NAT 出口访问。
   2. 在可用地区使用私有连通性（例如 VPC peering/PrivateLink/VPN）。确保已启用 VPC DNS hostnames/resolution，并且 DNS 能解析私有端点。
7. 验证连通性（快速检查清单）：
   1. 在 connector 运行环境中，解析 MSK bootstrap DNS，并通过 TLS 连接到 broker 端口。
   2. 在端口 9440 与 ClickHouse 建立 TLS 连接（或使用 8443 进行 HTTPS）。
   3. 如果使用 AWS 服务（Glue/Secrets Manager），允许对这些端点的出口访问。