CDC 与实时数据管道：Canal、Debezium、Flink CDC 的架构与实战

传统 ETL（拉取）：
  定时全量 SELECT → 传输 → 写入
  · 延迟：小时级
  · 成本：全表扫描，冲击源库
  · 捕获不到"删了哪些行/改了什么"（只有当前状态）

CDC（订阅）：
  解析数据库日志 → 逐条变更事件 → 实时推送
  · 延迟：毫秒级
  · 成本：只读日志，几乎零影响源库
  · 捕获完整变更（INSERT/UPDATE/DELETE + before/after）

· 实时数仓：业务库 → CDC → Kafka → Flink → DWS/DW
· 缓存刷新：订单变更 → 实时刷新 Redis 缓存
· 搜索索引：商品变更 → 实时同步到 ES
· 微服务数据共享：订单库变更 → 同步给下游服务
· 审计与合规：记录所有变更（before/after）
· 数据迁移/双写过渡

binlog 三种格式：
  STATEMENT：记录 SQL（节省空间，但非确定性函数有风险）
  ROW      ：记录每行变更前后值（推荐，信息最全）
  MIXED    ：混合，默认 statement 必要时转 row

CDC 必须用 ROW 格式：
  ROW 事件包含 before_image / after_image
  → 才拿得到完整的字段变化

binlog 三种日志记录方式：
  · 简单轮询读取日志文件
  · 基于 binlog 位点（position）断点续传（Canal 传统方式）
  · GTID（全局事务 ID）：更可靠的断点恢复（推荐）

-- 确认 binlog 配置
SHOW VARIABLES LIKE 'binlog_format';        -- 需为 ROW
SHOW VARIABLES LIKE 'gtid_mode';            -- 建议 ON
SHOW VARIABLES LIKE 'log_bin';              -- 需 ON

PG 的 CDC 走"逻辑复制"：
  · 物理复制：WAL 原始字节（用于流复制，对库不可读）
  · 逻辑复制：WAL 解码为 SQL 语义的变更流（CDC 用）
  · 通过 pgoutput / wal2json 插件输出 JSON 变更

PG CDC 特点：
  · 发布/订阅模式（PUBLICATION/SUBSCRIPTION）
  · 天然支持 schema 变化与过滤
  · Debezium 等工具消费逻辑复制输出

CDC 必须记录"读到哪了"（offset/position/GTID）：
  · 进程重启 → 从上次位点继续，不丢不重
  · 存到 Kafka offset / state store / DB

Exactly-Once 挑战：
  · 源日志消费（offset）与下游写入（Kafka/DB）跨系统
  → 需"幂等 + 去重"或依赖 Flink checkpoint 的端到端一致

架构：
  Canal Instance 伪装成 MySQL 从库
    → 拉取 binlog
    → 解析为结构化变更事件（CanalEntry）
    → 推给下游（Kafka / RocketMQ / DB）

适用：
  · 纯 MySQL → Kafka/MQ 的轻量管道
  · 已有 Kafka 生态的团队
  · 对 Debezium 生态陌生、想更贴近阿里实践

# canal.properties / instance.properties
canal.instance.master.address=mysql-master:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=***
canal.instance.connectionCharset=UTF-8
canal.instance.filter.regex=orders_db\\..*
# 位点持久化
canal.instance.gtidon=false
canal.instance.master.position=

架构：
  Debezium Connector（Kafka Connect 插件）
    → 连接源库（MySQL/PG/MongoDB/Oracle/SQL Server...）
    → 解析日志 → 输出到 Kafka topic（每个表一个 topic）
    → Kafka Connect 管理位点（offset storage）

特性：
  · 原生 Kafka 生态（Avro/JSON、schema registry 集成）
  · 内置字段 flatten、过滤、列裁剪
  · 支持 schema 变更事件（DDL）
  · Exactly-Once 依赖 Kafka 事务 / Flink

