跳转到主要内容
在 Kafka Connect 框架中，Connector 定义并管理了一组用于执行特定数据集成任务（ETL）的 Task。在 AutoMQ 托管 Kafka Connect 服务中，每一个 Connector 底层运行的多个 Task 都被调度在独立的 Worker 集群中，确保了任务间的资源隔离和稳定运行。

​
前置条件

在创建和部署 Connector 之前，您需要确保已经准备好以下依赖资源：
  • 一个正常运行的 AutoMQ 实例及用于数据访问的 ACL 用户凭证。
  • 用于数据集成源头或目标的 Kafka Topic 和 Consumer Group。
  • 一个配置正确的 Amazon EKS（Elastic Kubernetes Service）集群，并在 EKS 集群内为 Connector 运行准备的 Service Account 和 Namespace。
  • 一个具备特定权限的 AWS IAM Role，用于授权 Connector 访问云资源，该 IAM Role 和 Service Account 必须绑定。

​
步骤 1：创建 AutoMQ 实例

数据集成任务需要一个 AutoMQ 实例作为元数据服务以及数据集成的 Sink、Source。如果您尚未创建实例，请参考以下文档完成创建。

​
步骤 2：创建 ACL 用户凭证，完成 ACL 授权

Connector 需要使用 Kafka 客户端协议访问 AutoMQ 集群，因此必须为其提供一套有效的 ACL 用户凭证。
重要约束： 在创建 ACL 用户时，您必须确保为其授予了 Connector 预期要访问的所有 Topic 的读/写权限。如果您的 Connector 是一个 Sink Connector（数据导出），还必须为其授予相关 Consumer Group 的读取权限。否则，Connector 任务将因权限不足而无法正常启动。

​
步骤 3：准备 EKS 环境及 IAM 角色

AutoMQ 托管 Kafka Connect 的 Worker 运行在您自己的 EKS 集群中，以保证数据的安全性和网络隔离性。同时，它需要通过一个绑定的 IAM Role 来获取访问其他云资源（如 S3）的权限。 在准备 EKS 环境时，请特别注意以下几点：
  • 服务账号与 IAM 角色的绑定：您需要创建一个 Kubernetes Service Account，并将其与一个 AWS IAM Role 进行绑定（通常通过 IRSA - IAM Roles for Service Accounts 实现）。
  • IAM Role 权限策略：您必须为上述 IAM Role 附加一个权限策略（Policy），至少授予其对当前 AutoMQ 环境运维存储桶（S3 Bucket）的读写权限。
以下是一个 IAM Policy 的示例（请将 YOUR_BUCKET_NAME 替换为实际的存储桶名称）： 
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
                "s3:DeleteObject",
                "s3:ListBucket",
                "s3:GetObjectVersion",
                "s3:DeleteObjectVersion",
                "s3:ListBucketVersions"
            ],
            "Resource": [
                "arn:aws:s3:::YOUR_BUCKET_NAME/*",
                "arn:aws:s3:::YOUR_BUCKET_NAME"
            ]
        }
    ]
}
重要提醒： 如果当前 Connector 需要访问除 AutoMQ 存储桶以外的其他云资源（如 RDS、DynamoDB 等），请确保在 IAM Role 的权限策略中添加相应的访问权限。
完成以上所有云资源和相关配置准备后，即可创建 Kafka Connector。

​
创建 Connector

创建 Connector 的流程主要分为部署信息、插件参数、Metrics 集成等步骤。

