【数据湖与开放表格式】小文件与 Compaction

第 16 章 留下一个尾巴：不可变快照链每次提交都产生新文件，旧的不删。如果提交很频繁——流式入湖每几秒一次、CDC 每个微批一次、分区切得太细每个分区只落几行——表很快变成「一堆小文件 + 一堆元数据」。这就是 lakehouse 头号运维问题：小文件。

小文件不是「占地方」这么简单。它直接拖慢查询 planning：每个 data file 在 manifest 里都是一条记录，文件越多，planning 要遍历的元数据越多、要打开的 Parquet footer 越多、调度的 task 越碎。本章先量化这个代价（真跑一组数据），再讲对症的治理操作：bin-pack、sort/z-order、rewrite manifests、expire snapshots、remove orphan files，最后讲 Puffin 里的 NDV sketch 怎么帮 planning。

版本锚定：Iceberg 表规范 V2，rewrite_data_files / rewrite_manifests / expire_snapshots / remove_orphan_files 维护过程，Puffin 规范（apache-datasketches-theta-v1 blob），对照 Delta OPTIMIZE。实验环境：Arch Linux on WSL2（kernel 6.6.87.2），i9-12900K（24 逻辑核），31 GiB RAM，Python 3.14.5，PyIceberg 0.11.1，PyArrow 24.0.0，本地文件系统 + SQLite SQL catalog。

一、小文件从哪来

1.1 三个根因

三者本质相同：写入的并发/频率与「理想文件大小」不匹配。理想的 data file 通常在几十到几百 MB（Iceberg 默认 write.target-file-size-bytes 为 512 MB 量级），但流式微批一次可能只有几十 KB。

1.2 小文件的连锁代价

flowchart TD
  SF[大量小文件] --> P[planning: 遍历更多 manifest 条目]
  SF --> FT[读取: 打开更多 Parquet footer]
  SF --> TASK[执行: task 碎片化, 调度开销大]
  SF --> META[元数据: manifest/metadata.json 膨胀]
  P --> SLOW[查询变慢]
  FT --> SLOW
  TASK --> SLOW
  META --> COST[catalog/存储压力上升]

下面把「planning 变慢」和「元数据膨胀」两条用真实数字钉死。

二、量化小文件代价（真实实验）

2.1 构造小文件

建一张表，做 200 次 append，每次 50 行——故意模拟「频繁提交」：

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

catalog = SqlCatalog("demo",
    uri="sqlite:////tmp/ice_wh17/catalog.db", warehouse="file:///tmp/ice_wh17")
catalog.create_namespace("db")
schema = pa.schema([("id", pa.int64(), False), ("v", pa.string())])
tbl = catalog.create_table("db.events", schema=schema)

for i in range(200):                       # 200 次提交
    base = i * 50
    d = pa.table({"id": list(range(base, base+50)),
                  "v": [f"r{base+j}" for j in range(50)]}, schema=schema)
    tbl.append(d)                          # 每次一个小文件

2.2 planning 耗时：compaction 前

测「列出当前快照所有待扫描文件」的耗时（scan().plan_files()，7 轮取中位数）：

import time, statistics
def plan_time_ms(runs=7):
    ts = []
    for _ in range(runs):
        t0 = time.perf_counter()
        files = list(tbl.scan().plan_files())
        ts.append((time.perf_counter()-t0)*1000)
    return statistics.median(ts), len(files)

真实输出：

=== compaction 前 ===
  当前快照数据文件数: 200
  plan_files 中位耗时: 184.15 ms (7 轮)
  总行数: 10000

1 万行数据，因为分成 200 个文件，光是 planning（还没开始读数据）就要 184 ms。

2.3 compaction：合并成大文件

PyIceberg 0.11 没有内置的 rewrite_data_files（那是 Spark 的存储过程）。这里用 overwrite 把全表读出再以大文件写回，复现 bin-pack 合并的效果——读全量、写成满足目标大小的少量文件：

all_rows = tbl.scan().to_arrow()   # 读全表
tbl.overwrite(all_rows)            # 重写成大文件（模拟 bin-pack）

真实输出：

