清理

背景

Cleaning（清理）是 Hudi 采用的一种表服务，用于回收旧版本数据所占用的空间，从而控制存储成本。Apache Hudi 通过 MVCC（多版本并发控制）管理多个带版本的文件，在写入者与读取者之间提供快照隔离。这些文件版本保留了历史数据，并支持时间旅行与回滚操作，但需要注意保留多少历史数据，以便在成本之间取得平衡。清理服务在保留数据长期历史与相应存储成本之间的权衡管理中起着至关重要的作用。

Hudi 默认启用自动 Hudi 清理。每次提交（commit）完成后会立即触发清理，以删除较旧的文件切片（file slice）。建议保持此功能启用，以确保元数据和数据存储的增长是有限的。也可以通过配置 hoodie.clean.max.commits，让清理器每隔若干次提交后执行，而不是每次提交后都执行。

清理保留策略

清理旧文件时，应小心不要删除正在被长时间运行的查询所使用的文件。

对于基于 Spark 的引擎：

配置名称默认值描述

hoodie.clean.policyKEEP_LATEST_COMMITS (Optional)org.apache.hudi.common.model.HoodieCleaningPolicy: 要使用的清理策略。

Config Param: CLEANER_POLICY

基于 Flink 的引擎对应配置为 clean.policy。

Hudi 清理器目前支持以下清理策略，用于保留一定数量的提交或文件版本：

• KEEP_LATEST_COMMITS：这是默认策略。它是一种基于时间的清理策略，确保能够回溯到最近 X 次提交中发生的所有变更。假设某个写入端每 30 分钟向 Hudi 数据集写入一次数据，而运行时间最长的查询可能需要 5 小时才能完成，那么用户应至少保留最近 10 次提交。通过这样的配置，我们可以保证文件的最旧版本在磁盘上至少保留 5 小时，从而确保最长运行的查询在任何时刻都不会失败。使用该策略还可以启用增量清理。要保留的提交数量可通过 hoodie.clean.commits.retained 进行配置，对应的 Flink 相关配置为 clean.retain_commits。
• KEEP_LATEST_FILE_VERSIONS：该策略的效果是无论时间如何，始终保留 N 个文件版本。当已知在任意时刻希望保留文件的最多（MAX）版本数时，该策略非常有用。要实现与前述相同的、防止长时间运行查询失败的效果，用户应根据数据模式进行相应计算。此外，如果用户只想保留文件的 1 个最新版本，该策略同样有用。要保留的文件版本数量可通过 hoodie.clean.fileversions.retained 进行配置，对应的 Flink 相关配置为 clean.retain_file_versions。
• KEEP_LATEST_BY_HOURS：该策略基于小时数进行清理，简单直观，适用于已知在任意时刻需要保留文件时长的场景。提交时间早于所配置保留小时数的提交将被清理。目前可通过参数 hoodie.clean.hours.retained 进行配置，对应的 Flink 相关配置为 clean.retain_hours。

追加表的空清理提交

追加式表从不累积更新，因此清理器的 earliest_commit_to_retain 指针永远不会推进，导致清理器每次运行都需要扫描整个表的历史记录。Hudi 1.2.0 引入了定期的空清理提交，即使没有需要删除的数据，也能推进该指针。

限制每次运行清理的提交数量

从 1.2.0 开始，你可以限制单次清理运行中清理的提交数量，这对于控制清理严重滞后的表上的作业运行时长很有用。

全量清理的分区过滤

当增量清理被禁用时（hoodie.clean.incremental.enabled=false），清理器每次运行都会扫描所有分区。对于非常大的表，这可能导致规划阶段出现 OOM。Hudi 1.2.0 新增了两个配置，用于限制要检查的分区范围。

note

这两个配置都要求设置 hoodie.clean.incremental.enabled=false。如果同时设置了两者，则 hoodie.clean.partition.filter.selected 的优先级高于正则表达式。

清理元数据中的即时时间

Hudi 1.x 为每个操作都标记了请求即时时间（requested instant time）和完成时间（completion time），并按完成时间在时间线上对操作排序——参见 timeline。然而，清理器自身的计划和元数据自始至终记录的都是即时（开始）时间。在阅读这些内容进行调试时请牢记这一点。

回读这些值时，有两个细节很容易让人踩坑：

