Schema 演进与批流一体：从表格式版本化到统一存储

0. 元信息

• 主题路径：docs/topics/data-engineering-basics/subtopics/schema-evolution-and-batch-stream-unified/README.md
• 父主题：data-engineering-basics
• 适合对象：会写 SQL/Python、跑过批流两条流水线、希望用 Lakehouse 把两者合一并支持 Schema 演进的工程师
• 建议周期：1.5～2 周，每周 10～14 小时
• 前置知识：SQL、Python、Airflow；建议先完成父主题第 5 阶段
• 最终目标：能用 Iceberg/Hudi 在同一份表上同时跑批流查询并保证结果一致；能管理 Schema 演进（加列、改类型、改分区）不阻塞读写

1. 学习路线

Schema 演进兼容性矩阵 → Iceberg Schema Evolution → Hudi Schema Evolution → Delta Schema Evolution → 批流一体存储 → 流批 SQL 引擎 → 端到端一致性验证

2. 阶段周数分配

每天 1.5～2 小时。1.5 周方案专注前 5 阶段；2 周方案多一天做端到端一致性。

3. 核心知识 / 产出 / 标准表

4. 第一周（每天 1.5~2 小时）

Day 1 约定：本计划使用 Iceberg on MinIO + Spark 3.5 + Flink 1.18。Schema 演进前先写兼容性矩阵（向后 / 向前 / 双向），再用 ALTER TABLE 操作。验证用 Time Travel 读历史数据。所有演进必须可回滚：演进失败时 CALL system.rollback_to_snapshot。

Day 7 执行次序

步骤 A —— 最小可用演进 + 一致性（60~90 分钟）

1. 起 Iceberg 表 2 列；
2. 演进 3 次（加列 + 改类型 + 重命名）；
3. 起 Flink Job 写 Iceberg；同时 Spark 批读；
4. 跑批流同查询 5 次，结果必须完全一致；
5. 写 notes/batch-stream-parity.md。

步骤 B —— 4 类边界用例（30~45 分钟）

Day 7 当天必完成步骤 A；步骤 B 至少完成 B1、B2。

5. 阶段通用验收

1. 不看答案独立重写 Iceberg 3 步 Schema 演进 + 批流一致性对比；
2. 用自己的话解释”为什么需要兼容性矩阵""为什么 Iceberg 用 column_id 而非列名""为什么批流结果可能不一致”；
3. 画一张图：兼容性矩阵表 + Iceberg manifest 演进图 + 批流架构图（三件套之一）；
4. 测试演进失败回滚、批流结果不一致、column mapping 漏配、Time Travel 失败、并发演进 5 类边界；
5. 准备至少 3 组自定义 schema 演进场景并贴出实际 snapshot 链；
6. 记录演进次数、批流同查询结果差异（应为 0）、查询 P95、并发写入冲突次数 4 个核心指标；
7. 能修改已有 schema（加列 / 改类型 / 换分区策略）并验证批流同查询结果一致。

交付存放：第 3 项的图、第 5 项的输出、第 6 项的指标，统一存到 week1/notes/ 或 week1/<table>/README.md。

6. 最终验收

• 独立设计并交付一套含 Schema 演进 + 批流一体的 Lakehouse，覆盖 Iceberg / Hudi / Delta 至少一种表格式；

• 至少完成 12 个实战项（分布建议）：

  子阶段	题目数量	难度	平台建议
  兼容性矩阵	2	Easy / Medium	Iceberg Spec + Confluent Blog
  Iceberg 演进	3	Medium	Iceberg 官方 Examples
  Hudi / Delta 演进	3	Medium	Apache Hudi / Delta 官方 Examples
  批流一体	4	Medium / Hard	Flink + Iceberg 案例

  约束：至少 6 项达到 Medium，至少 2 项达到 Hard；每项必须留可运行 SQL + 演进前后 snapshot 截图。

• 完成 1 个综合项目：批流一体数仓（Kafka → Flink → Iceberg + Spark 批读 + 5 次 Schema 演进演练）；