=== compaction 后（手动 overwrite 重写）===
  当前快照数据文件数: 1
  plan_files 中位耗时: 1.33 ms (7 轮)
  总行数: 10000

2.4 对比

planning 从 184 ms 降到 1.3 ms，约 138 倍——数据一行没变，纯靠把文件数从 200 压到 1。这就是 compaction 的直接收益：planning 成本大致随当前快照的文件数线性增长，合并文件 = 缩短规划路径。

2.5 扫描 N 验证线性关系

两点（200 和 1）只能给方向，不能证明线性。再做一组实验：分别构造文件数 \(N \in \{10, 25, 50, 100, 150, 200\}\) 的表，每个测 plan_files 中位耗时（7 轮）。真实输出：

N=  10  plan_median=    9.04 ms
N=  25  plan_median=   22.18 ms
N=  50  plan_median=   43.66 ms
N= 100  plan_median=   87.06 ms
N= 150  plan_median=  129.48 ms
N= 200  plan_median=  173.65 ms

这组点几乎落在一条直线上，印证：

\[ T_{\text{plan}} \approx c_0 + c_1 \cdot N_{\text{files}} \]

用首尾两点估斜率 \(c_1 \approx \dfrac{173.65 - 9.04}{200 - 10} \approx 0.87\ \text{ms/文件}\)，截距 \(c_0\) 接近 0。也就是说，本环境下每多一个当前快照里的数据文件，planning 多花约 0.87 ms。这正是「文件数」而非「数据量」主导 planning 成本的直接证据——6 组实验的总行数从 500 到 1 万不等，但耗时几乎只看 \(N\)。

绝对斜率依引擎与存储后端而变：本地文件系统上是元数据反序列化与对象打开开销；换到 S3 这类远端对象存储，每文件还要叠加一次 list/GET 的网络往返，\(c_1\) 通常显著更大，小文件的惩罚更狠。跨测量会话还有几个 ms 的抖动（如 \(N{=}200\) 在两次会话分别测得 173.65 和 184.15 ms），属正常方差，不改变线性结论。

复现：本图由同目录 plot_planning_cost.py 生成（PyIceberg 0.11.1、PyArrow 24.0.0、i9-12900K、本地 FS），脚本保留可重绘。

2.6 别忘了：物理文件并没立刻消失

compaction 后当前快照只剩 1 个文件，但磁盘上的物理文件并没减少。检查实验目录：

data/ 下 *.parquet 物理文件数: 201   （200 个旧小文件 + 1 个合并后大文件）
metadata/ 下:  *.metadata.json = 202,  *.avro(manifest/list) = 404
快照总数: 202

200 次 append + 1 次 overwrite = 202 个 snapshot，于是 202 个 metadata.json、404 个 manifest 相关 avro 文件，旧的 200 个小 parquet 仍躺在 data/ 下——它们只被已不是「当前」的历史快照引用。

这暴露了 compaction 的真相：它只让当前快照变干净，旧文件要靠 expire snapshots + remove orphan files 才真正回收。 这正是第四节的内容。

三、Compaction 策略：bin-pack 与 sort/z-order

Iceberg 的 rewrite_data_files（Spark 存储过程）支持几种策略，核心区别是重写时是否顺带重排数据。

3.1 bin-pack（默认）

最简单：把小文件「装箱」成接近 target-file-size 的大文件，不改变行的顺序。开销最小，纯解决文件数量问题。对应第二节实验的效果。关键参数（Spark）：

3.2 sort 与 z-order

bin-pack 只解决「文件多」，不解决「数据没按查询模式聚集」。如果查询常按某列过滤，但数据乱序，文件级 min/max 统计区间互相重叠，文件裁剪（第 18 章）就失效——每个文件的范围都覆盖全域，谁也裁不掉。

• sort rewrite：重写时按指定列排序，让同一列值聚集到相邻文件，使文件级 min/max 区间变窄、不重叠，单列过滤裁剪更有效。
• z-order rewrite：当查询会按多个列过滤时，单一排序键只能让一列聚集。z-order 用 Z-order 曲线把多维交织成一维排序，让多列同时获得一定的局部性。代价是单列的聚集度不如纯 sort，是多列之间的折中。

