如何在没有 MemoryStream 或本地文件的情况下将大量对象直接流式传输到 S3？

Question

我正在尝试将一个大型对象写入 AWS S3（例如 25 GB）。

目前我可以通过两种方式让它工作：

• 将内容写入本地磁盘上的文件，然后使用multi-part upload将文件发送到S3
• 将内容写入MemoryStream，然后使用分段上传将该流发送到 S3

但是，我不喜欢这两种方法，因为我需要为操作保留大量磁盘空间或内存。我在代码中生成这个内容，所以我希望打开一个到 S3 对象的流，并将内容直接生成到该对象。但我不知道如何让它发挥作用。

是否可以在不先将整个对象表示在本地文件或内存中的情况下在 S3 中构建一个海量对象？

（注意：我的问题与this question 非常相似，但该问题没有有用的答案。）

Answer 1

我能够通过将整个有效负载分成块并将每个单独的块作为单独的 MemoryStream 发送来使其工作。

从技术上讲，这个解决方案仍然使用MemoryStream，但这没关系，因为我可以通过调整块大小来控制使用多少内存。对于我的测试，我创建了一个 25GB 的文件，同时将内存使用量保持在远低于该值 (~2 GB IIRC)。

这是我的解决方案：

private const string BucketName = "YOUR-BUCKET-NAME-HERE";
private static readonly RegionEndpoint BucketRegion = RegionEndpoint.USEast1;
private const string Key = "massive-file-test";

// We're going to send 100 chunks of 256 MB each, for a total of 25 GB.
// The content will be the asterisk ("*") repeated for the desired size.
private const int ChunkSizeMb = 256;
private const int TotalSizeGb = 25;

public static void Main(string[] args)
{
    Console.WriteLine($"Writing object to {BucketName}, {Key}");

    int totalChunks = TotalSizeGb * 1024 / ChunkSizeMb;
    int chunkSizeBytes = ChunkSizeMb * 1024 * 1024;
    string payload = new String('*', chunkSizeBytes);

    // Initiate the request.
    InitiateMultipartUploadRequest initiateRequest = new InitiateMultipartUploadRequest
    {
        BucketName = BucketName,
        Key = Key
    };

    List<UploadPartResponse> uploadResponses = new List<UploadPartResponse>();
    IAmazonS3 s3Client = new AmazonS3Client(BucketRegion);
    InitiateMultipartUploadResponse initResponse = s3Client.InitiateMultipartUpload(initiateRequest);

    // Open a stream to build the input.
    for (int i = 0; i < totalChunks; i++)
    {
        // Write the next chunk to the input stream.
        Console.WriteLine($"Writing chunk {i} of {totalChunks}");
        using (var stream = ToStream(payload))
        {
            // Write the next chunk to s3.
            UploadPartRequest uploadRequest = new UploadPartRequest
            {
                BucketName = BucketName,
                Key = Key,
                UploadId = initResponse.UploadId,
                PartNumber = i + 1,
                PartSize = chunkSizeBytes,
                InputStream = stream,
            };

            uploadResponses.Add(s3Client.UploadPart(uploadRequest));
        }
    }

    // Complete the request.
    CompleteMultipartUploadRequest completeRequest = new CompleteMultipartUploadRequest
    {
        BucketName = BucketName,
        Key = Key,
        UploadId = initResponse.UploadId
    };

    completeRequest.AddPartETags(uploadResponses);
    s3Client.CompleteMultipartUpload(completeRequest);

    Console.WriteLine("Script is complete. Press any key to exit...");
    Console.ReadKey();
}

private static Stream ToStream(string s)
{
    var stream = new MemoryStream();
    var writer = new StreamWriter(stream);
    writer.Write(s);
    writer.Flush();
    stream.Position = 0;
    return stream;
}

Answer 2

这是 AnonCoward 开始的，通过添加 seek 结束 - 对于一个除了将星号写入其缓冲区之外什么都不做的流来说，这是一个微不足道的操作。如果您要生成更复杂的数据，这将是一项艰巨的工作，但要寻找所有您需要做的事情就是设置位置并说“是的，完成了”，因为无论您在流中的哪个位置寻找创建星号的行为始终是一样的

