Stream zip file to S3 using multi-part upload

[rubenrubiob]

Hello,

I have been banging my head all week with a limitation of the S3 compatible storages. I have to generate a zip file containing thousands of files already stored in an S3 bucket. My code looks a lot like the one shown by Chris White in this wonderful post, which I copy here as reference (hope is okay by him):

<?php
 
require './vendor/autoload.php';
 
use Aws\S3\S3Client;
use ZipStream\ZipStream;
use ZipStream\CompressionMethod;
 
$s3 = new S3Client([
    'region' => 'ca-central-1',
    'version' => '2006-03-01',
]);
 
// We need to call this method to allow PHP streams to work with the `s3://` protocol.
$s3->registerStreamWrapper();
 
// We define our list of files that we want to export into a ZIP file. Real production
// code would define this dynamically, we're just hard-coding it for an example.
$files = [
    new class {
        public string $name = 'test_photo.jpeg';
        public string $path = 's3://zipstream-testing/test_photo.jpeg';
    },
    new class {
        public string $name = 'test_photo2.png';
        public string $path = 's3://zipstream-testing/test_photo2.png';
    },
    new class {
        public string $name = 'test_video.mov';
        public string $path = 's3://zipstream-testing/test_video.mov';
    },
    new class {
        public string $name = 'test_video2.mov';
        public string $path = 's3://zipstream-testing/test_video2.mov';
    },
];
 
// We open a stream to the destination location of the ZIP export. In this case we're
// just placing it into the same bucket alongside the files we're exporting, but this
// could go to another S3 bucket, or anywhere else that you can fopen().
$archiveDestination = fopen('s3://zipstream-testing/archive.zip', 'w');
 
$zip = new ZipStream(
    // Configures the library to stream its output to our destination archive.
    outputStream: $archiveDestination,
 
    // We turn off ZIP compression to make the export faster. Omit this parameter
    // entirely if you do want to compress the resulting ZIP file.
    defaultCompressionMethod: CompressionMethod::STORE,
 
    // This parameter is required to work with remote streams.
    defaultEnableZeroHeader: true,
 
    // We're not sending the ZIP output to the user's browser, but back to S3.
    sendHttpHeaders: false,
);
 
// Iterate over each of our files, fopen() them, add them to the ZIP, then close the open stream.
foreach ($files as $file) {
    $fileStream = fopen($file->path, 'r');
    $zip->addFileFromStream($file->name, $fileStream);
    fclose($fileStream);
}
 
// Finally, finish the ZIP file which will write the required footer data and close the stream
// to our destination archive.
$zip->finish();
fclose($archiveDestination);
 
echo sprintf("Peak memory usage: %d MiB", (memory_get_peak_usage(true)/1024/1024));

As it turns out, any storage that I tried, i.e., Wasabi and Cloudflare's R2, have a limitation of 5 GiB when using single upload. AWS S3's has this limitation too.

The thing is, the compressed files I generated using ZipStream are larger than 20 GiB. My commands work, but the file is silently discarded by the storage and never shown, due to the limitation.

However, when using Cloudflare's R2 multi-part the file size limit is of 4.995 TiB. In S3 the limit is of 48.8 TiB.

Therefore, my question is if there exists a way to use a multi-part upload with ZipStream to compress the files directly in the storage, without touching my server.

Right now, my solution has been to split the zip files in batches that not go over 5 GiB, but I would like to have only one zip, if possible.

Thank you!

[maennchen]

@rubenrubiob This anwer is very theoretical, I’m not aware of anyone ever trying this.

Theoretically, you can use any php resource or psr stream as a target for the archive. This should allow you to create an output stream which starts a multipart upload, then chunk the outputs by the cunk sizes, upload them individually and finish the multipart upload once the zip is finished.

I would love to have an example for this in the docs as well.

[rubenrubiob]

@maennchen Thank you for your answer!