• 能用 15 分钟讲清兼容性矩阵、Iceberg column_id 设计、批流结果一致性验证、Schema 演进回归测试。

7. 综合项目

首选：批流一体数仓（必做：Kafka → Flink → Iceberg + Spark 批读 + Schema 演进演练）。
备选：跨表格式一致性对比（同一份数据写 Iceberg / Hudi / Delta，验证 Schema 演进与批流一致）。
备选：Schema Registry + 演进门禁（用 Confluent SR + Avro + Iceberg 演进 + 自动回滚）。

批流一体数仓必做要求：

• 输入：Kafka topic orders 实时事件 + 历史 CSV 批数据；
• 输出：必输出（1）Iceberg 表 dw.orders 同时被 Flink 流写与 Spark 批读；（2）5 次 Schema 演进（加列 2 次 + 改类型 1 次 + 重命名 1 次 + 改分区 1 次）；（3）批流同查询对比报告（至少 5 次查询）；（4）Time Travel 读历史数据截图；（5）兼容性矩阵 + 演进 SOP 文档；
• 算法 / 工程：Flink Job 配 enableCheckpointing(60_000, EXACTLY_ONCE)；Iceberg 配 TBLPROPERTIES ('format-version'='2')；批流对比用 EXCEPT 找差异；
• 进阶可选：用 Paimon 做流式 Lakehouse 对比；接 OpenLineage 暴露列级血缘。

任何综合项目都必须包含：

1. 需求说明与数据契约（源 schema、目标 schema、SLA、演进策略）；
2. 兼容性矩阵 + 演进 SOP 文档；
3. 核心代码（Iceberg DDL + Flink Job + Spark 批读 + 演进脚本）；
4. 边界测试（演进失败回滚、批流结果不一致、并发演进、Time Travel 失败、Schema Registry 不匹配）；
5. 可观测（Iceberg snapshot 元数据 + Flink checkpoint + Spark metrics）；
6. README（设计取舍、表格式选型、演进门禁规则）；
7. 复盘记录（模拟 1 次不兼容演进失败 + 1 次批流不一致，写 retrospective.md）；
8. notes/ 规范：

本主题贡献

Schema 演进的核心矛盾是”读历史数据用旧 schema vs 写新数据用新 schema”。本子主题专门讲清 Delta Lake / Iceberg + Avro + CDC 如何把这条边界守住——以 backward compat 为演进默认路径，Avro 为跨语言契约载体，CDC（Debezium / Flink CDC）为流式变更入口，Delta Lake column mapping 为兼容演进提供列追踪；不重复父主题的 Kafka 基础与 Airflow 编排。

3 职责

1. 用 Avro + Confluent Schema Registry 定义兼容性矩阵（BACKWARD / FULL / BACKWARD_TRANSITIVE），演进操作（加列 / 删列 / 改类型 / 重命名）必须落在 backward compat 范围内，下游 Avro 反序列化永不报错。
2. 用 Delta Lake delta.columnMapping.mode='name' + Iceberg column_id 做列追踪，重命名不破坏下游；演进前用 Time Travel 读历史验证，演进失败用 CALL system.rollback_to_snapshot 一键回退。
3. 用 CDC（Debezium 抓 MySQL binlog → Kafka topic cdc.orders → Flink CDC Source → Iceberg dw.orders）把 OLTP 变更流式接入 Lakehouse；批读用 Spark / Trino / Flink SQL 三引擎，跑同一查询对比批流结果必须 0 差异。

4 交付物