flowchart LR
  RAW[乱序小文件<br/>min/max 区间重叠] -->|bin-pack| BP[大文件<br/>仍乱序]
  RAW -->|sort by col| SORT[大文件<br/>单列聚集]
  RAW -->|z-order col_a,col_b| ZO[大文件<br/>多列联合聚集]

3.3 文件组（file group）与并行

rewrite_data_files 不是「把全表读进内存再写一个文件」。它把待重写的文件按分区切成文件组（file group），每个组独立重写、可并行，组内把若干小文件 bin-pack 成接近目标大小的输出。这带来三个实际后果：

• 可增量：只重写不达标的文件组，已经是大文件的不动（除非 rewrite-all），避免重写整张 PB 级表。
• 可控内存/并发：max-file-group-size-bytes 限制单组数据量，间接控制单个重写任务的内存峰值；组之间并行度由引擎调度。
• 提交粒度：重写完成后是一次原子提交（替换被重写文件、保留未触及文件），与并发写入按 第 11 章 的乐观并发竞争同一指针——大重写期间若有高频写入，重写提交可能因冲突重试，所以重写常安排在写入低谷。

3.4 含 delete 文件的 MoR compaction

Merge-on-Read 表（第 10 章）的小文件问题更复杂：除了 data file，还有 position/equality delete 文件。读取时要把 data file 与适用的 delete 文件合并，delete 文件越多，读放大越严重。

compaction 在 MoR 表上多了一层语义：重写时把 delete 物化进新的 data file——即真正删掉被标记的行，产出「干净」的 data file，并让相关 delete 文件失效。于是 compaction 同时解决两件事：小文件合并 + 清理累积的 delete。这也是 MoR 表必须定期 compaction 的原因：不重写，delete 文件无限累积，读路径越来越慢。

3.5 触发方式：手动、定时、内联

compaction 在什么时候跑也有取舍：

write.distribution-mode（hash/range/none）是写时少产小文件的手段：让同分区数据汇聚到同一 writer，减少每次提交产生的文件数。它不是 compaction，但能从源头压低小文件产生速度，和写后 compaction 互补。

3.6 与 Delta / liquid clustering 对照

Delta Lake 的对应操作是 OPTIMIZE：

Delta 的 liquid clustering 思路是把「聚集键」做成表的持续属性，写入与维护时增量保持聚集，避免每次全量 z-order 重排。两家都在往「聚集是表的持续性质，而非一次性动作」演进。

四、元数据治理：manifest、expire、orphan

数据文件之外，元数据也要治理——第二节实测的 404 个 avro、202 个 metadata.json 就是证据。

4.1 rewrite manifests

频繁提交会产生大量小 manifest（每次提交至少新增 manifest）。rewrite_manifests 把多个小 manifest 合并、并按分区重新聚簇 manifest 条目，让 planning 时「按分区找 manifest」更快。它动的是元数据层，不重写 data file，所以比 rewrite_data_files 便宜得多。

直觉：rewrite_data_files 治「data file 太多太碎」，rewrite_manifests 治「manifest 太多太碎」。两者正交，都要做。

4.2 expire snapshots

第 16 章 讲过：expire_snapshots 删除旧 snapshot，并物理删除「只被过期 snapshot 引用」的 data/manifest 文件。第二节那 200 个旧小 parquet，正是要靠它回收。

它是 compaction 闭环的关键一环：compaction 把数据合进新文件后，旧小文件变成「只被历史快照引用」，expire 把那些历史快照过期掉，文件才真正可删。只 compaction 不 expire，磁盘会越用越多（当前快照干净，物理占用反而上升，因为又写了一份合并后的数据）。

4.3 remove orphan files

孤儿文件（orphan files）= 物理躺在表目录里、但任何 snapshot 都不引用的文件。来源：失败的写入任务残留、提交前崩溃的临时文件、被 bug 漏删的文件。remove_orphan_files 扫描表目录与元数据引用做差集，删掉没人引用的。

