kafka

06. Kafka Connect 数据集成

Kafka Connect Source/Sink Connector、Debezium CDC、数据管道构建与 Kafka Connect 生产实践。

Leeting Yan 2026-08-14 6 分钟阅读 2732 字

1. Kafka Connect 概述

Kafka Connect 是 Kafka 官方提供的数据集成框架，用于在 Kafka 与外部系统（数据库、文件、对象存储、搜索引擎等）之间建立可靠的数据管道。它通过插件化的 Connector 机制，无需编写代码即可实现数据导入导出。

1.1 Connect 架构

┌─────────────────────────────────────────────────────────────┐
│                    Kafka Connect 集群                        │
│                                                             │
│  ┌──────────────┐      ┌──────────────┐      ┌──────────┐  │
│  │ Worker 1     │      │ Worker 2     │      │ Worker 3 │  │
│  │              │      │              │      │          │  │
│  │ [Connector A]│      │ [Connector B]│      │（备用）   │  │
│  │   Task 1     │      │   Task 1     │      │          │  │
│  │   Task 2     │      │   Task 2     │      │          │  │
│  └──────────────┘      └──────────────┘      └──────────┘  │
│         │                     │                             │
└─────────┼─────────────────────┼─────────────────────────────┘
          │                     │
    Source Connector      Sink Connector
          │                     │
  ┌───────▼──────┐       ┌─────▼──────┐
  │   外部系统    │       │  外部系统   │
  │  MySQL/Mongo │       │ ES/S3/HDFS │
  └──────────────┘       └────────────┘

1.2 Connector、Task、Worker 的关系

组件	角色	说明
Connector	逻辑定义	定义数据源/目标的配置，进行 Task 拆分
Task	执行单元	真正执行数据复制的工作单元，无状态
Worker	运行进程	承载 Task 运行的 JVM 进程，支持集群和分布式调度

2. Source Connector

Source Connector 从外部系统读取数据，写入 Kafka Topic。

2.1 Debezium CDC（变更数据捕获）

Debezium 是最常用的 Source Connector，基于数据库日志（binlog/WAL）实现增量数据捕获。

Debezium 工作原理：

MySQL Binlog               Debezium          Kafka
  INSERT ──→ [捕获] ──→  ──→ Avro/JSON ──→ db.shop.users
  UPDATE ──→ [解析] ──→  ──→ 变更事件   ──→ db.shop.orders
  DELETE ──→ [转换] ──→  ──→           ──→ db.shop.inventory

初始快照（Snapshot）：
  首次启动时，全量读取表数据 → 写入 Kafka → 然后切换到 Binlog 增量

2.2 Debezium 事件格式

{
  "before": { "id": 1, "name": "Alice", "age": 30 },
  "after":  { "id": 1, "name": "Alice", "age": 31 },
  "source": {
    "version": "2.5.0.Final",
    "connector": "mysql",
    "name": "db-shop",
    "ts_ms": 1705315200000,
    "db": "shop",
    "table": "users",
    "pos": "binlog.001:1234",
    "row": 0
  },
  "op": "u",
  "ts_ms": 1705315200001
}

字段	说明
`before`	更新/删除前的数据
`after`	更新/插入后的数据
`source`	元信息（数据库、表、binlog 位置）
`op`	操作类型：c=create, u=update, d=delete, r=read(快照)
`ts_ms`	处理时间戳

2.3 Debezium 配置

{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "db-shop",
    "database.include.list": "shop",
    "table.include.list": "shop.users,shop.orders",
    "snapshot.mode": "initial",
    "topic.prefix": "cdc",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes.shop",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

3. Sink Connector

Sink Connector 从 Kafka Topic 读取数据，写入外部系统。

3.1 JDBC Sink Connector

{
  "name": "jdbc-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "3",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://postgres:5432/analytics",
    "connection.user": "analytics",
    "connection.password": "secret",
    "auto.create": "true",
    "auto.evolve": "true",
    "insert.mode": "upsert",
    "pk.fields": "order_id",
    "pk.mode": "record_key"
  }
}

