Flink写入HDFS的Parquet文件过小问题求助（目标100MB）

1. 修正ParquetWriter参数错误 #

你代码里的withRowGroupSize(8*1024*10124)存在明显计算笔误，10124应该是1024。且如果目标是100MB左右的文件，需要把行组大小设置为接近目标值（行组是Parquet的核心存储单元，写满后才会刷入磁盘，压缩后会略小于设置值）：

AvroParquetWriter.<GenericRecord>builder(filePath)
 .withSchema(schema)
 .withCompressionCodec(CompressionCodecName.SNAPPY)
 .withConf(Configuration)
 .withDataModel(GenericData.get())
 .withWriteMode(Mode.OVERWRITE)
 .withRowGroupSize(100 * 1024 * 1024) // 设置为100MB行组
 .withPageSize(1 * 1024 * 1024) // PageSize无需过大，1MB左右足够，过大徒增内存消耗
 .build()

2. 调整路径分片逻辑 #

你的路径生成用tight%num of rows per file和counter/num of rows per file切割文件，问题大概率是num of rows per file设置过小。需要根据单条记录的平均大小估算目标行数：
比如单条记录平均1KB，100MB需要约102400条记录，调整参数：

// 示例：按单条1KB估算，100MB对应行数
int rowsPerFile = 100 * 1024;
String path = "hdfsLocation" + String.format("%d_%d.parquet", tid % rowsPerFile, counter / rowsPerFile);

3. 改用Flink官方FileSink（推荐） #

直接使用AvroParquetWriter在Flink分布式环境下很难精准控制文件大小，官方FileSink提供了完善的滚动策略和小文件合并机制，更适合流/批场景：

// 构建Parquet格式的FileSink
FileSink<GenericRecord> parquetSink = FileSink
    .forBulkFormat(new Path("hdfsLocation"), AvroParquetWriters.forGenericRecord(schema))
    .withBucketAssigner(new SimpleStringBucketAssigner<>()) // 可自定义分片规则，比如按业务key
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withMaxPartSize(100 * 1024 * 1024) // 文件达到100MB时自动滚动
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(30)) // 可选：超时强制滚动
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(10)) // 可选：无数据超时滚动
            .build()
    )
    .withOutputFileConfig(
        OutputFileConfig.builder()
            .withPartPrefix("data")
            .withPartSuffix(".parquet")
            .build()
    )
    .build();

// 绑定到数据流
dataStream.sinkTo(parquetSink);

4. 调整作业并行度 #

如果Flink作业并行度过高，每个并行子任务处理的数据量不足，会生成大量小文件。可以根据集群资源调整并行度：

// 设置全局并行度
env.setParallelism(4);
// 或单独设置sink并行度
parquetSink.setParallelism(4);

5. 流处理场景开启Checkpoint #

流处理中，FileSink需要依赖Checkpoint触发文件从“临时状态”转为“完成状态”，同时可配置小文件合并：

env.enableCheckpointing(TimeUnit.MINUTES.toMillis(5)); // 每5分钟触发一次Checkpoint

内容的提问来源于stack exchange，提问作者Mohammad Aamir Iqubal