危险操作警告：remove_orphan_files 默认有「文件年龄阈值」（如只删 3 天前的），是为了避免误删正在写入但还没提交的文件——那种文件此刻确实「没被任何 snapshot 引用」，但马上就要被引用。把阈值调太小可能删掉进行中的写入。生产里要给足缓冲。

4.4 维护组合拳的顺序

flowchart TD
  A[rewrite_data_files<br/>合并小数据文件] --> B[rewrite_manifests<br/>合并小 manifest]
  B --> C[expire_snapshots<br/>过期旧快照, 删其独占文件]
  C --> D[remove_orphan_files<br/>清理无人引用的残留]

顺序有讲究：先合并（产生新当前快照），让旧文件沦为历史快照独占；再 expire 把那些历史快照过期、删除其独占文件；最后 remove orphan 兜底清理失败残留。expire 放在 compaction 之后，回收才有意义。

五、统计信息与 Puffin 中的 NDV sketch

5.1 planning 不只看文件数，还看统计

第二节量化的是「文件数 → planning 耗时」。但 planning 的质量还取决于统计信息：列的 min/max（裁剪用）、null count、以及NDV（number of distinct values，去重基数）。NDV 主要服务于基于代价的优化（CBO）：估算过滤后行数、决定 join 顺序与 join 算法。

精确 NDV 要全表去重计数，代价高且难增量维护。于是用概率 sketch 做近似。

5.2 Puffin 与 Theta sketch

Iceberg 用 Puffin 文件格式承载这类统计/索引 blob（同一格式也用于 V3 deletion vector，见 第 10 章）。其中 NDV 用 Apache DataSketches 的 Theta sketch，blob 类型标识为 apache-datasketches-theta-v1。

Theta sketch 的核心思想：对每个值哈希到 \([0,1)\)，只保留最小的若干个哈希值，用「最小哈希阈值 \(\theta\)」反推总基数：

\[ \hat{n} \approx \frac{k}{\theta} \]

其中 \(k\) 是保留的样本数、\(\theta\) 是当前阈值。它用固定大小的内存给出有误差界的基数估计，且可合并——多个文件/分区的 sketch 能 union 出全表估计，天然适配 Iceberg「分片元数据 + 汇总」的结构。

5.3 对 planning 的影响

• 表的 statistics 字段在 metadata 里指向 Puffin 文件，记录各列的 NDV sketch。
• 引擎做 CBO 时读这些 sketch 估算基数，得到更好的 join 顺序、更准的过滤选择率，避免「把大表当小表广播」之类的灾难性计划。
• sketch 是近似的：有误差界，不能当精确计数用；过期或不新鲜的统计会误导优化器，所以统计维护要跟数据更新节奏对齐（写入后刷新）。

5.4 Puffin 文件结构

Puffin 是个独立的二进制文件格式，不是塞在 Parquet 里的附属。它的布局是「魔数 + 一串 blob + footer」：

Magic(4 字节) | blob_1 | blob_2 | … | Footer(blob 元数据 JSON + magic)

• 每个 blob 有类型（如 apache-datasketches-theta-v1）、关联的字段 ID 列表、压缩 codec，以及它对应的 snapshot/sequence-number。
• Footer 是 JSON，列出所有 blob 的偏移、长度、类型与属性。读取时先读 footer 定位需要的 blob，按需取，不必读整个文件。

表的 metadata.json 里有一个 statistics 数组，每项指向一个 Puffin 文件并记录它覆盖哪个 snapshot、含哪些字段的统计。引擎 planning 时按当前 snapshot 找到对应的 Puffin 统计文件，取出需要列的 NDV sketch。这套结构和 Iceberg「元数据按 snapshot 版本化」的整体设计一致：统计也绑定到某个 snapshot，数据变了要重算。

边界：本节讲「湖侧如何存与用 NDV」；具体引擎是否读取、如何用进 CBO，因引擎而异（第 18 章）。Puffin/Theta 的存在不等于所有引擎都已接入。

六、流式写入下的拉扯与调度

6.1 延迟与文件大小的根本矛盾

流式入湖（第 19 章）把小文件问题推到极致：为了低延迟，sink 希望尽快提交（每个 checkpoint 就 commit），但「尽快提交」直接等于「小文件多」。这是一对结构性矛盾：