class AsteriskGeneratingStream : Stream
{
    long _pos = 0;
    long _length = 0;
    public AsteriskGeneratingStream(long length)
    {
        _length = length;
    }
    public override long Length => _length;

    public override int Read(byte[] buffer, int offset, int count)
    {
        // Create the data as needed
        if (count + _pos > _length)
            count = (int)(_length - _pos);
        
        for (int i = offset; i < count; i++)
            buffer[i] = (byte)'*';
        
        _pos += count;
        return count;
    }

    public override bool CanRead => true;

    public override long Seek(long offset, SeekOrigin origin)
    { 
        if(origin == SeekOrigin.Begin) //lets just trust that the caller will be sensible and not set e.g. negative offset
            _pos = offset;
        else if(origin == SeekOrigin.Current)
            _pos += offset;
        else if(origin == SeekOrigin.End)
            _pos = _length + offset;

        return _pos;
    }

    public override bool CanSeek => true;
    public override bool CanWrite => false;
    public override long Position { get => _pos; set => _pos = value; }
    public override void Flush() { }
    public override void SetLength(long value) { _length = value; }
    public override void Write(byte[] buffer, int offset, int count) { throw new NotImplementedException(); }
}

class Program
{
    static void Main(string[] args)
    {
        long objectSize = 25L * 1024 * 1024;
        var s3 = new AmazonS3Client(Amazon.RegionEndpoint.USWest1);
        var xfer = new TransferUtility(s3,new TransferUtilityConfig
        {
            MinSizeBeforePartUpload = 5L * 1024 * 1024
        });

        var helper = new AsteriskGeneratingStream(objectSize);
        xfer.Upload(helper, "bucket-name", "object-key");
    }
}

注意，我不能保证它会立即工作，因为我在手机上，无法通过 c# fiddle 测试它，但让我们看看它是如何爆炸的！ ?

Answer 3

如果您可以动态创建对象，或者至少缓存相当小的段，则可以创建一个流，将数据提供到 S3。请注意，除非您还可以乱序创建对象的任何部分，否则您需要防止 AWS SDK 使用分段上传，这会降低传输速度。

    class DataStream : Stream
    {
        long _pos = 0;
        long _length = 0;
        public DataStream(long length)
        {
            _length = length;
        }
        public override long Length => _length;

        public override int Read(byte[] buffer, int offset, int count)
        {
            // Create the data as needed, on demand
            // For this example, just cycle through 0 to 256 in the data over and over again
            if (count + _pos > _length)
            {
                count = (int)(_length - _pos);
            }
            for (int i = 0; i < count; i++)
            {
                buffer[i + offset] = (byte)((_pos + i) % 256);
            }
            _pos += count;
            return count;
        }

        public override bool CanRead => true;

        // Stub out all other methods.  For a seekable stream
        // Seek() and Postion need to be implemented, along with CanSeek changed
        public override long Seek(long offset, SeekOrigin origin) { throw new NotImplementedException(); }
        public override bool CanSeek => false;
        public override bool CanWrite => false;
        public override long Position { get => _pos; set => throw new NotImplementedException(); }
        public override void Flush() { throw new NotImplementedException(); }
        public override void SetLength(long value) { throw new NotImplementedException(); }
        public override void Write(byte[] buffer, int offset, int count) { throw new NotImplementedException(); }
    }

    class Program
    {
        static void Main(string[] args)
        {
            long objectSize = 25L * 1024 * 1024;
            var s3 = new AmazonS3Client(Amazon.RegionEndpoint.USWest1);
            // Prevent a multi-part upload, which requires a seekable stream
            var xfer = new TransferUtility(s3, new TransferUtilityConfig
            {
                MinSizeBeforePartUpload = objectSize + 1
            });

            var helper = new DataStream(objectSize);
            xfer.Upload(helper, "bucket-name", "object-key");
        }
    }