当前位置：首页 > news >正文

Iceberg 小文件合并与治理：从写放大到读优化的全链路

news 2026/8/2 1:39:27

Iceberg 小文件合并与治理：从写放大到读优化的全链路

一、小文件是怎么"长"出来的

在 Lakehouse 里，小文件是性能的头号杀手。查询引擎打开一个分区，要先列出成百上千个文件。每个文件都有独立的元数据读取与调度开销。文件越小、数量越多，查询的规划阶段就越慢，I/O 利用率也越低。

小文件的成因，几乎都来自写入侧。Flink 流式入湖时，为保障实时性，往往按固定间隔或条数触发提交。每一次 checkpoint，都可能落出一批几 KB 到几 MB 的碎片文件。若分区粒度过细，比如按小时甚至分钟分区，碎片会被进一步放大。

另一类来源是 CDC 更新。Iceberg 的 MERGE INTO 或行级更新，会为被改动的行生成新的数据文件，旧文件进入"待删除"状态。频繁更新之下，失效文件快速堆积，既占用存储，又拖慢快照扫描。

写入并发也会放大小文件问题。上游并行度过高、单任务产出量却很小，会产生大量并行小文件。这类问题靠调大批大小（write.target-file-size-bytes）通常能缓解，但存量已经形成的碎片，必须靠合并来收口。

治理的目标，不是消灭小文件本身，而是把"写时碎"重新组织成"读时整"。这是一条从写放大到读优化的全链路。

二、Compaction 与快照治理的全链路

Iceberg 的治理，本质是对"文件"和"快照"两套状态的维护。文件层面，靠 Compaction（重写数据文件）把碎片合并成大文件；快照层面，靠过期清理回收无效文件与元数据。两者必须配合，否则只合并不清理，存储永远不会真正下降。

下面用一张流程图，呈现一次完整的治理调度链路。它从调度器触发，到重写、再到清理与校验，形成可观测的闭环。

flowchart TD A[调度器每日触发] --> B[扫描表清单] B --> C{是否存在小文件?} C -->|否| Z[跳过,记录基线] C -->|是| D[提交RewriteDataFiles任务] D --> E[按分区并行重写大文件] E --> F[生成新快照NewSnapshot] F --> G[保留期窗口内旧文件仍可见] G --> H[执行ExpireSnapshots] H --> I[标记孤儿文件待删] I --> J[OrphanFilesCleanup回收] J --> K[更新元数据大小指标] K --> L[推送治理报告] L --> M{存储下降达标?} M -->|否| B M -->|是| Z style D fill:#4A90D9,color:#fff style E fill:#4A90D9,color:#fff style H fill:#E0573E,color:#fff style J fill:#E0573E,color:#fff style L fill:#F2B705,color:#000 style Z fill:#50C878,color:#fff

快照过期策略需要留"后悔药"。生产上不能一有过期就立即物理删除。应保留一个安全窗口，比如七天，应对下游迟到的增量读取或回溯重放。孤儿文件清理更要谨慎，必须确认没有任何正在运行的作业引用，再真正从存储层删除。

Compaction 的收益，不只是文件变少。合并后，列存的统计信息（min/max、null 计数）更准确。下游查询的谓词下推更高效，跳读比例显著提升，这正是"读优化"的落点。

三、生产级合并与清理实现

下面给出基于 PyIceberg 的治理编排实现。代码覆盖超时、重试、空表跳过、并发分区控制与异常兜底。实际部署时，应把它挂到调度系统的定时任务上，并对每张表设置独立的合并阈值。

import logging from datetime import datetime, timedelta from pyiceberg.catalog import load_catalog from pyiceberg.exceptions import NoSuchTableError logger = logging.getLogger("iceberg_compaction") # 安全窗口：过期快照保留 7 天，避免误删正在被引用的数据 RETENTION_DAYS = 7 # 小文件判定阈值：小于该尺寸的文件计入碎片 SMALL_FILE_BYTES = 32 * 1024 * 1024 # 单次合并目标大文件尺寸 TARGET_FILE_BYTES = 512 * 1024 * 1024 def compact_table(catalog, table_id: str, max_retry: int = 3) -> dict: """对单张表执行重写数据文件与快照过期，返回治理摘要。""" for attempt in range(max_retry + 1): try: table = catalog.load_table(table_id) # 先统计当前文件分布，决定是否值得合并 files = list(table.files()) if not files: return {"table": table_id, "skipped": True, "reason": "空表"} small = [f for f in files if f.file_size_in_bytes < SMALL_FILE_BYTES] ratio = len(small) / max(len(files), 1) if ratio < 0.3: return {"table": table_id, "skipped": True, "small_ratio": round(ratio, 2)} # 重写数据文件：按分区并行，目标大文件尺寸受控 table.rewrite_data_files( strategy="sort", target_file_size_bytes=TARGET_FILE_BYTES, use_caching=True, ) # 快照过期：保留窗口内不物理删除 older_than = datetime.now() - timedelta(days=RETENTION_DAYS) table.expire_snapshots(older_than=older_than, retain_last=3) # 孤儿文件清理，默认也按窗口兜底 table.delete_orphan_files(older_than=older_than) after = list(table.files()) return { "table": table_id, "before_files": len(files), "after_files": len(after), "before_bytes": sum(f.file_size_in_bytes for f in files), "after_bytes": sum(f.file_size_in_bytes for f in after), } except NoSuchTableError: return {"table": table_id, "error": "表不存在，跳过"} except Exception as exc: # 兜底：单表失败不影响批次 logger.warning("合并 %s 失败(第%d次): %s", table_id, attempt + 1, exc) if attempt == max_retry: return {"table": table_id, "error": str(exc)} return {"table": table_id, "error": "未知错误"} def run_governance(table_ids: list, catalog_name: str = "default") -> list: catalog = load_catalog(catalog_name) reports = [] for tid in table_ids: # 串行处理单表，但表间可并发；此处用串行降低对元数据的冲击 reports.append(compact_table(catalog, tid)) return reports if __name__ == "__main__": tables = ["lake.ods_user_event", "lake.dwd_order_detail"] summary = run_governance(tables) for row in summary: print(row)

写入侧也要同步调优。把write.target-file-size-bytes调大，并适当增大 Flink 的 checkpoint 间隔，能从源头减少碎片产生。治理是兜底，写入调优才是治本。