3.2 Elasticsearch Sink Connector

{
  "name": "es-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "3",
    "topics": "orders",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "key.ignore": "false",
    "schema.ignore": "true",
    "transforms": "extractKey",
    "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractKey.field": "order_id"
  }
}

4. 常用 Connector 列表

Connector	类型	来源	用途
Debezium	Source	Debezium 社区	MySQL/PostgreSQL/MongoDB CDC
JDBC Source	Source	Confluent	轮询查询数据库
FileStream	Source	Kafka 官方	读取日志文件
S3 Source	Source	Confluent	从 S3 读取数据
JDBC Sink	Sink	Confluent	写入关系型数据库
Elasticsearch Sink	Sink	Confluent	写入 ES 索引
S3 Sink	Sink	Confluent	写入 S3（Parquet/JSON）
HDFS Sink	Sink	Confluent	写入 HDFS
Redis Sink	Sink	社区	写入 Redis
InfluxDB Sink	Sink	社区	写入时序数据库

5. Kafka Connect 运行时

5.1 Standalone vs Distributed

模式	适用场景	特点
Standalone	开发测试、简单任务	单进程，配置存储在本地文件
Distributed	生产环境	多 Worker 集群，配置存储在 Kafka topic（offset.storage.topic）

5.2 部署 Distributed 模式

# connect-distributed.properties
bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
group.id=connect-cluster

# 配置存储topic
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status

# 启动 Worker
connect-distributed.sh config/connect-distributed.properties

# 注册 Connector（REST API）
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @mysql-source-connector.json

# 查看状态
curl http://localhost:8083/connectors/mysql-source-connector/status

6. Transform（数据转换）

Kafka Connect 支持单消息转换（SMT，Single Message Transform），在数据流转过程中转换格式。

{
  "transforms": "maskField,extractValue",
  "transforms.maskField.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskField.fields": "password,ssn",
  "transforms.extractValue.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.extractValue.field": "id"
}

Transform	作用
`ExtractField`	提取嵌套字段
`ReplaceField`	重命名/删除字段
`MaskField`	脱敏（替换为 null 或掩码）
`ValueToKey`	将 value 中的字段设为 key
`TimestampConverter`	时间格式转换
`Filter`	按条件过滤消息

7. 生产实践

7.1 高可用配置

Worker 数量 ≥ 2（防止单点）
tasks.max = min(分区数, Worker 数量 × 单 Worker 承载能力)

监控指标：
- source-record-active-count（活跃记录数）
- put-batch-avg-time-ms（批处理平均时间）
- sink-record-send-rate（发送速率）

7.2 Schema Registry 集成

Schema Registry 确保 Source 和 Sink 之间的数据格式一致。

{
  "name": "avro-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schemas.enable": "true"
  }
}

7.3 Exactly-Once 语义

Kafka Connect 从 2.6 开始支持 Source Exactly-Once：

{
  "config": {
    "exactly.once.support": "requested",
    "transaction.boundary": "poll",
    "transactions.timeout.ms": "60000"
  }
}

7.4 常见问题

问题	原因	解决
数据延迟	Sink 消费慢	增加 tasks.max、优化目标数据库写入
字段不匹配	Schema 变更	启用 auto.evolve、使用 Schema Registry
消息格式错误	非 Avro/JSON 格式	配置正确的 Converter（key.converter、value.converter）
连接断开	网络不稳定	配置重试策略、连接池保持
Offset 重置导致重复消费	Worker 重启或 Rebalance	启用幂等写入、消费端去重

8. Kafka Connect 监控与运维

8.1 JMX 监控指标

