Apache Flink的ZIP压缩输入

这个 FileInputFormat 将读取压缩文件委托给 GZIPInputStream ,它将在解压缩时返回部分解压缩的数据。

我想分享我同时实现的解决方案。

所以,在创建了我自己的 InputFormat 我在 open() 方法:

@Override
public void open(final FileInputSplit ignored) throws IOException {
    ...
    final XMLInputFactory xmlif = XMLInputFactory.newInstance();
    final XMLStreamReader xmlr = xmlif.createXMLStreamReader(filePath.toString(),
              InputFormatUtil.readFileWithinZipArchive(filePath, nestedXmlFileName));
    while (xmlr.hasNext()) {
    ...
}

其中 readFileWithinZipArchive(...) 是:

public static InputStream readFileWithinZipArchive(final Path zipPath, final String filename) throws IOException {
    // using org.apache.flink.core.fs.Path for getting the InputStream from the (remote) zip archive
    final InputStream zipInputStream = zipPath.getFileSystem().open(zipPath);
    // generating a temporary local copy of the zip file
    final File tmpFile = stream2file(zipInputStream);
    // then using java.util.zip.ZipFile for extracting the InputStream for the specific file within the zip archive
    final ZipFile zipFile = new ZipFile(tmpFile);
    return zipFile.getInputStream(zipFile.getEntry(filename));
}