Apache Flink中Parquet bulk格式的compaction使用方法咨询

从Apache Flink 1.15版本开始，用户可使用Compaction特性将多个小文件合并为单个文件，默认内置的RecordWiseFileCompactor下的DecoderBasedReader、InputFormatBasedReader实现仅适配行式编码格式，无法直接支持Parquet这类Bulk列存格式，目前官方也未提供Parquet等Bulk格式的Compaction操作示例，可按如下方案实现Parquet格式的Compaction能力：

• 替换Compactor实现类：不要使用面向行式场景的RecordWiseFileCompactor，选择BulkFileCompactor作为基类实现Parquet专属合并逻辑，该类是Flink为块级读写的二进制/列存格式设计的专用Compactor抽象，不需要逐行解析文件内容。
• 实现块级合并逻辑：编写Parquet Compactor逻辑时，不需要将源文件解码为逐行记录再重新写入，直接读取Parquet文件的RowGroup块，通过原生Parquet写入API将块直接追加到目标合并文件即可，整个过程不会破坏Parquet原有的列索引、压缩编码和块统计信息，性能远高于逐行读写方案。
  注意：强行使用DecoderBasedReader或InputFormatBasedReader适配Parquet会导致列存特性失效、压缩率下降、读写性能骤降，生产环境不建议使用。
• 配置启用Compaction：初始化FileSink时开启Bulk格式Compaction开关，传入自定义的Parquet Compactor实现，配置参考如下：

FileSink<YourRecordType> parquetSink = FileSink
    .forBulkFormat(targetPath, ParquetAvroWriters.forReflectRecord(YourRecordType.class))
    .enableCompaction(CompactionConfig.builder()
        .setCompactorClass(CustomParquetBulkCompactor.class)
        .setTargetFileSize(128 * 1024 * 1024) // 配置合并后单目标文件大小，可按需调整
        .build())
    .build();

实现自定义Parquet BulkCompactor时只需要做文件块的拷贝拼接，不需要做额外的数据格式转换，整体实现代码量不超过100行，生产环境使用稳定性和性能都能得到保障。目前Flink官方暂未内置Parquet、ORC等主流Bulk格式的开箱即用Compactor实现，相关内置能力会在后续版本逐步迭代上线。

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