I was thinking something similar. Actually, checking S3's multipart upload docs, we have this code:

// Create an S3Client.
$s3Client = new S3Client([
    'profile' => 'default',
    'region' => 'us-east-2',
    'version' => '2006-03-01'
]);

$bucket = 'your-bucket';
$key = 'my-file.zip';

// Use a stream instead of a file path.
$source = fopen('/path/to/large/file.zip', 'rb');

$uploader = new ObjectUploader(
    $s3Client,
    $bucket,
    $key,
    $source
);

do {
    try {
        $result = $uploader->upload();
        if ($result["@metadata"]["statusCode"] == '200') {
            print('<p>File successfully uploaded to ' . $result["ObjectURL"] . '.</p>');
        }
        print($result);
        // If the SDK chooses a multipart upload, try again if there is an exception.
        // Unlike PutObject calls, multipart upload calls are not automatically retried.
    } catch (MultipartUploadException $e) {
        rewind($source);
        $uploader = new MultipartUploader($s3Client, $source, [
            'state' => $e->getState(),
        ]);
    }
} while (!isset($result));

fclose($source);

I think I would need to open ZipStream to a stream such as php://temp and, then, change the source used by ObjectUploader to that same stream:

$source = fopen('php://temp', 'rb');

I am not sure if we can rewind the source in case of an error. We will see.

I would be glad to write an example for the documentation if I can get it to work.

[maennchen]

The problem with writing to php://temp is that you will accumulate the complete archive and only be able to write it afterwards. That might not be a great idea with archives this size.

In this case, I would recommend to implement either a custom streamWrapper or a PSR StreamInterface that writes the data off to S3 when it is being written.

You could for example create an internal buffer, all write calls up to a defined chunk size are appended to the buffer. Once the chunk is full, you write off the chunk and start to buffer the next one. That way PHP can garbage collect the old buffers and you should be able to run this with a fraction of the memory.

[rubenrubiob]

That was what I was afraid of. I will try an implementation of a PSR StreamInterface. Guzzle's BufferStream seems like a good starting point.

Looking at AWS's MultipartUploader, it looks I will have to call:

• CreateMultipartUpload before starting
• UploadPart when the buffer's chunk size is reached
• CompleteMultipartUpload after the last file is written to zip

There is a limit as to how many parts can a multipart upload have, so I would have to have a large chunk, such as 500 MiB or so. That would occupy PHP's memory, but I think it is needed.

Thank you for your answer!

[maennchen]

@rubenrubiob That sounds like you're on the right track.

I'm looking forward to see what you can come up with.

[rubenrubiob]

I will try to look into it this weekend. I will let you know what I find!

[rubenrubiob]

Well, it looks like I got it. Thanks to your help @maennchen it was easier than I thought it would be.

I found an AWS low level multipart upload example in the official documentation and that helped me to know how to use the required commands.

Following your advice, I implemented a StreamInterface that, whenever it reaches a limit, uploads a part within that multipart upload. Here is the code:

<?php

declare(strict_types=1);

use Aws\S3\S3ClientInterface;
use Psr\Http\Message\StreamInterface;

use function strlen;
use function substr;

final class MultipartUploadBufferStream implements StreamInterface
{
    private const int PART_SIZE = 5242880; // 5 MiB in bytes

    private string $buffer = '';
    private int $bufferSize = 0;
    private int $partNumber = 1;
    private array $parts = [];

    private string $uploadId;

    public function __construct(
        private readonly string $destinationFileName,
        private readonly string $bucket,
        private readonly S3ClientInterface $client
    ) {
        $result = $this->client->createMultipartUpload(
            [
                'Bucket'       => $bucket,
                'Key'          => $destinationFileName,
                'StorageClass' => 'REDUCED_REDUNDANCY',
            ]
        );

        $this->uploadId = $result['UploadId'];

        $this->parts['Parts'] = [];
    }

