如何使用 aws-java-sdk 从 S3 逐块读取文件

Question

我正在尝试将大文件从 S3 读取到块中，而无需为并行处理切割任何行。

举例说明：S3上有1G大小的文件。我想将此文件分成 64 MB 的夹头。我可以很容易地做到：

S3Object s3object = s3.getObject(new GetObjectRequest(bucketName, key));

InputStream stream = s3object.getObjectContent();

byte[] content = new byte[64*1024*1024];

while (stream.read(content)  != -1) {

//process content here 

}

但是块的问题是它可能有 100 行完整的行和一个不完整的行。但我无法处理不完整的行，也不想丢弃它。

有什么办法可以处理这种情况吗？表示所有夹头都没有局部线。

Answer 1

我通常的方法（InputStream-> BufferedReader.lines()-> 成批的行 -> CompletableFuture）在这里不起作用，因为S3ObjectInputStream对于大文件，底层最终会超时。

所以我创建了一个新类S3InputStream，它不关心它的开放时间和使用短期 AWS SDK 调用按需读取字节块。您提供一个byte[]将被重用的。new byte[1 << 24](16Mb) 似乎运行良好。

package org.harrison;

import java.io.IOException;
import java.io.InputStream;

import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.GetObjectRequest;

/**
 * An {@link InputStream} for S3 files that does not care how big the file is.
 *
 * @author stephen harrison
 */
public class S3InputStream extends InputStream {
    private static class LazyHolder {
        private static final AmazonS3 S3 = AmazonS3ClientBuilder.defaultClient();
    }

    private final String bucket;
    private final String file;
    private final byte[] buffer;
    private long lastByteOffset;

    private long offset = 0;
    private int next = 0;
    private int length = 0;

    public S3InputStream(final String bucket, final String file, final byte[] buffer) {
        this.bucket = bucket;
        this.file = file;
        this.buffer = buffer;
        this.lastByteOffset = LazyHolder.S3.getObjectMetadata(bucket, file).getContentLength() - 1;
    }

    @Override
    public int read() throws IOException {
        if (next >= length) {
            fill();

            if (length <= 0) {
                return -1;
            }

            next = 0;
        }

        if (next >= length) {
            return -1;
        }

        return buffer[this.next++];
    }

    public void fill() throws IOException {
        if (offset >= lastByteOffset) {
            length = -1;
        } else {
            try (final InputStream inputStream = s3Object()) {
                length = 0;
                int b;

                while ((b = inputStream.read()) != -1) {
                    buffer[length++] = (byte) b;
                }

                if (length > 0) {
                    offset += length;
                }
            }
        }
    }

    private InputStream s3Object() {
        final GetObjectRequest request = new GetObjectRequest(bucket, file).withRange(offset,
                offset + buffer.length - 1);

        return LazyHolder.S3.getObject(request).getObjectContent();
    }
}

Answer 2

aws-java-sdk 已为您的 S3 对象提供流功能。您必须调用“getObject”，结果将是一个InputStream。

1) AmazonS3Client.getObject(GetObjectRequest getObjectRequest) -> S3Object

2) S3Object.getObjectContent()

注意：该方法是一个简单的 getter，并不实际创建流。如果您检索 S3Object，则应尽快关闭此输入流，因为对象内容不会在内存中缓冲并直接从 Amazon S3 进行流式传输。此外，未能关闭此流可能会导致请求池被阻塞。

AWS Java 文档