Apache Hudi写入时自动调整文件大小功能在Flink SQL中失效

根据Hudi文档预期，Flink每分钟提交一次时，应累积记录写入现有Parquet文件，直到达到5MB的最大文件阈值，但实际每次提交都会生成约400KB的独立文件，且从未合并。

问题中的Flink SQL代码 #

SET 'parallelism.default' = '1';
SET 'execution.checkpointing.interval' = '1m';

CREATE TABLE datagen
(
    id   INT NOT NULL PRIMARY KEY NOT ENFORCED,
    data STRING
) WITH (
      'connector' = 'datagen',
      'rows-per-second' = '5'
);

CREATE TABLE hudi_tbl
(
    id   INT NOT NULL PRIMARY KEY NOT ENFORCED,
    data STRING
) WITH (
      'connector' = 'hudi',
      'path' = 'file:///opt/hudi',
      'table.type' = 'COPY_ON_WRITE',
      'write.parquet.block.size' = '1',
      'write.operation' = 'insert',
      'write.parquet.max.file.size' = '5'
);

INSERT INTO hudi_tbl SELECT * from datagen;

解决方案 #

• 修正参数单位：你设置的write.parquet.max.file.size和write.parquet.block.size值为5和1，但Hudi这些参数默认单位是字节，5字节远小于实际生成的文件大小，导致每次提交都触发新文件创建。需改为对应5MB的字节值5242880，同时write.parquet.block.size建议设置为Parquet默认的128MB（即134217728）。

• 开启小文件合并：COPY_ON_WRITE表在insert操作下默认不会追加已有文件，需主动开启合并策略：

  • 添加'write.merge.small.file.enabled' = 'true'启用合并
  • 配置'write.min.merge.file.size' = '5242880'，让小于5MB的文件触发合并
  • 可选设置'write.merge.small.file.limit' = '10'，当小文件数量超过10个时强制合并

• 确认提交机制：确保Flink Checkpoint完成后正常触发Hudi Commit，execution.checkpointing.interval与Hudi的write.commit.interval（默认同步Checkpoint间隔）保持一致即可。

修正后的Flink SQL示例 #

SET 'parallelism.default' = '1';
SET 'execution.checkpointing.interval' = '1m';

CREATE TABLE datagen
(
    id   INT NOT NULL PRIMARY KEY NOT ENFORCED,
    data STRING
) WITH (
      'connector' = 'datagen',
      'rows-per-second' = '5'
);

CREATE TABLE hudi_tbl
(
    id   INT NOT NULL PRIMARY KEY NOT ENFORCED,
    data STRING
) WITH (
      'connector' = 'hudi',
      'path' = 'file:///opt/hudi',
      'table.type' = 'COPY_ON_WRITE',
      'write.parquet.block.size' = '134217728', -- 128MB
      'write.operation' = 'insert',
      'write.parquet.max.file.size' = '5242880', -- 5MB
      'write.merge.small.file.enabled' = 'true',
      'write.min.merge.file.size' = '5242880'
);

INSERT INTO hudi_tbl SELECT * from datagen;

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