> ## 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 连接器 与 Amazon MSK

# 将 Amazon MSK 集成到 ClickHouse

<div class="vimeo-container">
  <Frame>
    <iframe
      src="//www.youtube.com/embed/6lKI_WlQ3-s"
      frameborder="0"
      allow="autoplay;
fullscreen;
picture-in-picture"
      allowfullscreen
    />
  </Frame>
</div>

> 注意：视频中展示的策略较为宽松，仅适用于快速入门。请参阅下方的最小权限 IAM 指南。

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

我们假设：

* 你已熟悉 [ClickHouse Connector Sink](/docs/zh/integrations/connectors/data-ingestion/kafka/kafka-clickhouse-connect-sink)；
* 你已熟悉 Amazon MSK 和 MSK Connectors。建议先阅读 Amazon MSK 的[入门指南](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html)以及 [MSK Connect 指南](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect.html)。

<div id="the-official-kafka-connector-from-clickhouse-with-amazon-msk">
  ## 将 ClickHouse 官方 Kafka 连接器与 Amazon MSK 搭配使用
</div>

<div id="gather-your-connection-details">
  ### 准备连接信息
</div>

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

| Parameter(s)              | Description                                |
| ------------------------- | ------------------------------------------ |
| `HOST` and `PORT`         | 通常，使用 TLS 时端口为 8443；不使用 TLS 时端口为 8123。     |
| `DATABASE NAME`           | 默认情况下，存在一个名为 `default` 的数据库。请使用你要连接的数据库名称。 |
| `USERNAME` and `PASSWORD` | 默认情况下，用户名为 `default`。请根据你的使用场景使用相应的用户名。    |

你的 ClickHouse Cloud 服务的连接信息可在 ClickHouse Cloud 控制台中查看。
选择一个服务，然后点击 **Connect**：

<div className="ch-image-md">
  <Frame>
    <img src="https://mintcdn.com/private-7c7dfe99/CFFsa2agBPbviR4r/images/_snippets/cloud-connect-button.webp?fit=max&auto=format&n=CFFsa2agBPbviR4r&q=85&s=ec0a298a33ca841e947fa5e8bae47362" alt="ClickHouse Cloud 服务连接按钮" width="998" height="932" data-path="images/_snippets/cloud-connect-button.webp" />
  </Frame>
</div>

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

<div className="ch-image-md">
  <Frame>
    <img src="https://mintcdn.com/private-7c7dfe99/CFFsa2agBPbviR4r/images/_snippets/connection-details-https.webp?fit=max&auto=format&n=CFFsa2agBPbviR4r&q=85&s=cb0fbd98aa2b5b7ca484c9f53395ee07" alt="ClickHouse Cloud HTTPS 连接信息" width="1320" height="1184" data-path="images/_snippets/connection-details-https.webp" />
  </Frame>
</div>

如果你使用的是自管理 ClickHouse，则连接信息由你的 ClickHouse 管理员配置。

<div id="steps">
  ### 步骤
</div>

