Flink ConcatFileCompactor处理GZIP文件失效问题求助

问题场景 #

使用Flink FileSink将GZIP压缩文件写入HDFS，期望合并小文件。由于误以为GZIP文件可直接拼接，尝试使用ConcatFileCompactor，但合并后的文件损坏，无法正常读取。相关代码如下：

FileSink<JsonNode> fileSink = FileSink.<JsonNode>forBulkFormat(
            new Path(hdfsUrl),
            new CompressWriterFactory<JsonNode>(new DefaultExtractor<>()).withHadoopCompression("GzipCodec")
    )
            .withBucketAssigner(new JsonNodeBucketAssigner())
            .enableCompact(FileCompactStrategy.Builder.newBuilder()
                    .setNumCompactThreads(hdfsCompactNumThread)
                    .setSizeThreshold(hdfsCompactThreshold)
                    .build(),
                    new ConcatFileCompactor())
            .build();

损坏原因 #

GZIP文件并非可以无限制直接拼接：

• 每个由Flink批量写入生成的GZIP文件，尾部都带有CRC32校验和与未压缩数据长度的元信息。直接拼接后，读取工具会将第一个文件的尾部元信息识别为整个文件的结束标识，后续内容会被判定为无效数据，导致文件损坏。
• 部分GZIP实现的文件开头还有额外头信息，多文件拼接后多个头信息并存，也会触发解析错误。

ConcatFileCompactor能否正常工作？ #

目前无法让ConcatFileCompactor直接兼容GZIP格式，该压缩器仅做简单的字节级拼接，不会处理GZIP文件特有的元信息（比如移除中间文件的尾部校验、合并全局统计信息）。相关问题已记录在Flink官方工单FLINK-32562中。

替代方案：RecordWiseFileCompactor #

如果需要合并GZIP文件，RecordWiseFileCompactor是当前可行的选择：

• 它会重新读取每个小文件的原始记录，再重新压缩写入合并后的文件，生成的GZIP文件完整有效。
• 缺点是资源消耗更高（需要解压原始文件、重新压缩），但这是保证GZIP文件有效性的必要操作。

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