// Debezium MySQL connector 配置（Kafka Connect REST API）
{
  "name": "orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-master",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "***",
    "database.server.id": "5400",
    "database.include.list": "orders_db",
    "table.include.list": "orders_db.orders,orders_db.order_items",
    "topic.prefix": "cdc",
    "schema.history.internal.kafka.topic": "schema-history",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

// 消费到的事件（Kafka topic: cdc.orders_db.orders）
{
  "op": "u",                                // c=create, u=update, d=delete
  "before": {"id": 1001, "status": "PENDING", "amount": 500},
  "after":  {"id": 1001, "status": "PAID",   "amount": 500},
  "source": {"db": "orders_db", "table": "orders", "lsn": 123456}
}

Flink CDC Connector 把"日志解析"内嵌进 Flink 作业：
  · Flink SQL 直接同步：Source 是 CDC 流，Sink 是目标表
  · 端到端 Exactly-Once（checkpoint + 幂等）
  · 支持动态 schema 演进
  · 一条 SQL 完成"实时入湖/入仓"

典型用法：
  · 实时同步 MySQL → Kafka / Hudi / ClickHouse
  · 双活/跨区同步
  · 实时维表关联

-- Flink SQL：MySQL → Hudi 实时入湖
CREATE TABLE orders_cdc (
  id INT, status STRING, amount DECIMAL(10,2),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'mysql-master',
  'database-name' = 'orders_db',
  'table-name' = 'orders',
  'server-id' = '5400-5404'
);

CREATE TABLE orders_ods (
  id INT, status STRING, amount DECIMAL(10,2),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'hudi',
  'path' = 's3://warehouse/ods/orders'
);

INSERT INTO orders_ods SELECT * FROM orders_cdc;

选型要点：
  · 已上 Kafka Connect → Debezium
  · 主要 MySQL + 阿里云栈 → Canal
  · 要实时入湖/入仓/复杂计算 → Flink CDC
  · 跨多种数据库统一 → Debezium / Flink CDC
  · 团队熟悉度 + 运维成本

业务库(MySQL/PG)
    │ binlog/WAL
    ▼
CDC (Canal/Debezium/Flink CDC)
    │ 变更事件流
    ▼
Kafka (每个表一个 topic / 分区按主键)
    │ 多路分发（消费者组）
    ├──► Flink → 实时数仓 / ClickHouse / Hudi
    ├──► Redis 缓存刷新
    ├──► Elasticsearch 索引同步
    └──► 审计/归档存储

关键设计：
  · 分区键 = 主键（保证同一行事件有序）
  · 下游消费幂等（upsert）
  · schema 变化通过 registry 协调

初次建管道 = 全量快照 + 增量追平：
  1. 先做全量初始同步（如已有备份/历史数据）
  2. 启动 CDC 增量（位点对齐）
  3. 合并去重 → 保证不丢不重

Flink CDC 内置此能力：启动时自动快照 + 增量衔接

业务库加列/改类型 → CDC 事件结构变化 → 下游 schema 要跟上

三种处理：
  · 统一 schema registry（Avro + Confluent Schema Registry）
  · 宽松 JSON（新增字段容忍）
  · Flink 动态列（重建表 schema）

Debezium 处理：
  · DDL 事件也发到 schema-history topic
  · 下游可从 schema history 重建表结构

· 只加列（可空/默认值）：向后兼容，直接推进
· 改列类型：评估影响面（类型不兼容需重放）
· 删列：下游若引用则报错，需提前治理
· 建议：源库 DDL 走审批 + 通知 CDC 管道

· 位点持久化（Kafka Connect offset / Flink checkpoint）
· 源库 binlog 保留时长（建议 24-72h，给管道容错窗口）
· 下游幂等（主键 upsert）
· 断点续跑：管道重启从位点继续，无重复无丢失
· 监控：事件延迟（lag）、处理吞吐、失败重试

同一条数据（同主键）的事件必须有序：
  · Kafka 分区按主键 hash → 同 key 同分区 → 有序
  · 下游单分区消费或按 key 聚合处理
  · 跨表（如 order + order_item）天然独立 topic，无需保序

· 源库 binlog 解析跟不上 → 事件积压 → 延迟上升
  → 扩容 CDC 消费者 / 增大 Kafka 分区
· 下游慢 → 背压传播
  → 用 Kafka 缓冲解耦 + 独立扩容消费组
· 大事务/批量变更 → 事件风暴
  → 下游幂等 + 限流（最大吞吐阈值）

· 版本升级：先扩容新版本实例 → 灰度切流
· 位点重置：确认不丢数据才允许
· 表结构变更：走 DDL 审批并同步更新下游
· 故障演练：模拟 binlog 中断 / 消费组故障

□ binlog_format=ROW + GTID 开启（MySQL）
□ 授权 CDC 账号（REPLICATION SLAVE/CLIENT 等）
□ 选型：Kafka Connect/Debezium vs Flink CDC vs Canal
□ Kafka topic 设计（每表一 topic，主键分区）
□ 全量+增量衔接（初次建管道）
□ schema registry / 演进策略
□ 位点持久化 + 断点续跑验证
□ 下游幂等 upsert
□ 延迟/积压监控 + 告警
□ binlog 保留时长评估

· 用轮询 SELECT 代替 CDC（重复拉取、伤源库）→ 应走日志
· 下游直接依赖"事件顺序"做业务逻辑（跨表无序）
· 把 CDC 当无限准确（大事务丢细粒度）→ 配合幂等
· 忽略 schema 演进（加列即崩）