• lastCompletedCommitTimestamp 混用了两种排序方式。该时间戳取自已完成提交时间线的末尾，而该时间线是按完成时间排序的，但实际记录下来的却是该实例的开始时间。并发写入者的完成顺序可能与其开始顺序不同，因此它并不总是已完成提交中最大的实例时间。
• 在 KEEP_LATEST_FILE_VERSIONS 策略下，earliestCommitToRetain 是一个空字符串。该策略是为每个文件组保留固定数量的文件版本，而不是时间线上的某个范围，因此计划中不携带可供元数据复制使用的 earliestInstantToRetain。

增量清理规划遵循同样的约定：它会选取请求实例时间大于或等于上次清理的 earliestCommitToRetain、且早于本次清理的 earliestCommitToRetain 的提交，然后只扫描这些提交所涉及的分区。

配置项

关于所有可用配置及其默认值的详细信息，请参阅配置文档。与 Flink 相关的配置请参见此处。

触发清理的方式

内联（Inline）

在基于 Spark 的写入中，清理默认会在每次提交之后以内联方式执行，并使用默认策略 KEEP_LATEST_COMMITS。建议保持启用该功能，以确保元数据和数据存储的增长是有界的。要启用它，用户无需设置任何配置。以下是相关的基本配置。

异步（Async）

如果你希望在写入的同时异步运行清理服务，请按如下方式启用 hoodie.clean.async：

hoodie.clean.automatic=true
hoodie.clean.async=true

对于基于 Flink 的写入，这是默认的清理模式。详情请参阅 clean.async.enabled。

写前清理策略

默认情况下，清理器在一次写入提交之后运行。Hudi 1.2.0 引入了 hoodie.prewrite.cleaner.policy，允许你在每次写入开始之前强制执行一次清理（或回滚失败的写入）。这在多写入器部署中非常有用，因为你会希望在每次写入之前表状态都是确定的——相关多写入器配置请参见并发控制。

独立运行

Hoodie Cleaner 也可以作为一个独立进程运行。以下是独立运行清理器的命令：

spark-submit --master local \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.HoodieCleaner `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` --help
        Usage: <main class> [options]
        Options:
        --help, -h

        --hoodie-conf
        Any configuration that can be set in the properties file (using the CLI
        parameter "--props") can also be passed command line using this
        parameter. This can be repeated
        Default: []
        --props
        path to properties file on localfs or dfs, with configurations for
        hoodie client for cleaning
        --spark-master
        spark master to use.
        Default: local[2]
        * --target-base-path
        base path for the hoodie table to be cleaner.

运行清理服务的示例

保留最近 10 次提交

spark-submit --master local \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.HoodieCleaner `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --target-base-path /path/to/hoodie_table \
  --hoodie-conf hoodie.clean.policy=KEEP_LATEST_COMMITS \
  --hoodie-conf hoodie.clean.commits.retained=10 \
  --hoodie-conf hoodie.clean.parallelism=200

保留最新的 3 个文件版本

spark-submit --master local \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.HoodieCleaner `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --hoodie-conf hoodie.clean.policy=KEEP_LATEST_FILE_VERSIONS \
  --hoodie-conf hoodie.clean.fileversions.retained=3 \
  --hoodie-conf hoodie.clean.parallelism=200

清理超过 24 小时的提交

spark-submit --master local \
  --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.2.0,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.0 \
  --class org.apache.hudi.utilities.HoodieCleaner `ls packaging/hudi-utilities-slim-bundle/target/hudi-utilities-slim-bundle-*.jar` \
  --target-base-path /path/to/hoodie_table \
  --hoodie-conf hoodie.clean.policy=KEEP_LATEST_BY_HOURS \
  --hoodie-conf hoodie.clean.hours.retained=24 \
  --hoodie-conf hoodie.clean.parallelism=200

注：并行度取待清理分区数与 hoodie.clean.parallelism 中的较小值。

CLI

你也可以使用 Hudi CLI 来运行 Hoodie Cleaner。

CLI 为清理服务提供以下命令：

• cleans show
• clean showpartitions
• cleans run

保留最近 10 次提交的清理示例

cleans run --sparkMaster local --hoodieConfigs hoodie.clean.policy=KEEP_LATEST_COMMITS hoodie.clean.commits.retained=10 hoodie.clean.parallelism=200

你可以在 org.apache.hudi.cli.commands.CleansCommand 类中找到这些命令的更多细节和相关代码。

博客

• Apache Hudi 中的清理与归档（Cleaner and Archival in Apache Hudi）

视频

• 清理服务：数据湖存储成本节省高达 40% | Hudi Labs
• 使用 Apache Hudi Cleaner 高效管理数据湖：调度数据清理的好处 devlive-community/knowforge#1
• 使用 Apache Hudi Cleaner 高效管理数据湖：调度数据清理的好处 devlive-community/knowforge#2