\[ \text{提交频率} \uparrow \;\Rightarrow\; \text{端到端延迟} \downarrow,\quad \text{文件数} \uparrow,\quad \text{planning 成本} \uparrow \]

没有一劳永逸的设置，只有按 SLA 取舍：要秒级新鲜度，就接受更碎的文件，并用更勤的后台 compaction 去填坑；能容忍分钟级新鲜度，就攒大微批，源头少产小文件。

6.2 两道防线：写时与写后

写时防线便宜但受延迟 SLA 约束；写后防线兜底但要额外计算资源，且与写入争抢提交指针。两者通常都要上。

6.3 调度与监控

compaction 不是「跑一次」，是持续运维。可监控的信号：

• 当前快照文件数 / 平均文件大小：直接对应第二节的 planning 成本，是首要指标。
• delete 文件数（MoR 表）：累积过多说明 compaction 跟不上 delete 产生速度。
• snapshot 数量与 metadata.json / manifest 数：膨胀说明 expire / rewrite manifests 没跟上（实测一节里 200 次提交就攒了 202 个 metadata.json、404 个 avro）。
• 孤儿文件占用：物理占用与当前快照引用之差，反映需要 remove orphan 的程度。

flowchart LR
  M1[文件数 / 平均大小] --> D{超阈值?}
  M2[delete 文件数] --> D
  M3[snapshot / 元数据数] --> D
  D -->|是| ACT[触发 rewrite + expire]
  D -->|否| WAIT[继续观察]
  ACT --> M1

调度上常见做法：compaction 与 expire 安排在写入低谷（减少提交冲突），保留窗口由「最久要时间旅行多远」决定，remove orphan 低频且留足年龄缓冲。这套指标体系可对接 可观测性系列 的指标方法论。

七、compaction 的代价与写放大

compaction 不是免费的。它本质是「读一批旧文件、写一批新文件、再原子提交」，代价体现在三处。

7.1 写放大

合并 \(N\) 个小文件成大文件，要把这 \(N\) 个文件的数据全部读出再全部写回。若一轮 compaction 处理的数据量是 \(S\)，则它产生约 \(S\) 的读 + \(S\) 的写 IO，外加新文件占用的存储（旧文件在 expire 前还在）。流式表如果每来一点新数据就全量重写，写放大会失控。所以实务上 compaction 按文件组增量做、只碰未达标文件，把写放大限制在「新产生的小文件」这一小部分，而不是全表。

\[ \text{写放大} \approx \frac{\text{compaction 重写的字节}}{\text{新写入的字节}} \]

目标是让这个比值接近一个小常数，而不是随表规模增长。

7.2 与并发写入争提交点

compaction 结束要提交一个「替换被重写文件」的新 snapshot，这和正常写入竞争同一个 catalog 指针（第 11 章）。大重写耗时长，期间若有高频写入抢先提交，compaction 的提交可能因 assert-ref-snapshot-id 失败而重试甚至放弃。缓解：把大 compaction 切成小文件组、安排在写入低谷、用部分重写而非全表。

7.3 保留窗口与元数据增长的算账

第二节实测：200 次提交 = 202 个 metadata.json + 404 个 manifest avro。若一张流式表每分钟提交一次，一天就是 1440 个 snapshot。不做治理，元数据文件数大致随提交次数线性增长：

\[ N_{\text{snapshot}} \approx \text{提交频率} \times \text{保留时长} \]

保留策略就是在「可时间旅行多远」和「元数据/存储开销」之间定刻度。一个常见的双约束写法（语义层面）：

保留：最近 7 天的所有 snapshot，且至少保留最近 100 个；
      被任意 tag 引用的 snapshot 不过期（审计快照永久保留）。

配套：rewrite_manifests 把碎 manifest 合并，expire_snapshots 按上面窗口删旧快照，remove_orphan_files 兜底。三者频率不同——manifest 重写可较勤，expire 按窗口（常每日），orphan 清理低频。

八、复现实验

完整脚本（已在开头声明环境真实运行，输出见第二节）：

import os, shutil, time, statistics, glob
import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog

