全栈攻城狮

Appearance

Kafka Connect 深入 ​

概述 ​

Kafka Connect 是 Kafka 生态的数据集成框架：Source Connector 把外部数据导入 Kafka，Sink Connector 把 Kafka 数据导出到外部系统。支持 Standalone（单机）与 Distributed（集群）两种模式，REST API 统一管理。本文讲透 Connector 开发、API 管理与部署模式。

一、Kafka Connect 定位 ​

外部系统 ⇄ Kafka
  Source Connector：外部 → Kafka
  Sink Connector：Kafka → 外部
特性说明
框架分布式集成
管理REST API
扩展插件生态
可靠Offset 管理
适用：
  数据库同步（Debezium）
  HDFS/S3 导出
  文件/日志采集

二、核心概念 ​

概念说明
Connector连接器定义（逻辑）
Task执行单元（物理）
Worker运行进程
Config连接器配置
Offset消费位点
Schema数据格式
关系：
  Connector（1）→ Task（1..n）
  Worker 集群运行 Task

三、Source Connector ​

3.1 作用 ​

外部数据 → Kafka Topic：
  JDBC、文件、MQ、S3...

3.2 工作原理 ​

轮询外部数据源：
  拉取新数据 → 转为 SinkRecord → 写入 Kafka
  Offset 记录位置
接口说明
start初始化
poll拉取数据
flush提交
stop停止

3.3 示例：JDBC Source ​

json
{
  "name": "jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:mysql://mysql:3306/db",
    "connection.user": "root",
    "connection.password": "***",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "mysql-"
  }
}
模式说明
incrementing自增列
timestamp时间戳列
bulk全量
增量组合时间+自增

四、Sink Connector ​

4.1 作用 ​

Kafka Topic → 外部系统：
  HDFS、ES、JDBC、S3...

4.2 工作原理 ​

消费 Kafka：
  按 Topic/分区
  批量写入外部
  记录 offset（幂等）
接口说明
start初始化
put写入批次
flush提交
stop停止

4.3 示例：JDBC Sink ​

json
{
  "name": "jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:mysql://target:3306/dw",
    "connection.user": "root",
    "connection.password": "***",
    "topics": "orders",
    "insert.mode": "upsert",
    "pk.mode": "record_key",
    "pk.fields": "order_id"
  }
}
配置说明
insert.modeinsert/upsert/update
pk.mode主键策略
batch.size批量
幂等唯一键

五、Connector 开发 ​

5.1 开发流程 ​

实现接口：
  Source：SourceConnector + SourceTask
  Sink：SinkConnector + SinkTask
打包成插件 → 放 plugins 目录

5.2 关键实现 ​

java
// Source Task 核心
public class MySourceTask extends SourceTask {
  public List<SourceRecord> poll() throws InterruptedException {
    // 拉取外部数据 → SourceRecord（topic, key, value）
  }
}

// Sink Task 核心
public class MySinkTask extends SinkTask {
  public void put(Collection<SinkRecord> records) {
    // 写入外部系统
  }
}
开发要点说明
OffsetSource 记录位点
幂等Sink 可重放
配置ConfigDef
版本元数据

六、REST API ​

6.1 管理接口 ​

方法路径说明
GET/connectors列表
POST/connectors创建
GET/connectors/状态
PUT/connectors/{name}/config更新
DELETE/connectors/删除
POST/connectors/{name}/pause暂停
POST/connectors/{name}/resume恢复
GET/connectors/{name}/tasks任务
GET/connectors/{name}/status详细状态
bash
# 查看状态
curl http://connect:8083/connectors/jdbc-source/status

6.2 运维操作 ​

操作场景
创建新同步任务
更新修改配置（热更新）
暂停/恢复维护
删除下线
状态字段：
  RUNNING / PAUSED / FAILED
  TASK 状态与错误

七、Standalone 模式 ​

7.1 特点 ​

单进程运行：
  配置本地文件
  单 Worker
properties
# connect-standalone.properties
bootstrap.servers=kafka:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.file.filename=/tmp/connect.offsets
特点说明
简单单机
单点无 HA
适用开发/小规模

八、Distributed 模式 ​

8.1 特点 ​

集群运行：
  多 Worker
  任务自动分配/再平衡
  Offset 存 Kafka
properties
# connect-distributed.properties
bootstrap.servers=kafka:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status
特性说明
HA节点故障任务迁移
扩展加节点
存储Kafka Topic
再平衡任务重分配

8.2 状态存储 ​

Topic用途
connect-configs连接器配置
connect-offsets消费位点
connect-status状态信息
分布式模式生产推荐：
  高可用 + 扩展性

九、部署与运维 ​

9.1 插件管理 ​

plugins 目录：
  打包插件（fat jar）
  配置 plugin.path
  多版本隔离

9.2 监控 ​

指标说明
任务状态RUNNING/FAILED
延迟端到端
吞吐记录/秒
错误重试/失败
监控方案：
  JMX / Prometheus
  任务状态轮询

9.3 常见问题 ​

问题处理
任务失败看日志/状态
数据重复Sink 幂等
延迟高扩 Task/并行
插件缺失打包部署
再平衡频繁检查节点

常见问题速查 ​

问题要点
Source vs Sink进 Kafka vs 出 Kafka
单机还是集群Standalone vs Distributed
怎么管理REST API
开发 ConnectorConnector+Task 接口
数据不丢Offset + 幂等