数据湖与湖仓一体

本章目标：能讲清数据湖、湖仓一体和表格式，理解 Iceberg 的 ACID、时间旅行、schema 演化和增量读，并能为订单数据设计一张入湖表。 前置知识：读过 数仓分层、Hive 分区；了解 Parquet 和对象存储更顺。

1. 为什么需要它

离线数仓通常把数据放在 Hive 表里：先定义 schema，再按分区写 Parquet，查询时用 Hive/Spark。它适合结构化数据和 T+1 批处理，但遇到下面几类需求会别扭：

• 原始日志是 JSON、埋点字段经常变，先建严格 schema 太重；
• 数据要能被 Spark、Flink、Trino 同时读写，不想被一个引擎绑死；
• 需要按主键更新、删除，而不只是追加分区；
• 需要查“昨天 10 点的数据长什么样”，或者误删后回滚；
• 只想读新增数据，不想每次全表扫描。

数据湖先解决“便宜地存各种原始数据”：HDFS、S3、OSS 上放 Parquet/ORC/JSON，读的时候再解析，也就是 schema-on-read。但裸文件只是“文件堆”，没有事务、没有快照、没有统一元数据，多个作业一起写还可能互相覆盖。

湖仓一体在数据湖的文件之上加一层表格式，让这批文件具备表的能力：ACID、时间旅行、schema 演化、增量读。典型代表是 Iceberg、Hudi、Paimon。

面试高频：湖仓一体不是“数据湖取代数据仓库”，而是“对象存储 + 开放表格式 + 多种计算引擎”，让湖上的数据能按数仓的方式管理。

2. 核心概念

2.1 数仓、数据湖、湖仓一体

湖仓的分层可以画成这样：

Spark / Flink / Trino / Doris（计算引擎）
                 ↓
Iceberg / Hudi / Paimon（表格式：元数据、事务、快照）
                 ↓
Parquet / ORC / Avro（开放文件格式）
                 ↓
HDFS / S3 / OSS（廉价存储）

计算和存储分离：数据放在对象存储，引擎按需拉起；表格式负责“哪些文件构成这张表、当前版本是什么”。

2.2 表格式解决了什么

把一张 Iceberg 表拆开看，它不是一个目录里的散文件，而是有层级的元数据：

catalog（表名 → 当前 metadata 文件）
   ↓
metadata.json（schema、分区规则、快照列表）
   ↓
snapshot → manifest list → manifest（数据文件清单 + 列统计）
   ↓
data files（Parquet）

由此得到几个能力：

• ACID：一次写入生成新的数据文件和 manifest，最后原子地切换 snapshot 指针。要么全部可见，要么全部不可见，不会读到写了一半的数据。
• 时间旅行：每个 snapshot 都保留当时的文件清单，可以按 snapshot ID 或时间戳查询历史版本。
• schema 演化：加列、改列名、调整字段顺序时只改 metadata，历史 Parquet 文件不动；读旧文件时按字段 ID 映射。
• 增量读：对比两个 snapshot 的 manifest，只读新增或变更的文件，适合增量同步和 CDC 入湖。
• 隐藏分区：分区由列表达式派生，例如按 days(dt) 分区，查询写 WHERE dt = ... 即可裁剪，不必额外维护分区列。
• 分区演化：业务变化时可以改分区策略，旧数据仍按旧规则读，不需要重写全表。

注意：时间旅行依赖历史文件还在。快照过期（snapshot expiration）和文件清理会删掉旧数据，保留多久要按审计和回滚需求配置。

2.3 Copy-on-Write 和 Merge-on-Read

行级更新有两种实现：

Merge-on-Read 读的时候要把基础文件和删除文件合并，长期不 compaction 会越来越慢；Copy-on-Write 每次更新重写整个文件，写放大明显。没有绝对好坏，看读写比例。

2.4 Iceberg、Hudi、Paimon

三者都在解决“湖上建表”，差异在生态和写入模型。选型先看团队主力引擎：Spark 为主可以优先 Iceberg/Hudi，Flink 实时链路为主可以重点评估 Paimon。