wh = "/tmp/ice_wh17"; shutil.rmtree(wh, ignore_errors=True); os.makedirs(wh)
catalog = SqlCatalog("demo", uri=f"sqlite:///{wh}/catalog.db", warehouse=f"file://{wh}")
catalog.create_namespace("db")
schema = pa.schema([("id", pa.int64(), False), ("v", pa.string())])
tbl = catalog.create_table("db.events", schema=schema)

for i in range(200):
    base = i*50
    tbl.append(pa.table({"id": list(range(base, base+50)),
                         "v": [f"r{base+j}" for j in range(50)]}, schema=schema))

def plan(runs=7):
    ts=[]
    for _ in range(runs):
        t0=time.perf_counter(); files=list(tbl.scan().plan_files()); ts.append((time.perf_counter()-t0)*1000)
    return statistics.median(ts), len(files)

m,n = plan(); print("前: 文件", n, "plan中位", round(m,2), "ms")
tbl.overwrite(tbl.scan().to_arrow())          # 模拟 bin-pack 合并
m,n = plan(); print("后: 文件", n, "plan中位", round(m,2), "ms")
print("物理 parquet:", len(glob.glob(f"{wh}/db/events/data/*.parquet")))
print("snapshot 数:", len(list(tbl.snapshots())))

无依赖时的安装步骤：

python3 -m venv /tmp/lakeenv
/tmp/lakeenv/bin/python -m pip install pyarrow "pyiceberg[sql-sqlite]"
/tmp/lakeenv/bin/python exp17.py

说明：PyIceberg 0.11 用 overwrite 复现 bin-pack 合并的效果（文件数与 planning 变化），不等于 Spark rewrite_data_files 的全部能力（sort/z-order、增量重写、文件组并行）；后者结论锚定 Iceberg 维护文档。未运行环境下不要把上述输出当自测数据。

九、小结

• 小文件三大根因——频繁提交、流式写入、过细分区——本质都是写入频率/并发与理想文件大小不匹配。
• 小文件直接拖慢 planning。实测：1 万行分成 200 个文件时 plan_files 中位 184 ms，合并成 1 个文件后降到 1.33 ms，约 138 倍，数据完全等价。planning 成本大致随当前快照文件数线性增长。
• compaction 只让当前快照变干净，旧物理文件不会自动消失（实测：合并后磁盘仍有 201 个 parquet、202 个 metadata.json、404 个 manifest avro）。必须配 expire_snapshots + remove_orphan_files 才真正回收。
• 治理操作分工：rewrite_data_files（bin-pack 装箱 / sort / z-order 聚集）治数据文件，rewrite_manifests 治元数据碎片，expire_snapshots 删旧快照及独占文件，remove_orphan_files 清残留。顺序是先合并、再过期、最后兜底，且删数据类操作要留年龄缓冲。
• Puffin 用 Theta sketch 存近似 NDV，服务 CBO 的基数估计；sketch 可合并、有误差界，不是精确计数，统计要随数据更新刷新；是否用进计划因引擎而异。

至此，治理三章（catalog、演进、compaction）完成。下一章转向查询引擎：Trino / Spark / DuckDB / DataFusion 如何利用这些元数据与统计，把 partition pruning、file pruning、row-group/page pruning 串成一条下推链路。

返回 系列目录 ｜ 上一篇：时间旅行、Schema 与分区演进 ｜ 下一篇：查询引擎如何读湖

参考资料

1. Apache Iceberg 文档, Maintenance — rewrite_data_files（bin-pack / sort / z-order）、rewrite_manifests、expire_snapshots、remove_orphan_files — A 级。
2. Apache Iceberg, Puffin 规范 — blob 类型 apache-datasketches-theta-v1、statistics 字段 — A 级。
3. Apache DataSketches, Theta sketch 文档（近似去重基数、可合并性）— A 级。
4. Delta Lake 文档, OPTIMIZE、Z-Order、liquid clustering — A 级（对照）。
5. 本机实验：PyIceberg 0.11.1 + PyArrow 24.0.0 + SQLite SQL catalog，小文件构造与合并前后 planning 对比（环境见开头与第六节）— A 级（实测）。