Kafka Connect vs Flink CDC——CDC到Kafka用哪个方案更稳？

// 注册一个MySQL CDC connector
POST /connectors
{
  "name": "mysql-orders-source",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql-readonly",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "****",
    "database.server.id": "5400",
    "topic.prefix": "mysql.order_db",
    "table.include.list": "order_db.orders,order_db.payments",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema_history.order_db",
    "snapshot.mode": "initial",
    // SMT: 简单转换可以做
    "transforms": "unwrap,routing",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.routing.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.routing.regex": "mysql.order_db.(.*)",
    "transforms.routing.replacement": "ods_$1"
  }
}

-- Flink SQL一段搞定:CDC采集+维度关联+脱敏+写Kafka
CREATE TABLE orders_cdc (
  id BIGINT, user_id BIGINT, phone STRING, amount DECIMAL(10,2),
  PRIMARY KEY (id) NOT ENFORCED
) WITH ('connector' = 'mysql-cdc', ...);
 
-- 维度表(用户中心)
CREATE TABLE dim_users (
  user_id BIGINT, vip_level INT, country STRING,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH ('connector' = 'jdbc', ...);
 
-- Kafka sink
CREATE TABLE orders_kafka (
  id BIGINT, user_id BIGINT, phone_masked STRING,
  amount DECIMAL(10,2), vip_level INT, country STRING
) WITH ('connector' = 'kafka', 'topic' = 'ods_orders_enriched', ...);
 
-- 一段INSERT完成: CDC采集 + 关联维度 + 字段脱敏 + 写Kafka
INSERT INTO orders_kafka
SELECT
  o.id, o.user_id,
  CONCAT(SUBSTR(o.phone, 1, 3), '****', SUBSTR(o.phone, 8)) AS phone_masked,
  o.amount, u.vip_level, u.country
FROM orders_cdc o
LEFT JOIN dim_users FOR SYSTEM_TIME AS OF o.proctime u ON o.user_id = u.user_id;

适用场景：已有Flink集群、需要在采集时做ETL、希望减少链路跳数。

两方案核心差异

维度	Kafka Connect	Flink CDC
架构定位	Kafka基础设施	Flink作业
运维方式	REST API管理connector	提交Flink作业
资源占用	Connect Worker进程	Flink TM/JM
容错恢复	task级restart	Checkpoint+Savepoint
处理能力	弱(只SMT)	强(完整Flink SQL)
Schema演进	支持(Avro+Registry)	支持(2.4+)
扩展性	动态加tasks.max	改作业并行度+rescale
监控生态	JMX+Connect REST	Flink UI+Metrics
典型用法	采集即转储	采集+转换+多sink

场景化选型

▶ 选 Kafka Connect 的场景:
  - 已有Kafka集群,Connect Worker部署简单
  - 只是"原样转储"到Kafka,无ETL需求
  - 团队倾向声明式配置,不想写代码
  - 跨语言/多团队消费,Avro+Registry成熟
  - SaaS产品(如Confluent Cloud)集成方便
 
▶ 选 Flink CDC 的场景:
  - 已有Flink集群,不想再维护Connect
  - 采集时需要关联维度表/字段脱敏/过滤
  - 需要写多个Sink(Kafka+Iceberg+Hologres)
  - 已经用Flink SQL,人才储备友好
  - 希望全量+增量自动衔接(Connect需要单独snapshot配置)
 
▶ 混合方案(大公司常见):
  - 核心库稳定无ETL → Kafka Connect
  - 需要实时融合的库 → Flink CDC
  - 团队基础设施: Kafka Connect托管 + Flink计算平台

面试高频追问

追问一：Kafka Connect的Exactly-Once怎么保证？

Connect框架本身保证source connector的at-least-once（offset commit到Kafka的__consumer_offsets）。Exactly-Once需要source connector自己实现幂等输出——Debezium MySQL Connector通过把binlog position存在Kafka source topic的header里，重启时从这个position继续，不会重复发送。但下游消费者还是要做幂等，因为重启窗口可能有重复。Connect 2.x开始支持EOS sink（精确一次写入sink），source侧仍依赖connector实现。

追问二：两者的Schema管理有什么差异？

Kafka Connect天然集成Confluent Schema Registry——connector发出的每条消息都带schema id，下游解析时从Registry取schema。Schema演进通过Registry的兼容性策略管控（FORWARD/BACKWARD）。Flink CDC的schema管理在2.4之前比较弱（Schema硬编码在DDL里），2.4+引入Pipeline模式才支持自动schema evolution。生产上如果schema变更频繁，Kafka Connect+Registry的方案更成熟。

追问三：Flink CDC的Source会不会比Connect对源库压力更大？

两者底层都是Debezium，对源库的binlog dump压力相同。差异在snapshot阶段：Kafka Connect Debezium默认要锁全局或锁表（高风险），Flink CDC 2.x无锁snapshot更友好。Flink CDC的并行度（server-id范围）可能产生多个并行snapshot reader，理论上对源库IO更高，但通过限流和走从库可以控制。生产上两者的源库压力都可控，关键是不连主库。

实战踩坑：Connect任务静默失败导致数据停止流转

有次发现下游Kafka消费者lag一直为0，本来以为是好事，结果业务报表少了一天数据——查Kafka Connect的REST API才发现connector状态是FAILED但没有任何告警。Connect默认不会主动告警，只能通过API轮询。后来加了Prometheus监控+JMX exporter，把connector状态、task状态、binlog position作为关键指标，FAILED超过60秒就钉钉告警。

# Connect监控API(必须自己定时轮询)
curl -s http://kafka-connect:8083/connectors/mysql-orders-source/status | jq
{
  "name": "mysql-orders-source",
  "connector": { "state": "RUNNING" },
  "tasks": [
    { "id": 0, "state": "FAILED",  // ← 关键!静默失败
      "trace": "Cannot find binlog file: mysql-bin.001234..." }
  ]
}
 
# 加Prometheus jmx exporter
java -javaagent:jmx_prometheus_javaagent.jar=8080:config.yaml \
     -jar connect-distributed.sh ...
 
# 关键告警指标
# kafka_connect_connector_status{state="FAILED"} > 0 持续60s → P0
# kafka_connect_task_failed_count > 0 → P1
# debezium_mysql_milliseconds_behind_source > 60000 → P2

解决方案：Kafka Connect监控不能依赖默认行为，必须自己加Prometheus监控+主动告警。同时connector要配restart.policy（Connect 3.x支持自动重试），减少人工介入。这个坑出过一次后我把所有Connect集群都加了完整监控。

总结

Kafka Connect和Flink CDC不是"哪个更好"的问题，是"哪个适合"的问题。Connect适合纯转储的稳定基础设施，Flink CDC适合需要在采集时做处理的场景。大公司经常并存，核心稳定库用Connect、需要融合的用Flink CDC。面试时讲"两者定位差异+场景选型+混合方案"，比说"Flink CDC更先进"分高得多。

下一篇聊一个很烦的实战题：历史数据补采跑了8小时，怎么设计回溯采集机制？

你们公司CDC到Kafka用的哪个方案？踩过哪些坑？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长