指标	含义	告警阈值
`source-record-poll-rate`	Source 每秒拉取记录数	< 预期吞吐的 50%
`sink-record-read-rate`	Sink 每秒读取记录数	< 预期吞吐的 50%
`put-batch-avg-time-ms`	Sink 批处理平均时间	> 5000ms
`task-count`	当前运行的 Task 数	< 配置的 tasks.max
`connector-total-tasks`	Connector 总 Task 数	确认无失败 Task
`task-running-ratio`	运行中 Task 占比	< 1.0

8.2 Prometheus + Grafana 集成

# JMX Exporter 配置
lowercaseOutputName: true
rules:
  - pattern: kafka.connect<type=connector-task-metrics, connector=(.+), task=(.+)><>(.+)total
    name: kafka_connect_$3_total
    labels:
      connector: "$1"
      task: "$2"

8.3 REST API 运维

# 查看所有 Connector
curl http://localhost:8083/connectors

# 验证 Connector 是否健康
curl http://localhost:8083/connectors/mysql-source-connector/status | jq .
# 输出：{"name":"mysql-source-connector","connector":{"state":"RUNNING"},"tasks":[{"id":0,"state":"RUNNING"}]}

8.4 生产环境 checklist

检查项	说明	是否必须
Distributed 模式部署	至少两个 Worker 保证高可用	✅
配置自动备份	config.storage.topic 已持久化	✅
Converter 统一	Source 和 Sink 使用一致序列化格式	✅
监控告警	JMX + Prometheus 已接入	✅
Schema Registry	Avro/Protobuf Schema 已注册	推荐
Exactly-Once	关键业务启用事务语义	视场景

查看 Connector 配置

curl http://localhost:8083/connectors/mysql-source-connector/config

暂停 Connector（ graceful 停止）

curl -X PUT http://localhost:8083/connectors/mysql-source-connector/pause

恢复 Connector

curl -X PUT http://localhost:8083/connectors/mysql-source-connector/resume

重启失败的 Task

curl -X POST http://localhost:8083/connectors/mysql-source-connector/tasks/0/restart

删除 Connector

curl -X DELETE http://localhost:8083/connectors/mysql-source-connector


---

## 9. 高阶场景

### 9.1 CDC + 流处理 + 分析管道

MySQL (业务数据库)
│
▼ Debezium Source Connector
Kafka Topic (db.shop.orders)
│
├──→ Kafka Streams (实时聚合：订单统计)
│ └── Kafka Topic (aggregated-orders)
│
├──→ Elasticsearch Sink (全文检索)
│
└──→ S3 Sink (Parquet 格式，数据湖)
└── Athena / Spark SQL (离线分析)


### 9.3 测试与调试技巧

```bash
## 查看 Connector 的 Task 日志
curl http://localhost:8083/connectors/mysql-source-connector/tasks/0/log

## 发送测试消息验证 Sink
kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic test-orders \
  --property parse.key=true \
  --property key.separator=":"
> order-001:{"id":"001","amount":100}

## 使用 FileStream Source 快速验证 Connect 链路
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "file-source-test",
    "config": {
      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
      "tasks.max": "1",
      "file": "/tmp/test-input.txt",
      "topic": "connect-test"
    }
  }'

9.2 多集群数据镜像

跨数据中心或云厂商的数据同步：

{
  "name": "mirror-maker-2",
  "config": {
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "source.cluster.alias": "us-east",
    "target.cluster.alias": "us-west",
    "topics": "orders,payments,inventory",
    "replication.factor": 3,
    "sync.topic.configs.enabled": "true"
  }
}

总结

Kafka Connect 是构建数据管道的利器，它通过插件化的 Connector 机制屏蔽了底层数据源的复杂度。Debezium CDC 让数据库变更实时流入 Kafka，JDBC 和 Elasticsearch Sink 则将数据分发到各个分析系统。在生产环境中，务必采用分布式部署模式，配置 JMX 监控和 Prometheus 告警，同时关注 Exactly-Once 语义和 Schema 兼容性。一个成熟的数据集成平台往往是 Kafka Connect、Kafka Streams 和 Schema Registry 协同工作的结果。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页