在现代数据架构中，系统通常由数十个异构数据源组成。传统全量 ETL 在实时性和资源开销上越来越难满足需求。变更数据捕获（CDC）提供了一种在源数据发生变更时实时捕获并传播这些变更的机制，而 Kafka Connect 与 Debezium 的组合已成为业界最受欢迎的 CDC 解决方案之一。
一、CDC 核心概念与适用场景
CDC 是一种用于识别和捕获数据库数据变更的技术，允许下游以增量方式获取 Insert、Update 和 Delete 操作。CDC 主要分为两种实现模式：基于查询的 CDC 和基于日志的 CDC。
基于查询的 CDC 依赖时间戳或版本号列，通过周期性轮询识别变更。这种方式实现简单，但延迟高、无法捕获删除、对源库造成持续查询压力。基于日志的 CDC 直接解析数据库事务日志（MySQL binlog、PostgreSQL WAL、SQL Server CDC 表），以非侵入方式获取完整的变更流，包括结构化的前后镜像数据。
以下是两种模式的对比：
| 对比维度 | 基于查询的 CDC | 基于日志的 CDC（Debezium） |
|---|---|---|
| 实现复杂度 | 低，仅需 SQL 轮询 | 中，需解析数据库日志 |
| 数据延迟 | 高（秒级至分钟级） | 低（毫秒级至秒级） |
| 删除事件捕获 | 不支持或需软删除 | 原生支持，含完整前后镜像 |
| 源库性能影响 | 中，持续轮询消耗资源 | 极低，仅读取已落盘的日志 |
| 事务边界感知 | 无 | 支持，可还原完整事务 |
| Schema 变更感知 | 无 | 支持 Schema Evolution |
| 适用场景 | 简单同步、无实时性要求 | 实时数仓、事件驱动架构 |
CDC 的典型应用场景包括：构建实时数据仓库、实现读写分离与缓存一致性、支撑微服务间的数据事件总线，以及满足审计合规的数据血缘追踪。
二、Debezium 架构设计与核心组件
Debezium 是一个专为 CDC 而生的分布式平台，架构建立在 Apache Kafka Connect 之上，充分利用 Kafka 的持久化、容错和高吞吐特性。核心组件包括 Connector、Kafka Connect Runtime 和 Kafka Cluster。
Connector 是执行单元，分为 Source Connector 和 Sink Connector。Source Connector 连接源数据库并捕获变更事件，将事件序列化为 JSON 或 Avro 后写入 Kafka Topic。每个 Connector 实例由一个或多个 Task 组成，Task 是实际执行捕获和传输的工作进程。Kafka Connect 会自动进行任务重新平衡，确保负载均匀分布。
Offset 持久化是 CDC 场景的关键特性。Kafka Connect 将 Source Connector 读取的日志位点定期提交至 connect-offsets Topic，Connector 重启后从上次位点继续消费，确保数据不丢失、不重复（至少一次语义，结合幂等 Sink 可实现精确一次）。
Schema Registry 用于管理数据格式的演进，当使用 Avro 或 Protobuf 格式时，会为每个 Topic 维护 Schema 版本。Confluent Schema Registry 与 Apicurio Registry 是生产中最常用的实现。
以下是 Debezium 部署的 Docker Compose 示例：
version: '3.8'
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.6.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
kafka:
image: confluentinc/cp-kafka:7.6.0
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
connect:
image: debezium/connect:2.5
ports:
- "8083:8083"
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: 1
CONFIG_STORAGE_TOPIC: connect-configs
OFFSET_STORAGE_TOPIC: connect-offsets
STATUS_STORAGE_TOPIC: connect-status
schema-registry:
image: confluentinc/cp-schema-registry:7.6.0
ports:
- "8081:8081"
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
Debezium Connect 镜像已预装主流 Source Connector 插件。如需自定义连接器，可通过挂载插件目录或构建自定义镜像扩展。
三、Source Connector 实战：MySQL 与 PostgreSQL
Source Connector 是 CDC 管道的入口。Debezium 为 MySQL、PostgreSQL、SQL Server、MongoDB、Oracle 等数据库提供成熟连接器。
3.1 MySQL Source Connector
MySQL Connector 依赖 binlog 捕获变更。配置前需确认 binlog 已开启且采用 ROW 格式，否则无法提供完整行级变更信息。
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';
若 log_bin 为 OFF，在 my.cnf 中添加：
[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7
为 Debezium 创建数据库用户，需具备 binlog 读取和快照权限：
CREATE USER 'debezium'@'%' IDENTIFIED BY 'dbz-secret';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, LOCK TABLES
ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;
通过 Kafka Connect REST API 注册 MySQL Source Connector：
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "mysql-cdc-connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "mysql",
"database.port": "3306",
"database.user": "debezium",
"database.password": "dbz-secret",
"database.server.id": "184054",
"database.server.name": "dbserver1",
"database.include.list": "inventory",
"table.include.list": "inventory.customers,inventory.orders",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "schema-changes.inventory",
"snapshot.mode": "initial",
"tombstones.on.delete": "true",
"decimal.handling.mode": "string",
"time.precision.mode": "connect"
}
}'
snapshot.mode 参数控制首次启动行为。initial 模式先执行一致性快照导入历史数据，再切换到 binlog 流式读取。其他可选值包括 schema_only（仅同步结构）、when_needed（无法定位位点时触发）、never（跳过快照）和 incremental（增量快照，适合超大数据集）。
数据捕获后，每个表对应一个 Topic，默认命名格式 <server.name>.<database>.<table>。事件消息包含 before、after、source、op 和 ts_ms 字段，其中 op 标识操作类型：c 插入、u 更新、d 删除、r 快照读取。
3.2 PostgreSQL Source Connector
PostgreSQL Connector 通过逻辑解码读取 WAL。需将 wal_level 设为 logical，并创建逻辑复制槽和解码插件。
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;
SELECT pg_reload_conf();
Debezium 推荐 pgoutput 插件（PostgreSQL 10+ 原生支持）。创建具有复制权限的用户和 Publication：
CREATE USER debezium WITH REPLICATION LOGIN PASSWORD 'dbz-secret';
GRANT USAGE ON SCHEMA inventory TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA inventory TO debezium;
CREATE PUBLICATION dbz_publication FOR TABLE inventory.customers, inventory.orders;
注册 PostgreSQL Source Connector：
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "postgres-cdc-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"database.hostname": "postgres",
"database.port": "5432",
"database.user": "debezium",
"database.password": "dbz-secret",
"database.dbname": "inventory",
"database.server.name": "pgserver1",
"schema.include.list": "inventory",
"table.include.list": "inventory.customers,inventory.orders",
"plugin.name": "pgoutput",
"publication.name": "dbz_publication",
"slot.name": "debezium_slot",
"snapshot.mode": "initial",
"heartbeat.interval.ms": "10000",
"decimal.handling.mode": "string",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false"
}
}'
heartbeat.interval.ms 和 slot.name 是两个关键参数。PostgreSQL 逻辑复制槽只在产生数据时推进 WAL 位点，长时间无变更会阻塞 WAL 回收导致磁盘膨胀，心跳机制通过定期发送伪事件解决此问题。slot.name 必须唯一稳定，Connector 切换时需绑定同一复制槽。
四、Sink Connector 实战：Elasticsearch 与 S3
数据流入 Kafka 后，通过 Sink Connector 写入下游系统。
4.1 Elasticsearch Sink Connector
Elasticsearch Sink Connector 适合构建实时搜索索引。Insert 和 Update 映射为 Index/Update API，Delete 映射为 Delete API。
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "es-sink-connector",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "2",
"topics": "dbserver1.inventory.customers",
"connection.url": "http://elasticsearch:9200",
"key.ignore": "false",
"schema.ignore": "true",
"behavior.on.malformed.documents": "warn",
"behavior.on.null.values": "delete",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false"
}
}'
behavior.on.null.values 设为 delete 时，Debezium 发送的 tombstone 消息会触发 Elasticsearch 删除对应文档。生产环境建议开启幂等写入，通过 write.method=upsert 结合 Kafka Key 作为文档 ID 实现。数据量大时合理配置 batch.size 提升吞吐。
4.2 S3 Sink Connector
将 CDC 数据持久化到 S3 是数据湖和归档的常见需求。Parquet 或 Avro 格式便于 Athena、Spark 和 Presto 分析。
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "s3-sink-connector",
"config": {
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"tasks.max": "4",
"topics": "dbserver1.inventory.customers,dbserver1.inventory.orders",
"s3.region": "us-east-1",
"s3.bucket.name": "my-cdc-data-lake",
"flush.size": "10000",
"rotate.interval.ms": "600000",
"format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
"partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
"partition.duration.ms": "3600000",
"path.format": "'\''year'=YYYY/'month'=MM/'day'=dd'\''",
"timestamp.extractor": "RecordField",
"timestamp.field": "ts_ms",
"schema.compatibility": "FULL"
}
}'
配置使用基于时间的分区策略，按 ts_ms 划分到 year=2026/month=09/day=01 目录。flush.size（10000 条）和 rotate.interval.ms（10 分钟）任一满足即触发刷写到 S3。Parquet 格式有效压缩存储空间，同时保留完整 Schema 信息。
五、Schema Evolution 策略与版本兼容
长时间运行的 CDC 管道中，源库表结构不可避免地会变更。新增列、修改类型、删除列若处理不当，会导致下游解析失败。Schema Evolution 目标是在变化时保持管道连续性。
Debezium 通过两种机制应对 Schema 变更：一是事件内嵌 Schema 信息（JSON Converter + schemas.enable=true），下游根据 schema 动态解析 payload；二是与 Schema Registry 集成，以 Avro/Protobuf 实现 Schema 集中管理和版本演进。
Avro 格式下，Schema Registry 为每个 Topic 维护单调递增的版本号。Debezium 检测结构变更后注册新 Schema，Registry 按兼容性策略决定是否允许。常见级别：
- BACKWARD（默认值）：用新 Schema 可读旧数据，消费者先升级。
- FORWARD：用旧 Schema 可读新数据，生产者先升级。
- FULL：同时具备 BACKWARD 和 FORWARD，适合同步升级。
- NONE：不做校验，允许任意变更。
新增可选列通常同时兼容 BACKWARD 和 FORWARD；新增必填列破坏 BACKWARD；删除列破坏 FORWARD；修改类型是否兼容取决于 Avro 映射规则。
数据湖场景建议用 FULL 兼容配合 Parquet Schema 合并：
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("CDC-Schema-Evolution") \
.config("spark.sql.parquet.mergeSchema", "true") \
.getOrCreate()
df = spark.read \
.option("mergeSchema", "true") \
.parquet("s3a://my-cdc-data-lake/topics/dbserver1.inventory.customers/")
df.createOrReplaceTempView("customers")
spark.sql("SELECT after.*, op, ts_ms FROM customers WHERE op IN ('c','u','r')").show(10)
MySQL 的 database.history.kafka.topic 参数会将所有 DDL 变更记录到专用 Topic，下游可消费该 Topic 感知完整结构演进。运维中建议建立 Schema 变更审批流程，危险操作（删除列、改主键、变类型）先在测试环境验证影响。
六、单条消息转换（SMT）与数据治理
Kafka Connect 的 Single Message Transform（SMT）在数据流入或流出时对单条记录进行轻量级转换。SMT 运行在 Connector 进程内，无需额外流处理引擎，适合简单 ETL。
6.1 提取变更后状态
Debezium 默认消息包含完整 Envelope（before、after、source、op），很多 Sink 场景只需 after。ExtractNewRecordState 转换器可展开 Envelope，生成扁平化记录。
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.add.fields": "op,ts_ms,db"
add.fields 将操作类型、时间戳、数据库名附加到扁平记录中，方便审计或分区。delete.handling.mode=rewrite 将删除事件重写为带 __deleted=true 的记录，而非 tombstone，这对不支持 null value 的下游很有用。
6.2 字段过滤与重命名
ReplaceField 可过滤敏感字段或重命名列。
"transforms": "filter,rename",
"transforms.filter.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.filter.blacklist": "password,ssn",
"transforms.rename.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.rename.renames": "customer_name:name"
6.3 Topic 重定向
RegexRouter 可将 Debezium 自动生成的 Topic 映射为下游更易识别的名称。
"transforms": "route",
"transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.route.regex": "dbserver1\\.inventory\\.(.*)",
"transforms.route.replacement": "cdc-$1"
上述配置将 dbserver1.inventory.customers 重命名为 cdc-customers。
SMT 仅能单条处理记录，无法实现 Join、窗口聚合或状态化计算。链路复杂时，应将计算逻辑下沉至 Kafka Streams、Flink 或 Spark Streaming，让 Kafka Connect 回归数据搬运的核心定位。
FAQ
Q1: Debezium Connector 重启后是否会丢失数据或产生重复？
不会丢失。Offset 记录了日志位点，重启后从最近位点继续。但 Kafka Connect 是至少一次语义，极端情况下（提交 Offset 后宕机）可能产生少量重复。下游应设计幂等处理，或在 Sink 开启幂等写入。
Q2: PostgreSQL 逻辑复制槽导致 WAL 无限增长，磁盘爆满如何处理？
根本原因是复制槽位点长时间未推进。开启 heartbeat.interval.ms 保持心跳；监控 pg_replication_slots 延迟；PostgreSQL 13+ 可设 max_slot_wal_keep_size 限制保留上限；定期检查清理废弃复制槽。
Q3: Schema 变更后下游报错 “Schema not found” 如何解决？
确认 Schema Registry 兼容策略与变更类型匹配。新增必填字段破坏 BACKWARD 兼容，可临时降级为 FORWARD/NONE，更根本的是协调上下游升级顺序。检查 database.history 配置是否正确，历史 Schema 是解析增量变更的基础。
Q4: MySQL 大事务导致 Kafka 消息过大，Broker 拒收怎么办？
优先拆分业务大事务。无法避免时，可增大 Broker message.max.bytes 和 Topic max.message.bytes，或调优 max.batch.size 和 max.queue.size。也可使用 ExtractNewRecordState 展开后仅保留关键字段，减小消息体积。
总结
Kafka Connect 与 Debezium 为现代数据架构提供了强大、灵活的 CDC 能力。从基于数据库日志的低侵入捕获，到通过 Source Connector 注入 Kafka，再到 Sink Connector 分发至 Elasticsearch、S3 等异构存储，整个链路建立于开放标准之上。Schema Evolution 和 SMT 进一步提升了管道在复杂环境中的适应性。
落地 CDC 项目时，建议按数据源日志特性（binlog / WAL / oplog）选择 Connector 并确保源库配置正确；建立位点、Schema 和吞吐的监控告警；简单转换用 SMT 控制成本，复杂计算适时引入流处理引擎。通过合理设计，CDC 管道可以成为企业实时数据基础设施中最稳定高效的一环。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。