搭建 Kafka Connect 数据同步管道

适用场景

需要把数据库、文件或消息系统实时同步到 Kafka，或把 Kafka 数据写入 Elasticsearch、对象存储、数据仓库。Kafka Connect 用配置而不是代码实现连接器管理。

前置条件

• 可用的 Kafka 集群，版本与 Connect 兼容。
• 连接器插件（例如 Debezium、Elasticsearch Sink、S3 Sink）。
• 内部 topics：connect-configs、connect-offsets、connect-status。
• 源端数据库的 CDC 权限或目标端的写入权限。

步骤

1. 准备 Connect 配置：

cat > connect-distributed.properties <<'EOF'
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
config.storage.topic=connect-configs
config.storage.replication.factor=3
status.storage.topic=connect-status
status.storage.replication.factor=3
plugin.path=/opt/kafka/plugins
EOF

2. 启动分布式 Worker：

connect-distributed.sh connect-distributed.properties
curl -s http://localhost:8083/ | jq

3. 安装 Debezium 等插件到 plugin.path，重启 Worker。

4. 创建 Source Connector：

curl -X POST http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "mysql-source",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "tasks.max": "1",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "***",
      "database.server.id": "184054",
      "topic.prefix": "cdc",
      "database.include.list": "app",
      "schema.history.internal.kafka.bootstrap.servers": "kafka1:9092",
      "schema.history.internal.kafka.topic": "schema-changes.app"
    }
  }'

5. 创建 Sink Connector：

curl -X POST http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "es-sink",
    "config": {
      "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
      "tasks.max": "2",
      "topics": "cdc.app.orders",
      "connection.url": "http://elasticsearch:9200",
      "key.ignore": "true",
      "schema.ignore": "true"
    }
  }'

6. 检查连接器和任务状态：

curl -s http://localhost:8083/connectors | jq
curl -s http://localhost:8083/connectors/mysql-source/status | jq
curl -s http://localhost:8083/connectors/mysql-source/tasks | jq

验证

• Connector 和 Task 状态为 RUNNING。
• 源库新增/更新/删除能出现在目标 topic 或目标系统。
• 重启 Worker 后，offset 从内部 topic 恢复，不重复消费。
• 增加 Task 数能提升并行度（受分区数限制）。
• 目标端数据与源端一致，延迟在可接受范围。

回滚

• 暂停连接器而不是直接删除：PUT /connectors/<name>/pause。
• 保留 connector 配置和 offset topic，便于恢复。
• Sink 写入错误时先修数据或转换器，再恢复任务。
• 升级插件前备份 connector 配置和内部 topics。

常见坑

• plugin.path 配置错误，Worker 启动时找不到 connector class。
• 内部 topics 的副本因子为 1，Broker 故障后 offset 丢失。
• Source 连接器的 database.server.id 与现有 MySQL 复制冲突。
• Topic 分区数少于 Task 数，多余 Task 处于空闲。
• Schema 变更没有兼容策略，Sink 端解析失败。
• 直接删除 connector 而不是 pause，offset 和状态丢失。
• Worker 单节点部署，故障后没有高可用。

相关命令

• connect-distributed.sh — 启动分布式 Worker
• curl — Connect REST API
• kafka-topics — 查看内部 topics
• kafka-consumer-groups — 检查消费进度
• jq — 格式化 REST 返回