1. 一份 Avro 契约 + Schema Registry 兼容性配置（BACKWARD / FULL / BACKWARD_TRANSITIVE），含加列 / 改类型 / 重命名 3 步演进脚本，每次演进后老 Job 读旧数据 OK。
2. 一份 Delta Lake Schema 演进演练（开 columnMapping.mode='name' → 重命名列 → DESCRIBE HISTORY 查演进历史），含 Iceberg ADD COLUMN + RENAME COLUMN 对比表，配 format-version='2'。
3. 一份 CDC 接入管道（Debezium MySQL connector → Kafka cdc.orders → Flink MySqlSource + IcebergSink），含 binlog row 模式配置 + enableCheckpointing(60_000, EXACTLY_ONCE)。
4. 一份批流一致性对比报告（5+ 次同查询 Spark vs Flink SQL，结果 0 差异），含演进 SOP 文档（演进前兼容性检查 → 演进 → 灰度 → 全量 → 回滚预案）。

3 指标

1. Schema 演进 backward compat 通过率 100%（所有演进操作都通过 Schema Registry 兼容性校验，老消费者零修改）。
2. 批流同查询 0 差异（5 次查询 Spark vs Flink SQL 结果完全一致，含行数与聚合值）。
3. 演进回滚时长 < 1 min（CALL system.rollback_to_snapshot 执行到表 schema 与数据恢复）。

8. 推荐开源资料

许可证提示：Iceberg / Hudi / Delta / Flink / Spark 都是 Apache-2.0；Confluent Schema Registry 有社区版与商业版之分。复制 Apache 项目示例前请保留 LICENSE 与 NOTICE。默认做法是读规范后自己写演进脚本，而不是复制官方 Example。

默认使用顺序：先读 Iceberg Spec v2 schema section → 写兼容性矩阵 → 装 Iceberg on MinIO + Spark + Flink → 跑 Iceberg 3 步演进 → 对照 Hudi / Delta 演进差异 → 配 Flink 写 Iceberg + Spark 批读 → 跑批流同查询 5 次对比 → 用 Time Travel 验证演进前历史可读 → 写 notes/retrospective.md。

9. 学习资料汇聚（v0.3 自包含）

本节由本计划生成。链接指向原始材料或作者公开内容。规范会演进，记录时务必写明版本与日期。

9.1 背景与动机

传统数据仓库（Teradata、Snowflake、BigQuery）支持 Schema 演进，但实时性差；Kafka 等消息流支持实时，但不支持复杂查询。Lakehouse 通过 Iceberg/Delta/Hudi 等表格式，把”ACID 表 + 实时流写入 + 批 SQL 查询”三件事合并到同一份 Parquet 数据上。Schema 演进（Schema Evolution）从传统数据库继承而来，但在流处理场景下面临新约束——流读需要兼容历史事件，批读需要兼容新写。今天数据团队用”列添加 / 列重命名 / 列删除”等演进操作时，必须保证”读历史数据 + 写新数据”两种路径都不报错。

9.2 概念地图

flowchart LR
  Schema[Schema] --> Compat[兼容性矩阵]
  Compat --> Back[向后兼容]
  Compat --> Forw[向前兼容]
  Compat --> BackForw[双向兼容]
  Ice[Iceberg] --> SchemaEvo[Schema Evolution]
  Hu[Hudi] --> SchemaEvo
  Delta[Delta] --> SchemaEvo
  Lake[Lakehouse] --> Unify[批流一体]
  Unify --> BatchSQL[批 SQL Spark/Trino]
  Unify --> StreamSQL[流 SQL Flink]
  BatchSQL --> Verify[结果一致性]
  StreamSQL --> Verify

9.3 基础知识讲解

9.3.1 经典论文 / 规范

9.3.2 经典书籍

9.3.3 优秀博客与文档

9.3.4 核心人物

9.3.5 开发方法

9.4 经典问题与经典案例（≥5 道）

9.5 学习难点

概念难点

思维难点

工程难点

9.6 技术标准与接口

Entity

Scope

Iceberg/Hudi/Delta/Paimon 是表格式；它们解决 Schema 演进与 ACID；不替代计算引擎与存储。

Structure

• Iceberg：column_id 标识列，支持 schema evolution by ID。
• Hudi：SchemaProvider 维护 schema，支持 inline / external。
• Delta：columnMapping 模式，重命名列安全。
• Paimon：流式 Lakehouse，schema evolution 与流写入深度集成。

