如何将Hudi中的小Parquet文件合并为大文件以优化查询速度？

问题背景 #

使用Spark+Hudi通过bulk_insert模式将数据写入S3后，Hudi表生成大量约4MB的小Parquet文件；尝试配置inline clustering合并文件以生成100M左右的大文件，但未达到预期，仍存在大量小文件，需优化查询速度。

用户执行的clustering配置及代码如下：

hudi_options = {
    'hoodie.table.name': 'my_table',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.metadata.record.index.enable': True,
    'hoodie.datasource.write.table.name': 'my_table',
    'hoodie.datasource.write.precombine.field': 'ts',
    'hoodie.keep.max.commits': 3,
    'hoodie.keep.min.commits': 2,
    'hoodie.cleaner.commits.retained': 1,
    'hoodie.clustering.inline': 'true',
    'hoodie.clustering.inline.max.commits': 2,
    "hoodie.clustering.plan.strategy.target.file.max.bytes": "1073741824",
    "hoodie.clustering.plan.strategy.small.file.limit": "629145600"
}
df = spark.read.format('hudi').load('s3://path/to/hudi')
df.write.format('hudi').options(**hudi_options).mode('append').save('s3://path/to/hudi')

问题分析 #

当前配置未生效的核心原因：

• small.file.limit阈值过高：设置为600MB，虽4MB小文件理论上符合合并条件，但结合其他参数可能导致触发逻辑不匹配。
• inline clustering触发条件未满足：hoodie.clustering.inline.max.commits=2要求未clustering的commit数达到2才触发，单次append操作无法满足该条件。
• bulk_insert写入特性：该模式直接写入原始文件，未经过Hudi文件大小控制逻辑，后续clustering需匹配正确的分区与策略。

正确解决方法 #

方法1：调整Clustering参数并启用异步Clustering #

修改配置降低小文件阈值，改用异步clustering更适合批量合并场景：

hudi_options = {
    'hoodie.table.name': 'my_table',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.metadata.record.index.enable': True,
    'hoodie.datasource.write.table.name': 'my_table',
    'hoodie.datasource.write.precombine.field': 'ts',
    'hoodie.keep.max.commits': 3,
    'hoodie.keep.min.commits': 2,
    'hoodie.cleaner.commits.retained': 1,
    # 启用异步clustering，后台执行合并任务
    'hoodie.clustering.inline': 'false',
    'hoodie.clustering.async.enabled': 'true',
    # 小文件阈值设为4MB，覆盖目标小文件
    "hoodie.clustering.plan.strategy.small.file.limit": "4194304",
    # 目标文件大小设为100MB（104857600字节）
    "hoodie.clustering.plan.strategy.target.file.max.bytes": "104857600",
    # 限制每次clustering处理的分区数，避免资源过载
    "hoodie.clustering.plan.strategy.max.num.groups": "10",
    # 强制对所有符合条件的小文件执行合并
    "hoodie.clustering.plan.strategy.skip.partition.validation": "true"
}
# 触发clustering任务
spark.read.format('hudi').load('s3://path/to/hudi') \
    .write.format('hudi').options(**hudi_options) \
    .mode('append').save('s3://path/to/hudi')

方法2：使用Hudi Clustering CLI工具触发合并 #

通过命令行直接触发clustering，可控性更强：

spark-submit --class org.apache.hudi.clustering.HoodieClusteringJob \
  --master yarn \
  --conf spark.driver.memory=4g \
  --conf spark.executor.memory=8g \
  --conf spark.executor.cores=4 \
  /path/to/hudi-utilities-bundle.jar \
  --props /path/to/hudi-clustering.properties \
  --mode run

hudi-clustering.properties配置示例：

hoodie.table.name=my_table
hoodie.base.path=s3://path/to/hudi
hoodie.clustering.plan.strategy.small.file.limit=4194304
hoodie.clustering.plan.strategy.target.file.max.bytes=104857600
hoodie.clustering.plan.strategy.max.num.groups=10

方法3：从源头控制bulk_insert的文件大小 #

在bulk_insert阶段直接设置文件大小，减少后续合并成本：

bulk_insert_options = {
    'hoodie.table.name': 'my_table',
    'hoodie.datasource.write.operation': 'bulk_insert',
    'hoodie.datasource.write.recordkey.field': 'id',
    'hoodie.datasource.write.precombine.field': 'ts',
    # 设置shuffle并行度，配合文件大小参数控制输出
    'hoodie.datasource.write.bulk_insert.shuffle.parallelism': '20',
    # 目标文件大小设为100MB
    'hoodie.datasource.write.file.max.bytes': '104857600'
}
df.write.format('hudi').options(**bulk_insert_options) \
    .mode('overwrite').save('s3://path/to/hudi')

注意事项 #

• 执行clustering后，需等待cleaner任务清理旧小文件，可通过hoodie.cleaner.policy=FULL强制触发清理。
• 异步clustering在后台执行，可通过Hudi元数据表查看进度：spark.read.format('hudi').load('s3://path/to/hudi/.hoodie/metadata')。
• 针对S3存储，建议开启Hudi元数据表（hoodie.metadata.enable=true），提升文件发现与clustering效率。

内容的提问来源于stack exchange，提问作者Rinze