​
步骤 1：填写部署配置

  1. 登录 AutoMQ 控制台，在左侧导航栏中选择 Kafka Connect。
  2. 点击 Connector 页签，然后点击 创建 Connector 按钮。
  3. 在创建页面，您需要填写以下基础参数：
    • 选择插件：从下拉列表中选择一个您希望使用的插件及其版本。此为必选项。
    • 选择 Kafka 集群：在此处配置 Connector 访问 AutoMQ 集群所需的信息。
      • AutoMQ 实例：选择您已准备好的 AutoMQ 实例。
      • ACL 用户名/密码：填写用于数据访问的 ACL 用户凭证。
      • Topic / Consumer Group：根据插件是 Source 还是 Sink 类型，填写需要读写的 Topic 或消费组。
    • 选择 EKS 集群：配置 Connector 运行的底层计算资源。
      • EKS 集群：选择您已准备好的 EKS 集群。
      • Namespace：填写为 Connector 准备的 Kubernetes 命名空间。
      • Service Account：填写已绑定了 IAM Role 的服务账号。
    • 填写容量配置：根据您的数据负载，为 Connector 分配资源。
      • 最大任务数：指定 Connector 可并行运行的最大 Task 数量。
      • Worker 数量：指定运行这些 Task 的 Worker 进程数量。
      • Worker 规格：为每个 Worker 选择合适的 CPU 和内存规格。您可以根据插件的资源消耗情况选择不同大小的规格。
你需要为 Connector 运行提供必要的 Kubernetes 资源配额（CPU、内存等）。请确保 EKS 集群中的节点池有足够的资源来满足 Connector 的运行需求。

​
步骤 2：填写插件配置

完成基础配置后，点击 下一步，进入插件参数配置页面。AutoMQ 支持两种模式来填写插件所需的自定义配置：
  • 表单模式：AutoMQ 会自动解析插件的配置定义（Schema），并将其渲染成一个结构化的表单。您可以根据每个参数的描述和提示直接填写，操作直观便捷。
  • 自定义模式：如果您更熟悉插件的参数，或者需要填写表单模式未覆盖的高级参数，可以选择此模式。在该模式下，您可以完全自由地以键值对（Key-Value）的形式指定所有配置。

​
步骤 3：配置 Metrics 导出（可选）

为了实现自定义监控和报警，AutoMQ 支持将 Connector 运行时的 Metrics 数据导出到您自有的 Prometheus 系统中。 在这一步，您可以选择配置外部 Prometheus 的 remote_write 地址和访问凭证（如 Basic Auth）。配置完成后，所有相关的 Metrics 数据将被实时转发。

​
步骤 4：预览并创建

完成所有配置后，您可以在预览页面检查所有参数。确认无误后，点击 创建 按钮，AutoMQ 将开始部署并启动您的 Connector 任务。

​
管理 Connector

Connector 创建后，您可以在其详情页进行全方位的管理和运维操作。

​
修改配置

您可以随时进入 Connector 详情页，对其配置进行修改。
  • 修改容量配置：根据业务负载的变化，您可以动态调整 最大任务数、Worker 数量 以及 Worker 规格，以实现资源的弹性伸缩，满足不同场景下的性能需求。
  • 修改运行参数：您可以修改插件的自定义配置，以调整其运行行为。

​
管理任务生命周期

在 Connector 详情页，您可以对底层的 Task 执行生命周期管理操作，方便进行调试或维护。
  • 暂停（Pause）：临时停止所有 Task 的数据处理活动。
  • 恢复（Resume）：从暂停状态恢复所有 Task 的运行。
  • 重启（Restart）：批量重启所有 Task，通常用于加载新的配置或从错误状态中恢复。

​
删除 Connector

当您不再需要运行某个 Connector 时，可以随时将其删除。 操作步骤：
  1. 在 Connector 列表页面，选择目标 Connector。
  2. 点击 删除 按钮。
  3. 在二次确认对话框中确认操作后，Connector 将被永久删除。
注意：删除操作不可撤销。删除 Connector 将会一并移除其下所有正在运行的 Task，以及这些 Task 相关的消费位点和进度信息。

​
生产环境监控报警

对于生产环境运行的 Connector，完善的监控和报警是保障其稳定性的关键。AutoMQ 提供了 Metrics 和日志两大维度的可观测性能力。
  • Metrics 指标集成：您可以在创建或编辑 Connector 时，通过配置 Metrics 导出，将 Connector 运行的核心指标（如吞吐、延迟、错误率等）对接到您自己的 Prometheus 监控系统中，并基于这些数据配置告警规则。
  • 日志查询：当需要排查问题时，您可以随时前往 Connector 的详情页面，查询其运行日志，快速定位问题根源。