2.5 小文件、compaction 和 catalog

流式写入如果每分钟提交一次，一天会产生大量小文件，元数据和读取都会变慢。表格式不会自动消灭小文件，需要：

• 控制提交频率，先攒批再写；
• 配置 compaction，定期把小文件合并成大文件；
• 清理过期快照和孤儿文件，避免存储无限增长。

catalog 负责保存“表名到当前 metadata 文件”的映射，常见实现有 Hive Metastore、JDBC、REST Catalog、Nessie。catalog 是元数据入口，生产要考虑高可用和备份；丢了 catalog，数据文件还在，但表可能找不回来。

3. 动手示例

3.1 用 PyIceberg 建一张本地表

数据湖的真实环境通常是 Spark/Flink + 对象存储，本章用 PyIceberg + SQLite catalog + 本地目录做本地模拟，验证表格式的基本行为。

pip install "pyiceberg[sql-sqlite]" pyarrow

新建 lake_demo/iceberg_demo.py：

from decimal import Decimal
from pathlib import Path

import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog

catalog = SqlCatalog(
    "local",
    **{
        "uri": "sqlite:///lake_demo/catalog.db",
        "warehouse": Path("lake_demo/warehouse").resolve().as_uri(),
    },
)
catalog.create_namespace_if_not_exists("ecom")

schema = pa.schema([
    ("order_id", pa.string()),
    ("category", pa.string()),
    ("amount", pa.decimal128(18, 2)),
    ("dt", pa.date32()),
])

table = catalog.create_table_if_not_exists("ecom.dwd_order", schema=schema)

batch = pa.table({
    "order_id": ["O1001", "O1002"],
    "category": ["手机", "图书"],
    "amount": [Decimal("2999.00"), Decimal("59.00")],
    "dt": pa.array(["2018-08-01", "2018-08-01"], type=pa.date32()),
})
table.append(batch)
print(table.scan().to_arrow())

warehouse 是数据文件目录，catalog.db 存表名和当前 metadata 位置。append 会生成新的 Parquet 文件和 snapshot，而不是改旧文件。

3.2 时间旅行

history = list(table.history())
for snapshot in history:
    print(snapshot.snapshot_id, snapshot.timestamp_ms, snapshot.summary)

first_id = history[0].snapshot_id
old = table.scan(snapshot_id=first_id).to_arrow()
print(old)

误删、误更新后，可以按快照读回旧版本，再写回当前表。前提是对应快照和文件还没被清理。

3.3 Spark SQL 写法

生产更常见的是 Spark 或 Flink 读写 Iceberg。下面用 Spark SQL 展示 DDL、更新和时间旅行（需要 Iceberg Spark runtime，本地练习可用 Docker 镜像，属于本地模拟）：

CREATE TABLE local.ecom.dwd_order (
  order_id STRING,
  category STRING,
  amount DECIMAL(18, 2),
  dt DATE
) USING iceberg
PARTITIONED BY (days(dt));

INSERT INTO local.ecom.dwd_order VALUES
('O1001', '手机', 2999.00, DATE '2018-08-01'),
('O1002', '图书', 59.00, DATE '2018-08-01');

MERGE INTO local.ecom.dwd_order t
USING (SELECT 'O1001' AS order_id, '家电' AS category, 3099.00 AS amount) s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET category = s.category, amount = s.amount;

SELECT * FROM local.ecom.dwd_order VERSION AS OF 1234567890;
ALTER TABLE local.ecom.dwd_order ADD COLUMN channel STRING;

MERGE INTO 的更新在 COW 表上重写文件，在 MOR 表上写删除文件；VERSION AS OF 用 snapshot ID 查历史。实际语法和连接器版本有关，跑之前先确认 Iceberg 与 Spark 的版本匹配。

4. 常见坑

