跳到主要内容

Kafka Connect REST API

Kafka Connect 是 Apache Kafka 生态系统中的一个重要组件，用于在 Kafka 和其他系统之间高效地传输数据。Kafka Connect REST API 提供了一种通过 HTTP 请求与 Kafka Connect 集群进行交互的方式，允许开发者和管理员动态管理连接器（Connectors）、任务（Tasks）以及监控集群状态。

本文将逐步介绍 Kafka Connect REST API 的核心功能，并通过实际案例展示其应用场景。

什么是 Kafka Connect REST API？​

Kafka Connect REST API 是一组基于 HTTP 的接口，用于与 Kafka Connect 集群进行交互。通过 REST API，您可以执行以下操作：

  • 创建、更新、删除连接器。
  • 查看连接器和任务的状态。
  • 重启连接器或任务。
  • 获取集群配置和插件信息。

REST API 的默认端口是 8083，您可以通过发送 HTTP 请求（如 GET、POST、PUT、DELETE）来管理 Kafka Connect 集群。

Kafka Connect REST API 的核心功能​

1. 获取集群信息​

您可以通过以下 API 获取 Kafka Connect 集群的基本信息：

bash
GET /

示例请求：

bash
curl -X GET http://localhost:8083/

示例响应：

json
{
  "version": "3.5.0",
  "commit": "abcdef123456",
  "kafka_cluster_id": "cluster-123"
}

2. 创建连接器​

使用以下 API 创建一个新的连接器：

bash
POST /connectors

示例请求：

bash
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "my-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://localhost:3306/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "my-topic-"
  }
}'

示例响应：

json
{
  "name": "my-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://localhost:3306/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "my-topic-"
  },
  "tasks": []
}

3. 查看连接器状态​

使用以下 API 查看连接器的状态：

bash
GET /connectors/{connector_name}/status

示例请求：

bash
curl -X GET http://localhost:8083/connectors/my-source-connector/status

示例响应：

json
{
  "name": "my-source-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "192.168.1.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "192.168.1.1:8083"
    }
  ]
}

4. 重启连接器​

如果连接器出现问题，您可以通过以下 API 重启它：

bash
POST /connectors/{connector_name}/restart

示例请求：

bash
curl -X POST http://localhost:8083/connectors/my-source-connector/restart

实际应用场景​

场景：实时同步数据库数据到 Kafka​

假设您需要将 MySQL 数据库中的数据实时同步到 Kafka 中。您可以使用 Kafka Connect 的 JDBC Source Connector 来实现这一需求。

  1. 创建连接器：通过 REST API 创建一个 JDBC Source Connector。
  2. 监控状态：定期检查连接器的状态，确保数据同步正常。
  3. 处理异常：如果连接器失败，通过 REST API 重启连接器。

总结​

Kafka Connect REST API 是管理和监控 Kafka Connect 集群的强大工具。通过本文的学习，您已经掌握了如何使用 REST API 创建、管理和监控连接器。以下是进一步学习的资源：

练习

尝试使用 Kafka Connect REST API 创建一个 Sink Connector，将 Kafka 中的数据写入到 Elasticsearch 中。