    public function write($string): int
    {
        $chunkSize           = strlen($string);

        $this->buffer     .= $string;
        $this->bufferSize += $chunkSize;

        if ($this->bufferSize >= self::PART_SIZE) {
            $this->uploadPart();
        }

        return $chunkSize;
    }

    public function close(): void
    {
        // Upload remaining closing bytes from zip
        $this->uploadPart();

        $this->client->completeMultipartUpload([
            'Bucket'          => $this->bucket,
            'Key'             => $this->destinationFileName,
            'UploadId'        => $this->uploadId,
            'MultipartUpload' => $this->parts,
        ]);

        $this->buffer = '';
        $this->bufferSize = 0;
    }

    private function uploadPart(): void
    {
        $result = $this->client->uploadPart([
            'Bucket'     => $this->bucket,
            'Key'        => $this->destinationFileName,
            'UploadId'   => $this->uploadId,
            'PartNumber' => $this->partNumber,
            'Body'       => $this->buffer,
        ]);

        $this->buffer = '';
        $this->bufferSize = 0;

        $this->parts['Parts'][$this->partNumber] = [
            'PartNumber' => $this->partNumber,
            'ETag' => $result['ETag'],
        ];

        $this->partNumber++;
    }

    public function __toString(): string
    {
        return $this->getContents();
    }

    public function getContents(): string
    {
        $buffer = $this->buffer;
        $this->buffer = '';
        $this->bufferSize = 0;

        return $buffer;
    }

    public function detach(): null
    {
        $this->close();

        return null;
    }

    public function getSize(): int
    {
        return $this->bufferSize;
    }

    public function isReadable(): bool
    {
        return true;
    }

    public function isWritable(): bool
    {
        return true;
    }

    public function isSeekable(): bool
    {
        return false;
    }

    public function rewind(): void
    {
        $this->seek(0);
    }

    public function seek($offset, $whence = SEEK_SET): void
    {
        throw new \RuntimeException('Cannot seek a BufferStream');
    }

    public function eof(): bool
    {
        return $this->bufferSize === 0;
    }

    public function tell(): int
    {
        throw new \RuntimeException('Cannot determine the position of a BufferStream');
    }

    public function read($length): string
    {
        $currentLength = $this->bufferSize;

        if ($length >= $currentLength) {
            // No need to slice the buffer because we don't have enough data.
            $result = $this->buffer;
            $this->buffer = '';
            $this->bufferSize = 0;
        } else {
            // Slice up the result to provide a subset of the buffer.
            $result = substr($this->buffer, 0, $length);
            $this->buffer = substr($this->buffer, $length);
            $this->bufferSize -= $length;
        }

        return $result;
    }

    public function getMetadata($key = null)
    {
        return $key ? null : [];
    }
}

The implementation is based on Guzzle's BufferStream. Key points are:

• We create a multipart upload when we construct the object. We must provide the destination filename, the bucket and the S3 client.
• When the buffer is written into, we upload a part whenever the chunk size is reached. In this case, the chunk size is of 5 MiB, the minimum allowed size for a multipart upload.
• When we close the buffer, we upload the remaining bytes, i.e., the zip footer and we close the multipart upload.

This implementation is tightly coupled to S3. I have not thought about decoupling it, but it might be okay, as it is a concrete implementation. I have not yet checked for exceptions and retries, but it is something that should be done.

With this implementation in place, we can use it with ZipStream like so:

<?php

declare(strict_types=1);

$client = new S3Client([
    'region' => 'ca-central-1',
    'version' => '2006-03-01',
]);

$bucketName = 'zip-example';

$this->client->registerStreamWrapper();