1. 确保你已熟悉 [ClickHouse Connector Sink](/docs/zh/integrations/connectors/data-ingestion/kafka/kafka-clickhouse-connect-sink)
2. [创建一个 MSK 实例](https://docs.aws.amazon.com/msk/latest/developerguide/create-cluster.html)。
3. [创建并分配 IAM 角色](https://docs.aws.amazon.com/msk/latest/developerguide/create-client-iam-role.html)。
4. 从 ClickHouse Connector Sink 的 [Release 页面](https://github.com/ClickHouse/clickhouse-kafka-connect/releases)下载一个 `jar` 文件。
5. 在 Amazon MSK 控制台的 [自定义插件页面](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-plugins.html)中安装下载的 `jar` 文件。
6. 如果连接器与公网 ClickHouse 实例通信，请[启用互联网访问](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-internet-access.html)。
7. 在配置中提供 topic 名称、ClickHouse 实例主机名和密码。

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

<div id="iam-least-privilege">
  ## 推荐的 IAM 权限 (最小权限原则)
</div>

请使用满足你的部署需求所需的最小权限集。先采用下面的基础权限，仅在实际使用可选服务时再额外添加相应权限。

```json theme={null}
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "MSKClusterAccess",
      "Effect": "Allow",
      "Action": [
        "kafka:DescribeCluster",
        "kafka:GetBootstrapBrokers",
        "kafka:DescribeClusterV2",
        "kafka:ListClusters",
        "kafka:ListClustersV2"
      ],
      "Resource": "*"
    },
    {
      "Sid": "KafkaAuthorization",
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:Connect",
        "kafka-cluster:DescribeCluster",
        "kafka-cluster:DescribeGroup",
        "kafka-cluster:DescribeTopic",
        "kafka-cluster:ReadData"
      ],
      "Resource": "*"
    },
    {
      "Sid": "OptionalGlueSchemaRegistry",
      "Effect": "Allow",
      "Action": [
        "glue:GetSchema*",
        "glue:ListSchemas",
        "glue:ListSchemaVersions"
      ],
      "Resource": "*"
    },
    {
      "Sid": "OptionalSecretsManager",
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": [
        "arn:aws:secretsmanager:<region>:<account-id>:secret:<your-secret-name>*"
      ]
    },
    {
      "Sid": "OptionalS3Read",
      "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 块。请将范围限定到存储桶/前缀。

另请参阅：[Kafka 最佳实践 – IAM](/docs/zh/integrations/clickpipes/kafka/best-practices#iam)。

<div id="performance-tuning">
  ## 性能调优
</div>

提升性能的一种方法，是调整批次大小以及从 Kafka 拉取的记录数。为此，可在 **工作线程** 配置中添加以下内容：

```yml theme={null}
consumer.max.poll.records=[NUMBER OF RECORDS]
consumer.max.partition.fetch.bytes=[NUMBER OF RECORDS * RECORD SIZE IN BYTES]
```

具体使用哪些值会因所需的记录数量和记录大小而异。例如，默认值如下：

```yml theme={null}
consumer.max.poll.records=500
consumer.max.partition.fetch.bytes=1048576
```

你可以在官方的 [Kafka](https://kafka.apache.org/documentation/#consumerconfigs) 和
[Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-workers.html#msk-connect-create-custom-worker-config) 文档中了解更多细节 (包括实现细节和其他注意事项) 。

<div id="notes-on-networking-for-msk-connect">
  ## 关于 MSK Connect 网络的说明
</div>

为了让 MSK Connect 能够连接到 ClickHouse，我们建议将您的 MSK 集群部署在私有子网中，并通过 Private NAT 提供互联网访问。下面提供了相关设置说明。请注意，虽然也支持公有子网，但由于需要持续为您的 ENI 分配 Elastic IP 地址，因此不建议这样做。[AWS 在此提供了更多详细信息](https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-internet-access.html)

1. **创建私有子网：** 在您的 VPC 中创建一个新子网，并将其指定为私有子网。该子网不应可直接访问互联网。
2. **创建 NAT 网关：** 在您的 VPC 的公有子网中创建一个 NAT 网关。NAT 网关使私有子网中的实例能够连接到互联网或其他 AWS 服务，但会阻止互联网主动与这些实例建立连接。
3. **更新路由表：** 添加一条路由，将发往互联网的流量定向到 NAT 网关
4. **确保安全组和网络 ACL 配置正确：** 配置您的[安全组](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-security-groups.html)和[网络 ACL (访问控制列表) ](https://docs.aws.amazon.com/vpc/latest/userguide/vpc-network-acls.html)，以允许相关流量。
   1. 允许从 MSK Connect 工作线程 ENI 到 MSK 消息代理 TLS 端口 (通常为 9094) 的流量。
   2. 允许从 MSK Connect 工作线程 ENI 到 ClickHouse 端点的流量：9440 (原生 TLS) 或 8443 (HTTPS) 。
   3. 允许消息代理安全组 (SG) 接收来自 MSK Connect 工作线程 安全组的入站流量。
   4. 对于自托管 ClickHouse，开放您 server 中配置的端口 (HTTP 默认为 8123) 。
5. **将安全组附加到 MSK：** 确保这些安全组已附加到您的 MSK 集群和 MSK Connect 工作线程。
6. **到 ClickHouse Cloud 的连通性：**
   1. 公有端点 + IP 允许列表：需要私有子网具备通过 NAT 的出站访问能力。
   2. 在可用时使用私有连接 (例如 VPC peering/PrivateLink/VPN) 。确保已启用 VPC DNS 主机名/解析，并且 DNS 能够解析该专用端点。
7. **验证连通性 (快速检查清单) ：**
   1. 从 connector 运行环境中，解析 MSK bootstrap DNS，并通过 TLS 连接到消息代理端口。
   2. 在 9440 端口 (或 HTTPS 使用 8443) 上与 ClickHouse 建立 TLS 连接。
   3. 如果使用 AWS 服务 (Glue/Secrets Manager) ，请允许到这些端点的出站访问。