解压缩数据流中的.tar文件？

Question

我的GCP云存储桶中有很多.tar文件。每个.tar文件都有多个图层。我想使用GCP数据流解压缩这些.tar文件，并将它们放回另一个GCP存储桶中。

我找到了Google提供的用于批量解压缩云存储文件的实用工具模板，但它不支持.tar文件扩展名。

也许我应该在上传到云端之前尝试解压文件，或者Beam中是否存在其他内容？

每个tar文件未经压缩大约有15 TB。

Answer 1

这个片段借用了批量解压缩模板的代码。它也借用了这个问题

正如您所注意到的，不支持TAR，但一般来说，Beam中的压缩/解压缩似乎依赖于apachecommons的压缩库。

您可以编写一个管道来执行以下操作：

// Create the pipeline
Pipeline pipeline = Pipeline.create(options);

// Run the pipeline over the work items.
PCollectionTuple decompressOut =
    pipeline
        .apply("MatchFile(s)",
            FileIO.match().filepattern(options.getInputFilePattern()))
        .apply(
            "DecompressFile(s)",
            ParDo.of(new Decompress(options.getOutputDirectory());

您的＜code＞解压缩

class Dearchive extends DoFn<MatchResult.Metadata, String> {
  @ProcessElement
  public void process(@Context ProcessContext context) {
    ResourceId inputFile = context.element().resourceId();
    String outputFilename = Files.getNameWithoutExtension(inputFile.toString());
    ResourceId tempFileDir =
          this.outputDir.resolve(outputFilename, StandardResolveOptions.RESOLVE_DIRECTORY);
    
    TarArchiveInputStream tarInput = new TarArchiveInputStream(
        Channels.newInputStream(FileSystems.open(inputFile)));

    TarArchiveEntry currentEntry = tarInput.getNextTarEntry();

    while (currentEntry != null) {
        br = new BufferedReader(new InputStreamReader(tarInput)); // Read directly 
        ResourceId outputFile = tempFileDir.resolve(currentEntry.getName(), 
            StandardResolveOptions.RESOLVE_FILE);
        try (WritableByteChannel writerChannel = FileSystems.create(tempFile, MimeTypes.TEXT)) {
          ByteStreams.copy(tarInput, Channels.newOutputStream(writerChannel));
        }
        context.output(outputFile.toString());
        currentEntry = tarInput.getNextTarEntry(); // Iterate to the next file
    }
  }
}

这是一个非常粗糙且未经测试的代码片段，但它应该让你走上正确的道路。LMK，如果我们应该进一步澄清。