Ecosystem

• 表格式：Iceberg、Hudi、Delta、Paimon。
• 计算引擎：Spark、Trino、Flink、Presto、Hive。
• Schema 注册：Schema Registry、Avro、Protobuf。
• 演进测试：Schema Compatibility Checker。

Depth Tiers

本子主题目标：L3。

Source

• Apache Iceberg Spec：引用快照 2026-07-30。
• Apache Iceberg Schema Evolution：引用快照 2026-07-30。
• Apache Hudi Schema Evolution：引用快照 2026-07-30。
• Delta Schema：引用快照 2026-07-30。

9.7 关键代码

9.7.1 Iceberg Schema Evolution（加列 / 改类型 / 重命名）

-- 9.7.1 Iceberg Schema Evolution：三阶段演进
-- 初始 schema
CREATE TABLE dw.orders (
    order_id BIGINT,
    amount DECIMAL(10, 2)
) USING iceberg;

-- 阶段 1：加列
ALTER TABLE dw.orders ADD COLUMN user_id BIGINT;
ALTER TABLE dw.orders ADD COLUMN currency STRING;

-- 阶段 2：重命名（用 column id，旧数据仍可读）
ALTER TABLE dw.orders RENAME COLUMN amount TO total_amount;

-- 阶段 3：改类型（兼容类型：int -> bigint）
ALTER TABLE dw.orders ALTER COLUMN user_id TYPE BIGINT;

-- 验证：Time Travel 看历史数据
SELECT * FROM dw.orders FOR SYSTEM_TIME AS OF '2026-01-01';

9.7.2 Hudi Schema Evolution

# 9.7.2 PyHudi：Schema Evolution（HiveStyle 或 Inline）
from hudi import HudiTable

table = HudiTable(
    base_path="/data/hudi/orders",
    table_name="orders",
    storage_type="COPY_ON_WRITE",
)

# 加列
table.add_column("user_id", "bigint")
table.add_column("currency", "string")

# 重命名（需要 schema provider）
table.rename_column("amount", "total_amount", schema_provider="inline")

# 改类型
table.alter_column_type("user_id", "bigint")

9.7.3 批流一体：Flink 写 Iceberg + Spark 批读对比

// 9.7.3 Flink DataStream：写 Iceberg + 触发批读对比
DataStream<Order> orders = env.addSource(kafkaSource);
orders.addSink(new IcebergSink.Builder<Order>()
    .setTable(TableIdentifier.of("dw", "orders"))
    .setWriteParallelism(4)
    .build());

env.execute("stream-orders-to-iceberg");

// 同一份表用 Spark 批读，对比结果
// Spark SQL
val df = spark.read.format("iceberg").load("dw.orders")
df.createOrReplaceTempView("orders_batch")
spark.sql("SELECT user_id, SUM(total_amount) FROM orders_batch GROUP BY user_id")
  .show()

// 与 Flink SQL 跑相同查询对比结果

10. 常见误区

• 加列后下游不重新读，新列缺失；
• 改类型用不兼容类型（int → string）；
• 列重命名直接改 Parquet 文件名，破坏下游；
• 分区演进后旧分区失效；
• 批流结果差异当成”流处理特性”，不验证；
• 不写演进测试就直接生产演进；
• Iceberg column mapping 没开就重命名；
• Hudi schema provider 选错（inline vs external）；
• Delta schemaTracking 没启用就演进；
• 演进后没 Time Travel 验证；
• 演进没兼容矩阵文档；
• 用裸 Parquet 做流批一体。

11. 所有知识点分类

1. 编程语言
2. 数据结构与算法
3. 计算机基础
4. 工程技术
5. Web 与后端
6. 前端与客户端
7. 数据与人工智能
8. 项目与职业能力
9. 安全与可靠性

本计划归属：数据与人工智能 主 + 工程技术 辅。