60.3 数据湖与湖仓一体
一句话
湖仓一体不是「湖 + 仓」两个系统拼起来，而是用开放文件格式 + 表格式元数据 + catalog 把数据库的 ACID、时间旅行、schema 演进搬到对象存储上；存储与计算解耦，但元数据成为新的瓶颈与故障点。
三个概念先分清
| 数据湖 | 数据仓库 | 湖仓一体 | |
|---|---|---|---|
| 存储 | 对象存储 / HDFS 上的原始文件 | 厂商私有格式 + 本地/块存储 | 对象存储上的开放格式（Parquet / ORC） |
| Schema | 读时定义（schema-on-read） | 写时定义（schema-on-write） | 写时定义 + 可演进（field-id） |
| 事务 | 无（文件即真相） | 强 ACID | 表级 ACID（快照隔离） |
| 更新 / 删除 | 重写分区或整表 | UPDATE / DELETE | MERGE / DELETE + 快照 |
| 查询性能 | 依赖引擎与文件布局 | 优化器 + 索引 + 统计 | 列存 + 分区裁剪 + 统计；点查仍弱 |
| 成本 | 最低 | 最高 | 接近湖，接近仓的能力 |
| 治理 | 弱（靠目录约定） | 强（库表权限） | 依赖 catalog 与表格式 |
🔑 「湖仓」多出来的不是存储，而是表格式这一层元数据：它记录「哪些文件构成这张表的这个版本」。没有这一层，湖就只是「一堆 Parquet 文件」，ACID、时间旅行、并发写都无从谈起。
对象存储逼出来的设计
对象存储只提供三个语义：PUT、GET、DELETE（外加 LIST）。它的语义细节（原子性、LIST 一致性、分页）见 10.2 对象存储语义。它没有：
- 原子
rename/ 目录原子替换； - 跨对象的原子多写；
- 跨厂商一致的
LIST语义（S3 自 2020 年起强一致，但分页、缓存与性能问题仍在，其他对象存储未必）； - 文件锁与事务。
Hive 表的做法是「目录即分区、文件即数据」，查询时靠 LIST 找文件。这会带来三个生产问题：
- 并发写互相看不见：A 正在写
dt=2026-10-01/part-1，B 同时LIST，可能看到半个文件或看不到新文件； - 提交不原子：任务失败后留下「半套文件」，没有标记能区分「已完成」与「写坏了」；
- LIST 昂贵：分区下几万个文件时，查询计划阶段就要几秒到几十秒。
表格式的解法是把「文件列表」从目录搬到元数据文件里：
catalog（指针：当前 metadata 路径）
└── metadata/v3.json ← 表结构、分区规范、快照列表
└── snapshot s2 ← 一次提交的完整视图
└── manifest-list ← 这次快照包含哪些 manifest（Avro）
└── manifest ← 每个数据文件的路径、分区、行数、min/max
└── data/xxx.parquet ← 不可变数据文件
一次写入的提交过程：
- 写新的数据文件（不可变，不覆盖旧文件）；
- 写新的 manifest，列出「旧文件 + 新文件」；
- 写新的 snapshot 与 metadata，指向新 manifest；
- 用一次原子操作替换 catalog 里的「当前 metadata 指针」（对象存储上是条件写 / CAS，或由 catalog 服务保证）。
第 4 步是提交点：替换成功 = 提交成功；替换失败 = 这次写入对读者不可见，数据文件成为孤儿（由清理任务回收）。读者永远只看到「某个完整 snapshot」，看不到半成品。
⚠️ 对象存储通常没有通用的 CAS，所以生产上必须有一个带条件的 catalog（REST catalog、Hive Metastore、Glue、Nessie、Unity 等）。「把 metadata 路径写进 S3 就完事」的方案，在并发写下会丢提交。
快照隔离、时间旅行、隐藏分区
| 能力 | 实现 | 读者看到什么 |
|---|---|---|
| 快照隔离 | 读固定 snapshot_id，文件不可变 | 写入过程中仍是旧快照，不会读到半成品 |
| 时间旅行 | 旧 snapshot 仍指向旧文件，直到过期清理 | VERSION AS OF / TIMESTAMP AS OF 查历史 |
| 回滚 | 把「当前指针」指回旧 snapshot | 秒级回滚，不重写数据 |
| 隐藏分区 | 分区规范用变换（days(ts)、bucket(16, user_id)），查询谓词自动推导 | SQL 不用写分区列，也不会写错 |
| Schema 演进 | 每列有稳定 field-id，文件按 id 映射 | 加列、改名、提升类型不重写历史 |
隐藏分区解决的是 Hive 分区的三个老问题：
- 分区列暴露给用户：用户必须知道
dt的格式，写错就全表扫描； - 分区粒度写死：按天分区后想按小时查，只能扫一天；
- 分区列基数失控：按
user_id分区会造出几百万个小目录。
Iceberg 的 PARTITIONED BY (days(ts), bucket(16, user_id)) 把「怎么分」交给表定义，查询 WHERE ts >= '2026-10-01' 时由引擎自动做分区裁剪。
三种表格式：Iceberg / Delta / Hudi
| Apache Iceberg | Delta Lake | Apache Hudi | |
|---|---|---|---|
| 元数据 | metadata.json + manifest list + manifest（Avro） | _delta_log/*.json + checkpoint（Parquet） | timeline（commit / deltacommit）+ 文件组 |
| 提交协议 | catalog CAS / 乐观并发 | 单写者日志（多写者需协调） | timeline + 锁 |
| 更新方式 | copy-on-write；v2 支持 merge-on-read 删除文件 | copy-on-write；deletion vectors | COW / MOR 可选 |
| 强项 | 多引擎（Spark / Flink / Trino / Snowflake）、隐藏分区 | Spark / Databricks 生态、流批一体 | Upsert 密集、近实时入湖 |
| 适合 | 开放湖仓、多引擎共享 | Databricks 为主、Spark 为主 | CDC 高频更新、Hudi 运维团队 |
| 主要代价 | catalog 与 manifest 运维；小文件合并 | 日志膨胀、并发写限制 | 运维复杂、概念多 |
💡 我的判断：多引擎是硬需求就选 Iceberg；团队已经在 Databricks 上就选 Delta；CDC 高频 upsert 且能养运维就选 Hudi。 三者都能做「湖仓」，差别在生态与运维，不在名词。 行级更新还有一条实现分叉，选型时必须问清：
| 方式 | 写入 | 读取 | 适用 | 代价 |
|---|---|---|---|---|
| COW（copy-on-write） | 重写受影响的整个数据文件 | 直接读，无合并 | 读多写少、批式更新 | 写放大，单文件越大越贵 |
| MOR + position delete | 只写「删了哪个文件第几行」 | 读时按位置过滤 | 高频删除、CDC | 读放大，删除文件要定期合并 |
| MOR + equality delete | 只写「删掉满足条件的行」 | 读时按谓词过滤 | 按主键/条件批量删 | 谓词越复杂，读越慢 |
⚠️ MOR 不是「免费的快」：它把成本从写入转移到读取与合并。删除文件积压不清理，查询会越读越慢；而 COW 在宽表上重写一个文件可能就是几百 MB。
湖仓不是银弹：五笔代价
| 代价 | 表现 | 缓解 |
|---|---|---|
| 元数据成为单点 | catalog 挂 → 所有表不可读 | catalog 多副本 + 客户端缓存 + 降级到只读 |
| 小文件 | 查询计划慢、对象存储 GET 多 | 定时 compaction、控制写入批次大小 |
| 无索引 | 主键点查要扫很多文件；列存本身没有 B+ 树索引（见 44.16 OLAP 与列式存储） | 用 OLAP 引擎 + 排序聚簇；点查回 OLTP |
| 并发写冲突 | 两个任务写同一分区，后者提交失败 | 分区隔离 + 重试 + 排队 |
| 权限割裂 | 表权限给了，文件仍可被直读 | 关闭直读、用统一 catalog + 短期凭证 |
代码：200 行实现「快照 + 时间旅行 + 乐观并发」
下面的程序不是 Iceberg 实现，只保留四个关键机制：不可变数据文件 + manifest、快照提交、快照隔离、版本冲突重试。
#!/usr/bin/env python3
"""最小湖仓表格式：快照 / manifest / 时间旅行 / 乐观并发（无第三方依赖）。"""
import copy
import json
import os
import tempfile
import uuid
from pathlib import Path
class Conflict(Exception):
pass
class Table:
def __init__(self, root):
self.root = Path(root)
(self.root / "data").mkdir(parents=True, exist_ok=True)
(self.root / "metadata").mkdir(parents=True, exist_ok=True)
self.meta_path = self.root / "metadata" / "table.json"
if not self.meta_path.exists():
self._write({
"format": "mini-lake/1", "uuid": str(uuid.uuid4()), "version": 0,
"schema": [{"name": "id", "type": "long"},
{"name": "dt", "type": "string"},
{"name": "amount", "type": "long"}],
"snapshots": [], "current": None,
})
def _read(self):
return json.loads(self.meta_path.read_text())
def _write(self, meta):
tmp = self.meta_path.with_suffix(".json.tmp")
tmp.write_text(json.dumps(meta, ensure_ascii=False, indent=1))
os.replace(tmp, self.meta_path) # 单文件原子替换 = 提交点
def _commit(self, base_version, meta):
if self._read()["version"] != base_version:
raise Conflict(f"版本已变：base={base_version}")
meta["version"] = base_version + 1
self._write(meta)
def _snapshot(self, meta, op, manifest, summary):
ids = [s["id"] for s in meta["snapshots"]]
return {"id": (max(ids) + 1) if ids else 1, "parent": meta["current"],
"operation": op, "manifest": manifest, "summary": summary}
def _write_files(self, rows, partition_key="dt", max_rows=10):
files, by_part = [], {}
for r in rows:
by_part.setdefault(r.get(partition_key), []).append(r)
for part, group in by_part.items():
for i in range(0, len(group), max_rows):
chunk = group[i:i + max_rows]
rel = f"data/dt={part}/part-{uuid.uuid4().hex[:8]}.json"
p = self.root / rel
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text("\n".join(json.dumps(x, ensure_ascii=False) for x in chunk))
files.append({"path": rel, "partition": part, "rows": len(chunk)})
return files
def _find(self, meta, sid):
for s in meta["snapshots"]:
if s["id"] == sid:
return s
return None
def read(self, snapshot_id=None):
meta = self._read()
snap = self._find(meta, snapshot_id if snapshot_id is not None else meta["current"])
if snap is None:
return []
cols = [c["name"] for c in meta["schema"]]
out = []
for f in snap["manifest"]:
for line in (self.root / f["path"]).read_text().splitlines():
r = json.loads(line)
out.append({c: r.get(c) for c in cols}) # 新列旧文件读成 None
return out
def append(self, rows, retries=5, base_meta=None):
files = self._write_files(rows) # 数据文件只写一次，冲突时只重试提交
for attempt in range(1, retries + 1):
meta = copy.deepcopy(base_meta) if (base_meta and attempt == 1) else self._read()
base = meta["version"]
cur = self._find(meta, meta["current"]) or {"manifest": []}
snap = self._snapshot(meta, "append", cur["manifest"] + files, {"added": len(rows)})
meta["snapshots"].append(snap)
meta["current"] = snap["id"]
try:
self._commit(base, meta)
return snap["id"], attempt
except Conflict:
continue # 数据文件已落盘，只重算 manifest
raise Conflict("重试仍失败")
def overwrite(self, partition_value, rows):
meta = self._read()
base, cur = meta["version"], self._find(meta, meta["current"])
keep = [f for f in cur["manifest"] if f["partition"] != partition_value]
old = sum(f["rows"] for f in cur["manifest"] if f["partition"] == partition_value)
files = self._write_files(rows)
snap = self._snapshot(meta, "overwrite", keep + files,
{"added": len(rows), "deleted": old})
meta["snapshots"].append(snap)
meta["current"] = snap["id"]
self._commit(base, meta)
return snap["id"]
def delete(self, pred):
meta = self._read()
base, rows = meta["version"], self.read()
keep = [r for r in rows if not pred(r)]
files = self._write_files(keep) if keep else []
snap = self._snapshot(meta, "delete", files,
{"added": len(keep), "deleted": len(rows) - len(keep)})
meta["snapshots"].append(snap)
meta["current"] = snap["id"]
self._commit(base, meta)
return snap["id"]
def compact(self, max_rows=100):
meta = self._read()
base, rows = meta["version"], self.read()
before = len(self._find(meta, meta["current"])["manifest"])
files = self._write_files(rows, max_rows=max_rows)
snap = self._snapshot(meta, "compact", files,
{"added": len(rows), "files": f"{before}->{len(files)}"})
meta["snapshots"].append(snap)
meta["current"] = snap["id"]
self._commit(base, meta)
return snap["id"], len(rows), len(files)
def add_column(self, name, typ="string"):
meta = self._read()
base = meta["version"]
if any(c["name"] == name for c in meta["schema"]):
return
meta["schema"].append({"name": name, "type": typ})
cur = self._find(meta, meta["current"])
snap = self._snapshot(meta, "evolve-schema", cur["manifest"], {"added": 0})
meta["snapshots"].append(snap)
meta["current"] = snap["id"]
self._commit(base, meta)
def main():
root = Path(tempfile.mkdtemp(prefix="mini-lake-"))
t = Table(root)
day1 = [{"id": i, "dt": "2026-10-01", "amount": 100 + i} for i in range(1, 61)]
day2 = [{"id": 100 + i, "dt": "2026-10-02", "amount": 200 + i} for i in range(1, 61)]
s1, _ = t.append(day1)
s2, _ = t.append(day2)
pinned = t.read(s1)
print(f"① 初始写入：s{s1}=60 行，s{s2}=120 行；读者固定 s{s1}，读到 {len(pinned)} 行")
fixed = [{"id": 100 + i, "dt": "2026-10-02", "amount": 999} for i in range(1, 51)]
s3 = t.overwrite("2026-10-02", fixed)
print(f"② 覆盖分区 10-02：s{s3} 共 {len(t.read())} 行；固定 s{s1} 的读者仍读到 {len(t.read(s1))} 行")
t.add_column("channel")
day3 = [{"id": 300 + i, "dt": "2026-10-03", "amount": 300 + i, "channel": "app"}
for i in range(1, 41)]
s5, _ = t.append(day3)
rows = t.read()
print(f"③ 加列并追加：s{s5} 共 {len(rows)} 行；老数据 channel={rows[0]['channel']}，"
f"新数据 channel={rows[-1]['channel']}")
s6 = t.delete(lambda r: r["dt"] == "2026-10-01" and r["id"] % 12 == 0)
print(f"④ 删除 5 行：s{s6} 共 {len(t.read())} 行")
a, b = Table(root), Table(root)
stale = b._read()
before = stale["version"]
sa, att_a = a.append([{"id": 900 + i, "dt": "2026-10-04", "amount": 10} for i in range(10)])
sb, att_b = b.append([{"id": 950 + i, "dt": "2026-10-04", "amount": 20} for i in range(10)],
base_meta=stale)
print(f"⑤ 并发提交：base v{before}；A=s{sa}(第{att_a}次成功)，"
f"B=s{sb}(第{att_b}次成功，先冲突后重试)；当前 {len(t.read())} 行")
files_before = len(t._find(t._read(), t._read()["current"])["manifest"])
s7, n, files_after = t.compact()
print(f"⑥ 合并小文件：{files_before} 个 -> {files_after} 个，行数 {n} 不变")
print("\n=== 快照日志 ===")
print(f"{'id':>3} {'parent':>7} {'operation':<14} {'files':>5} {'rows':>5} summary")
for s in t._read()["snapshots"]:
print(f"{s['id']:>3} {str(s['parent']):>7} {s['operation']:<14} "
f"{len(s['manifest']):>5} {sum(f['rows'] for f in s['manifest']):>5} {s['summary']}")
print("\n=== 时间旅行 ===")
for sid in [s1, s2, s3, s5, s6, s7]:
print(f" s{sid}: {len(t.read(sid))} 行")
main()
本机实测（macOS，Apple M1，Python 3.14.8）：
① 初始写入：s1=60 行，s2=120 行；读者固定 s1，读到 60 行
② 覆盖分区 10-02：s3 共 110 行；固定 s1 的读者仍读到 60 行
③ 加列并追加：s5 共 150 行；老数据 channel=None，新数据 channel=app
④ 删除 5 行：s6 共 145 行
⑤ 并发提交：base v6；A=s7(第1次成功)，B=s8(第2次成功，先冲突后重试)；当前 165 行
⑥ 合并小文件：17 个文件 -> 4 个，行数 165 不变
=== 快照日志 ===
id parent operation files rows summary
1 None append 6 60 {'added': 60}
2 1 append 12 120 {'added': 60}
3 2 overwrite 11 110 {'added': 50, 'deleted': 60}
4 3 evolve-schema 11 110 {'added': 0}
5 4 append 15 150 {'added': 40}
6 5 delete 15 145 {'added': 145, 'deleted': 5}
7 6 append 16 155 {'added': 10}
8 7 append 17 165 {'added': 10}
9 8 compact 4 165 {'added': 165, 'files': '17->4'}
=== 时间旅行 ===
s1: 60 行
s2: 120 行
s3: 110 行
s5: 150 行
s6: 145 行
s9: 165 行
读数：
| 现象 | 结果 | 说明 |
|---|---|---|
| 读者固定 s1 时发生覆盖 / 追加 | 读者始终读到 60 行 | 快照隔离：读的是文件列表，不是「当前目录」 |
| 覆盖 10-02 | 120 → 110 行 | 只替换该分区的文件，其他分区文件不动 |
| 加列 | 老数据 None，新数据 app | schema 演进不重写历史文件，读时按列名补空 |
| 并发提交 | B 第 1 次冲突，第 2 次成功 | 版本号 CAS；重试后两条都保留，没有丢更新 |
| 合并小文件 | 17 → 4 个文件，165 行不变 | compaction 只改文件布局，不改逻辑数据 |
| 时间旅行 | s1=60、s3=110、s9=165 | 旧快照仍可读，直到 expire 清理 |
🔑 注意第 ⑤ 步：B 的数据文件在冲突前就已经写好了，冲突只影响「提交」；重试成功后仍引用同一批文件。如果 B 放弃重试或进程崩溃，这批文件才成为孤儿，要靠
orphan file removal清理；清理必须与快照过期共用引用计数，否则会删掉仍被旧快照引用的文件。
小文件：为什么在对象存储上更致命
同样 20 万行，2000 个 100 行的小文件 vs 8 个大文件，只测「打开 + 读取 + 解析」：
#!/usr/bin/env python3
import json, statistics, tempfile, time, shutil
from pathlib import Path
ROWS, PER_FILE = 200_000, 100
def write(path, files, rows_per_file):
path.mkdir(parents=True)
for i in range(files):
with (path / f"part-{i:05d}.json").open("w") as f:
for j in range(rows_per_file):
n = i * rows_per_file + j
f.write(json.dumps({"id": n, "amount": n % 1000}) + "\n")
def scan(path):
total = 0
for p in sorted(path.glob("*.json")):
with p.open() as f:
for line in f:
json.loads(line); total += 1
return total
def bench(path, rounds=3):
ts = []
for _ in range(rounds):
t0 = time.perf_counter(); n = scan(path); ts.append(time.perf_counter() - t0)
return statistics.median(ts), n
root = Path(tempfile.mkdtemp(prefix="smallfiles-"))
try:
small, large = root / "small", root / "large"
write(small, ROWS // PER_FILE, PER_FILE)
write(large, 8, ROWS // 8)
t_small, n1 = bench(small); t_large, n2 = bench(large)
print(f"行数：小文件 {n1}，大文件 {n2}")
print(f"文件数：{len(list(small.glob('*.json')))} vs {len(list(large.glob('*.json')))}")
print(f"全量扫描中位数：{t_small*1000:.0f} ms vs {t_large*1000:.0f} ms，慢 {t_small/t_large:.1f} 倍")
print(f"每多一个文件的固定开销：约 {(t_small-t_large)/(ROWS//PER_FILE-8)*1e6:.0f} µs（本机 SSD）")
for ms in (10, 30):
print(f"外推：对象存储每文件 {ms} ms GET -> 2000 文件仅取数就要 {2000*ms/1000:.0f} s")
finally:
shutil.rmtree(root)
本机实测（macOS，Apple M1，Python 3.14.8；同一台机器多次运行有 ±10% 波动）：
行数：小文件 200000，大文件 200000
文件数：2000 vs 8
全量扫描中位数：249 ms vs 207 ms，慢 1.2 倍
每多一个文件的固定开销：约 21 µs（本机 SSD）
外推：对象存储每文件 10 ms GET -> 2000 文件仅取数就要 20 s
外推：对象存储每文件 30 ms GET -> 2000 文件仅取数就要 60 s
读数：本机只有 1.2 倍，因为本地 SSD 的 open 很便宜（约 21 µs/文件），且页缓存把大文件也缓存了。真正的代价在对象存储：每个文件至少一次 GET，2000 个文件就是 2000 次网络往返，10 ms RTT 下仅取数就要 20 秒，还没算查询规划、manifest 解析和任务启动。所以「小文件治理」不是审美问题，是每文件固定成本 × 文件数的乘法。
生产上的目标值与动作：
| 参数 | 经验值 | 说明 |
|---|---|---|
| 单文件大小 | 128–512 MB | Parquet 压缩后；太小则开销占比高，太大则并行度不足 |
| 每分区文件数 | 控制在几十个以内 | 超过 1000 个就该 compaction |
| 写入批次 | 每 1–5 分钟一个文件 | 流式写入要配小文件合并，不能每批一个文件 |
| 快照保留 | 最近 10–20 个 + 按天保留 | 兼顾时间旅行与元数据大小 |
| compaction 频率 | 每天 / 每小时（按增量） | 全量重写成本高，优先合并小分区 |
⚠️ Compaction 不是免费的：它重写数据文件、产生新快照、消耗 IO 与计算，还会与并发写竞争。生产上要限流、按分区做、避开高峰，并且合并后保留旧快照一段时间，否则「刚合并完发现逻辑错了」就无法回滚。
生产落地：指名 → 可抄 → 可验 → 可退
指名：表格式用 Apache Iceberg（多引擎、隐藏分区；批/流写入的 shuffle 与微批见 61.2 Spark 计算模型）；catalog 用 Iceberg REST Catalog / Nessie / AWS Glue / Unity Catalog（按云与多租户选）；计算用 Spark / Flink / Trino；小文件合并用 Iceberg 的 rewrite_data_files。
可抄（Spark SQL + Iceberg，未验证：本机未安装 Spark；版本 Iceberg 1.6+、Spark 3.5+）：
-- 建表：按天隐藏分区，format-version=2 支持行级删除
CREATE TABLE lake.db.orders (
id BIGINT,
dt DATE,
amount DECIMAL(18,2),
status STRING
) USING iceberg
PARTITIONED BY (days(dt))
TBLPROPERTIES ('format-version' = '2');
-- 合并小文件：目标 128 MB，只重写小文件
CALL lake.system.rewrite_data_files(
table => 'db.orders',
options => map('target-file-size-bytes', '134217728',
'min-input-files', '5')
);
-- 过期快照：保留最近 10 个，7 天前的可删
CALL lake.system.expire_snapshots(
table => 'db.orders',
older_than => TIMESTAMP '2026-10-23 00:00:00',
retain_last => 10
);
-- 时间旅行：查某个快照
SELECT count(*) FROM lake.db.orders VERSION AS OF 1234567890;
-- 元数据表：看快照历史（Iceberg 内置）
SELECT snapshot_id, parent_id, operation, committed_at
FROM lake.db.orders.snapshots ORDER BY committed_at DESC LIMIT 5;
可验：SELECT * FROM lake.db.orders.snapshots 应返回每次 append / overwrite / rewrite 的记录；VERSION AS OF 查到的行数应与当时一致。本节脚本的「时间旅行」输出是真实验证。
可退：CALL lake.system.rollback_to_snapshot(table => 'db.orders', snapshot_id => 1234567890); 秒级回滚到旧快照；前提是旧快照还没被 expire_snapshots 清掉。因此过期策略要留足回滚窗口（例如 7 天），不要「合并完立刻清」。
反模式
| 反模式 | 后果 |
|---|---|
| 把湖仓当 OLTP 做点查 | 每次查询扫大量文件，P99 秒级；该回数据库 |
| 不做 compaction | 小文件越积越多，查询计划与 GET 成本线性涨 |
| 每个微批都 commit 一次 | 快照与 manifest 膨胀，catalog 压力大 |
| 绕过表格式直接写目录 | 元数据与文件不一致，时间旅行读到「幽灵数据」 |
| 没有 catalog，靠文件路径当表名 | 并发提交无法 CAS，丢更新；改表名要改所有任务 |
expire_snapshots 太激进 | 回滚与时间旅行失效，出错只能全量重算 |
| 多引擎并发写同一分区不配重试 | 后提交者失败，或（更糟）覆盖前者 |
| 只给表权限、不收紧对象存储直读 | 用户绕过表格式直接读文件，行/列权限失效（见 60.6 数据安全与权限） |
自检
Q1 「湖仓一体」比数据湖多了哪一层？为什么不能简单理解成「湖 + 数仓」？
▸ 答案：多的是表格式元数据层（Iceberg / Delta / Hudi），它定义「哪些不可变文件构成表的哪个快照」，并在其上实现 ACID、时间旅行、schema 演进。简单拼「湖 + 数仓」是两套存储、两套元数据、两套权限，数据要复制和同步；湖仓是一套开放文件 + 一套表元数据，多个引擎共享同一份真相。
Q2 对象存储没有原子 rename，表格式如何保证提交原子？
▸ 答案：提交点不是「移动目录」，而是原子替换 catalog 中的当前 metadata 指针。新数据文件、manifest、snapshot 都是新写的（不覆盖旧文件），只有最后一步「把当前指针指向新 snapshot」需要原子性，由 catalog 的条件写 / CAS 保证。替换成功读者才能看到新版本；失败则新文件成为孤儿，旧快照不受影响。
Q3 快照隔离下，读者为什么看不到「写了一半」的数据？
▸ 答案：读者读的是一个固定 snapshot 的 manifest 文件列表，而不是 LIST 目录。写入方的新文件在提交前不在任何 snapshot 的 manifest 里，读者根本不会去读；提交后新 snapshot 才引用它们。文件本身不可变，所以也不会出现「读的时候被改写」。
Q4 隐藏分区解决什么问题？举一个 Hive 分区写错的真实后果。
▸ 答案：隐藏分区把「分区列怎么算」收进表定义，查询只写业务谓词，引擎自动推导分区裁剪；同时避免用户把分区列写错、格式写错、基数失控。Hive 里常见后果：用户写 WHERE dt = '2026-10-1'（少补零）导致分区裁剪失效，全表扫描；或按 user_id 分区造出百万小目录，LIST 直接超时。
Q5 小文件为什么在对象存储上比本地磁盘严重得多？compaction 的代价是什么？
▸ 答案：本地磁盘 open 约几十微秒，且有页缓存；对象存储每个文件至少一次 GET，10–30 ms RTT，2000 个文件就是 20–60 秒，且查询规划还要解析 manifest。compaction 的代价是重写数据（IO + CPU）、产生新快照、与并发写竞争，并且合并后旧文件要等快照过期才能删，期间存储翻倍。
Q6 Schema 演进为什么能「改名不丢数据」？
▸ 答案：表格式给每列分配稳定的 field-id，数据文件按 id 而不是按列名存储/读取。改名只改元数据里的 name，id 不变，旧文件仍能映射到同一列；加列时旧文件没有该 id，读时补默认值（通常是 null）。这也是为什么不能靠「改 Parquet 列名」做演进：列名不是稳定标识。
Q7 什么场景不该用湖仓？
▸ 答案：① 毫秒级主键点查 / 高频事务更新——用 OLTP 数据库；② 数据量很小（GB 级）且团队没有运维能力——托管数仓更便宜；③ 强多行事务与唯一约束——湖仓只有表级快照，不提供跨表事务与唯一索引；④ 极低延迟流式（亚秒级）——湖仓的提交间隔通常在秒到分钟级，实时链路要配 OLAP / 流存储。