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

Question

我正在尝试从 S3 将大文件读取成 block ，而不切割任何行以进行并行处理。

让我通过例子来解释一下: S3上有1G大小的文件。我想将此文件分成 64 MB 的 block 。这很容易，我可以这样做:

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 

}

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

有什么办法可以处理这种情况吗？表示所有卡盘都没有偏线。

Answer 1

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

所以我创建了一个新类 S3InputStream ，它不关心它打开多长时间，并使用短暂的 AWS SDK 调用按需读取字节 block 。您提供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();
    }
}