业务库改了字段类型——下游CDC作业全挂了，Schema演进怎么处理？

Schema演进（Schema Evolution）是CDC方案落地后最让人头大的问题。曾经业务方一次把user_id从INT改成BIGINT，下游30+个CDC作业集体红屏，运维通宵恢复。这道题面试官想听的不是"加try-catch"，而是你有没有完整的Schema管理体系——从源头管控、传输兼容、到下游容错。

业务场景：user_id类型变更引发的全链路雪崩

业务团队某天上线一个DDL：ALTER TABLE users MODIFY user_id BIGINT（原来是INT）。原因是用户量突破21亿，INT放不下了。但下游30多个CDC作业（Flink/Spark/Hologres）的Schema是写死的INT，瞬间集体异常：

症状清单（按时间顺序）：
  10:05  业务执行ALTER TABLE，DDL在binlog里产生QUERY事件
  10:06  Canal/Flink CDC原样转发到Kafka，下游消费方还没感知
  10:07  新的INSERT进来，user_id是个超过INT_MAX的值
  10:08  Flink作业反序列化失败：NumberFormatException
         作业进入restart loop -> Checkpoint失败 -> 数据停止流转
  10:10  下游Hologres实时表停止更新，运营大盘报警
  10:30  人工介入，逐个作业改Schema、重启、追数据
  16:00  全部恢复，期间丢了6小时的实时数据

方案一：源端管控——DDL评审 + 兼容性规则

最有效的方案永远是"事前预防"。后来定了一套DDL评审规则，所有上线必须走DBA+数据团队双签。

-- ✓ 安全变更：自动通过
ALTER TABLE users ADD COLUMN nickname VARCHAR(64);  -- 加列
ALTER TABLE users MODIFY VARCHAR(32) -> VARCHAR(64); -- 同类型扩容
 
-- ⚠ 警告变更：需要数据团队评审
ALTER TABLE users MODIFY user_id INT -> BIGINT;  -- 兼容扩容(下游要改)
ALTER TABLE users ADD INDEX idx_email(email);    -- 加索引(影响同步压力)
 
-- ✗ 禁止变更：必须用迁移方案
ALTER TABLE users DROP COLUMN status;            -- 删列(下游必挂)
ALTER TABLE users MODIFY phone INT -> VARCHAR;   -- 不兼容类型变更
ALTER TABLE users CHANGE COLUMN old_name new_name; -- 改名
 
-- 禁止变更的替代方案：双写过渡
ALTER TABLE users ADD COLUMN phone_str VARCHAR(20);  -- 1.加新列
-- 2.业务双写一段时间(等下游全部消费完phone_str)
-- 3.最后删除老列phone(此时下游已经不依赖)

适用场景：有DBA团队+数据团队的中大型公司；业务方愿意配合走流程。

不适用：初创团队没流程；业务方话语权强不愿意配合（这种就只能事后补救）。

方案二：传输层契约——Schema Registry + Avro

自描述格式比JSON好的核心原因：消费端不需要知道字段类型，schema是带在数据里的。Confluent的Schema Registry是工业标准，支持FORWARD/BACKWARD/FULL三种兼容性模式。

// 生产端（CDC采集器）注册Schema
Schema schema = SchemaBuilder.record("User")
    .fields()
    .requiredLong("user_id")          // 注意是Long(对应BIGINT)
    .optionalString("nickname")       // 可选字段，加列时安全
    .name("phone").type().nullable().stringType().noDefault()
    .endRecord();
 
// Schema Registry会校验：新schema是否BACKWARD兼容老schema
// BACKWARD: 老消费者能读新数据 (典型: 加可选字段OK,删字段不OK)
// FORWARD:  新消费者能读老数据 (典型: 删字段OK,加必填字段不OK)
// FULL:     双向兼容
schemaRegistry.register("users-value", schema);  // 不兼容时直接拒绝
 
// 消费端（Flink）按schema id反序列化
// 老消费者拿到新数据时，多余字段自动丢弃
// 新消费者拿到老数据时，缺失字段用default值