$files = return [
    [
        'name' => 'dummy-01.dat',
        'path' => 's3://zip-example/zip-test/dummy-01.dat',
    ],
    [
        'name' => 'dummy-02.dat',
        'path' => 's3://zip-example/zip-test/dummy-02.dat',
    ],
    [
        'name' => 'dummy-03.dat',
        'path' => 's3://zip-example/zip-test/dummy-03.dat',
    ],
    [
        'name' => 'dummy-04.dat',
        'path' => 's3://zip-example/zip-test/dummy-04.dat',
    ],
    [
        'name' => 'dummy-05.dat',
        'path' => 's3://zip-example/zip-test/dummy-05.dat',
    ],
    [
        'name' => 'dummy-06.dat',
        'path' => 's3://zip-example/zip-test/dummy-06.dat',
    ],
    [
        'name' => 'dummy-07.dat',
        'path' => 's3://zip-example/zip-test/dummy-07.dat',
    ],
];

$destination = new MultipartUploadBufferStream(
    'zip-test/archive-multipart.zip',
    $bucketName,
    $client,
);

$zip = new ZipStream(
    outputStream: $destination,
    defaultCompressionMethod: CompressionMethod::STORE,
    defaultEnableZeroHeader: true,
    sendHttpHeaders: false,
);

foreach ($this->files() as $file) {
    $fileStream = fopen($file['path'], 'r');
    $zip->addFileFromStream($file['name'], $fileStream);
    fclose($fileStream);
}

$zip->finish();
$destination->close();

echo sprintf("Peak memory usage: %d MiB", (memory_get_peak_usage(true) / 1024 / 1024));

My understanding was that $zip->finish() would call the StreamInterface::close() method, but it does not. Therefore, we have to directly call the close method in order to write the remaining, closing bytes of the zip and to finish the multipart upload.

I did a test. I generated 7 dummy files of 1 GiB and compressed them into one, using Wasabi as a storage. It worked perfectly. However, the peak memory usage was of 4154 MiB. I am using a Symfony application, but it does not consume that much memory. Maybe it is because it had a lot of parts? Or because each individual file size was of 1 GiB?

@maennchen Do you think this would work as an example for the documentation? I could try to generate something more adequate, if you want to.

In any case, I thank you for your help and for your work in this library!

[maennchen]

Awesome. Peak Memory Usage is not always as straightforward as you might think. As long as you’re not under pressure, GC might not run directly and you might get elevated results.

But this clearly works with a fraction of the total size of the archive.

I would love a PR adding this to the docs. Maybe trimmed down a bit to the relevant parts. For example we could just add grabage instead of S3 files in the example since there’s already an example downloading from S3 separately.

[rubenrubiob]

I will be glad to open a PR with an example for the docs. Why do you mean with "we could just add grabage"? I think I do not follow you there.

About memory usage, I was more curious than anything, because it seemed a lot of memory usage. Maybe it is because of the readStream of such a big object? In any case, I am not really worried about it, since this would be a process to execute on the CLI or a consumer and not on an http request.

[maennchen]

The example could just use /a/very/big/file to add instead of downloading the source files from S3. That way we can clearly focus on the explanation of the multipart upload.

On memory usage, it would be important to have a look at garbage collection. There’s a lot of knobs to set it up correctly. The code you’ve posted looks like it should not cause any leaks.

One thing you could try is using a php://temp stream for the buffer since that should be cleaned up more instantly once you close it.

[rubenrubiob]

Yes, that makes more sense. The origin of the files is actually not important in this case, as it can be anything. That is the great thing about this library :) I hope will be able to open a PR next weekend.

As for the memory usage, I found the culprit. To save time with my tests, I switched to use 7 files of 100 MiB, and chunks of 50 MiB, generating a zip file of a total of 700 MiB. With the same example, the memory usage was 724 MiB. I used Symfony's profiler to measure it, and, without executing anything, it consumed 22 Mib, so I was sure the files were kept in memory somewhere. I implemented the buffer using php://temp as you mentioned, but the results were exactly the same.

That made me think there was something that did not get cleaned up, so I generated a profiler using xDebug, and this is what I got:

Analyzing it, I found the main memory usage was on AWS's client side. I found an old issue where they mentioned a leakage in exactly the same action, i.e., uploading a multipart file. The solution was to free the memory after uploading each part. So, I had to change the uploadPart method of my implementation, such as:

private function uploadPart(): void
    {
        $result = $this->client->uploadPart([
            'Bucket'     => $this->bucket,
            'Key'        => $this->destinationFileName,
            'UploadId'   => $this->uploadId,
            'PartNumber' => $this->partNumber,
            'Body'       => $this->buffer,
        ]);

        $this->buffer = '';
        $this->bufferSize = 0;

        $this->parts['Parts'][$this->partNumber] = [
            'PartNumber' => $this->partNumber,
            'ETag' => $result['ETag'],
        ];

        $this->partNumber++;

+++        $result = null;
+++        gc_collect_cycles();
    }

After this change, memory consumption for the whole compression was 24 MiB. Not counting initialization time, the whole compression process only takes 2 MiB! I think that is pretty neat. Memory consumption is exactly the same when using 1 GiB files.

Just as a reference in case it is useful, here is the adapted code with the relevant parts using php://temp instead of a string as a buffer. Memory consumption is exactly the same, so I prefer the string implementation.

final class MultipartUploadBufferStream implements StreamInterface
{
    private const int PART_SIZE = 52428800; // 50 MiB in bytes

    /** @var resource */
    private $buffer;
    private int $bufferSize = 0;
    private int $partNumber = 1;
    private array $parts = [];

    private string $uploadId;

    public function __construct(
        private readonly string $destinationFileName,
        private readonly string $bucket,
        private readonly S3ClientInterface $client
    ) {
        $this->buffer = \Safe\fopen('php://temp', 'wb');

        $result = $this->client->createMultipartUpload(
            [
                'Bucket'       => $bucket,
                'Key'          => $destinationFileName,
                'StorageClass' => 'REDUCED_REDUNDANCY',
            ]
        );

        $this->uploadId = $result['UploadId'];

        $this->parts['Parts'] = [];
    }

    public function close(): void
    {
        // Upload remaining closing bytes from zip
        $this->uploadPart();

        $this->client->completeMultipartUpload([
            'Bucket'          => $this->bucket,
            'Key'             => $this->destinationFileName,
            'UploadId'        => $this->uploadId,
            'MultipartUpload' => $this->parts,
        ]);

        $this->buffer = '';
        $this->bufferSize = 0;
    }

    /**
     * Writes data to the buffer.
     */
    public function write($string): int
    {
        $chunkSize           = strlen($string);

        \Safe\fwrite($this->buffer, $string);
        //$this->buffer     .= $string;
        $this->bufferSize += $chunkSize;

        if ($this->bufferSize >= self::PART_SIZE) {
            $this->uploadPart();
        }

        return $chunkSize;
    }

    private function uploadPart(): void
    {
        rewind($this->buffer);


        $result = $this->client->uploadPart([
            'Bucket'     => $this->bucket,
            'Key'        => $this->destinationFileName,
            'UploadId'   => $this->uploadId,
            'PartNumber' => $this->partNumber,
            'Body'       => stream_get_contents($this->buffer),
        ]);

        \Safe\fclose($this->buffer);

        $this->buffer = \Safe\fopen('php://temp', 'wb');
        $this->bufferSize = 0;

        $this->parts['Parts'][$this->partNumber] = [
            'PartNumber' => $this->partNumber,
            'ETag' => $result['ETag'],
        ];

        $this->partNumber++;

        $result = null;
        gc_collect_cycles();
    }

    public function __toString(): string
    {
        return $this->getContents();
    }

    public function getContents(): string
    {
        rewind($this->buffer);

        $buffer = stream_get_contents($this->buffer);
        \Safe\fclose($this->buffer);
        $this->bufferSize = 0;

        return $buffer;
    }
}