在 S3 中上传输入流块后无法解压缩 gzip 文件

Posted 2023-03-07

【中文标题】在 S3 中上传输入流块后无法解压缩 gzip 文件

【英文标题】：Unable to decompress gzipped files after uploading input stream chunks in S3

【发布时间】：2019-06-24 21:00:15

【问题描述】：

我想以与分段上传器类似的方式获取我的输入流并将压缩后的部分上传到 s3。 但是，我想将单个文件部分存储在 S3 中，而不是将这些部分变成单个文件。

为此，我创建了以下方法。 但是，当我尝试 gzip 解压缩每个部分时，gzip 会抛出一个错误并说：gzip: file_part_2.log.gz: not in gzip format.

我不确定我是否正确压缩了每个部分？

如果我重新初始化 gzipoutputstream:gzip = new GZIPOutputStream(baos); 并在重置字节数组输出流gzip.finish() 后设置baos.reset();，那么我可以解压缩每个部分。不知道为什么我需要这样做，gzipoutputstream 是否有类似的reset？

public void upload(String bucket, String key, InputStream is, int partSize) throws Exception

    String row;
    BufferedReader br = new BufferedReader(new InputStreamReader(is, ENCODING));
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    GZIPOutputStream gzip = new GZIPOutputStream(baos);

    int partCounter = 0;
    int lineCounter = 0;
    while ((row = br.readLine()) != null) 
        if (baos.size() >= partSize) 
            partCounter = this.uploadChunk(bucket, key, baos, partCounter);

            baos.reset();
        else if(!row.equals(""))
            row += '\n';
            gzip.write(row.getBytes(ENCODING));
            lineCounter++;
        
    

    gzip.finish();
    br.close();
    baos.close();

    if(lineCounter == 0)
        throw new Exception("Aborting upload, file contents is empty!");
    

    //Final chunk
    if (baos.size() > 0) 
        this.uploadChunk(bucket, key, baos, partCounter);
    


private int uploadChunk(String bucket, String key, ByteArrayOutputStream baos, int partCounter)

    ObjectMetadata metaData = new ObjectMetadata();
    metaData.setContentLength(baos.size());

    String[] path = key.split("/");
    String[] filename = path[path.length-1].split("\\.");

    filename[0] = filename[0]+"_part_"+partCounter;

    path[path.length-1] = String.join(".", filename);

    amazonS3.putObject(
            bucket,
            String.join("/", path),
            new ByteArrayInputStream(baos.toByteArray()),
            metaData
    );

    log.info("Upload chunk , size: ", partCounter, baos.size());

    return partCounter+1;

【问题讨论】：

你的问题是什么？

我不确定我是否正确压缩了每个部分？

我不明白你为什么在上传之前压缩文件？这似乎与您的问题/问题无关。

也许这个答案有帮助：***.com/questions/37336050/pipe-a-stream-to-s3-upload/…（您可以将流直接传递给s3.upload）

这是 ETL 管道的一部分，在这里分块文件可以让我在下游服务中进行并行处理。这里的问题与 gzip 有关。

【参考方案1】：

问题是您对所有块使用单个GZipOutputStream。因此，您实际上是在编写 GZipped 文件的片段，这些片段必须重新组合才能有用。

对现有代码进行最小的更改：

if (baos.size() >= partSize) 
    gzip.close(); 
    partCounter = this.uploadChunk(bucket, key, baos, partCounter);
    baos = baos = new ByteArrayOutputStream();
    gzip = new GZIPOutputStream(baos);

您需要在循环结束时执行相同的操作。此外，如果行计数器为 0，则不应抛出异常：文件完全有可能被完全分割成一定数量的块。

为了改进代码，我会将GZIPOutputStream 包装在OutputStreamWriter 和BufferedWriter 中，这样您就不需要显式地进行字符串字节转换。

最后，不要使用ByteArrayOutputStream.reset()。与创建新流相比，它不会为您节省任何费用，并且如果您忘记重置，则会为错误打开大门。

【讨论】：

感谢您提供如此详细的回答！