1. 把数据湖当“随便放文件”。没有表格式和 catalog，时间一长没人知道哪些文件属于哪张表。
2. 只建表不管 compaction。流式小文件越积越多，查询越来越慢，元数据也膨胀。
3. 以为时间旅行永久有效。快照过期和文件清理后旧版本就没了，保留策略要提前定。
4. schema 演化以为什么都能改。改类型通常只支持安全的向上兼容，删列、改语义要评估下游。
5. 分区过细。按小时甚至按分钟分区，小文件爆炸；按数据量选天/月，或依赖隐藏分区。
6. 并发写不做冲突处理。两个作业同时提交 snapshot 可能冲突，要配置重试和提交隔离。
7. 把湖仓当高并发查询引擎。湖表适合批查和明细，报表接口仍应同步到 Doris/ClickHouse/MySQL。
8. catalog 不做备份。元数据服务挂了或丢了，恢复成本很高。

5. 练习

练习 1：选存储方案

三个场景分别选 Hive 数仓、数据湖裸文件还是湖仓表：每天 10 亿条 JSON 埋点长期留存；订单表要按主键频繁更新并被多个引擎查询；财务日报只读、结构稳定、要求强 schema。

练习 2：设计订单入湖表

订单 CDC 要入湖，字段有 order_id、customer_id、order_status、pay_amount、update_time。给出分区、文件格式、表格式和读写方式。

练习 3：误删数据怎么恢复

一张 Iceberg 表误执行了删除，如何利用时间旅行恢复？说出步骤和前提。

练习 4：小文件治理

Flink 每分钟往湖表提交一次，运行一周后查询变慢。分析原因并给出至少三个处理办法。

6. 面试延伸

面试题：什么是湖仓一体？

是什么：在数据湖的廉价存储上增加开放表格式，让文件具备 ACID、schema 管理、时间旅行和增量读能力，计算引擎可以多样。

解决什么问题：裸数据湖缺少事务和元数据，难更新、难回滚；传统数仓成本高、格式封闭。湖仓把两者折中。

我的实操边界：我用 PyIceberg 在本地建过表、追加数据、按快照读历史，理解 metadata/manifest/snapshot 的结构；没有在生产环境运维过 Iceberg 集群和 PB 级数据湖。答题时强调“存储 + 表格式 + 引擎”三层，不把湖仓说成单一产品。

面试题：Iceberg 为什么能支持 ACID 和时间旅行？

是什么：Iceberg 用 metadata.json、snapshot、manifest 管理数据文件，写入先生成新文件和新快照，再原子切换指针。

解决什么问题：裸文件写入没有原子性，读者可能看到写了一半的数据；也没有版本概念，无法查历史。

我的实操边界：我验证过 append 后 snapshot 增加、按旧 snapshot 查询；没有做过并发提交冲突和高可用 catalog 的实战。回答时讲清“数据文件不可变、元数据原子提交、旧快照保留”三点即可。

面试题：Iceberg、Hudi、Paimon 怎么选？

是什么：三者都是湖仓表格式，Iceberg 引擎中立，Hudi 强在 upsert 和增量，Paimon 与 Flink 流式生态结合紧。

解决什么问题：不同团队的主力引擎、写入模式和实时性要求不同。

我的实操边界：我只在本地用过 Iceberg，Hudi 和 Paimon 了解定位；没有做过生产选型和压测。选型看主力计算引擎、是否需要高频 upsert、CDC 链路、社区和运维成本，不能只背“Iceberg 最流行”。

7. 本章验收

• 能用自己的话区分数据仓库、数据湖和湖仓一体，并说出各自适合的场景。
• 能画出 catalog、metadata、snapshot、manifest、数据文件的层级关系。
• 能解释 Iceberg 的 ACID、时间旅行、schema 演化和增量读分别靠什么实现。
• 能说出 Copy-on-Write 和 Merge-on-Read 的读写代价与适用场景。
• 能独立运行本地 PyIceberg 示例，追加数据并按旧快照查询。
• 能为订单 CDC 设计分区、文件格式、表格式和 compaction 策略。
• 能说出湖仓不适合直接承接高并发点查的原因，并给出替代方案。