alluxio.underfs

Class ObjectLowLevelOutputStream

java.lang.Object

java.io.OutputStream

alluxio.underfs.ObjectLowLevelOutputStream

All Implemented Interfaces:

ContentHashable, Closeable, Flushable, AutoCloseable

Direct Known Subclasses:

OBSLowLevelOutputStream, OSSLowLevelOutputStream, S3ALowLevelOutputStream

@NotThreadSafe
public abstract class ObjectLowLevelOutputStream
extends OutputStream
implements ContentHashable

[Experimental] A stream for writing a file into object storage using streaming upload. The data transfer is done using object storage low-level multipart upload.

We upload data in partitions. When write(), the data will be persisted to a temporary file mFile on the local disk. When the data mPartitionOffset in this temporary file reaches the mPartitionSize, the file will be submitted to the upload executor mExecutor and we do not wait for uploads to finish. A new temp file will be created for the future write and the mPartitionOffset will be reset to zero. The process goes until all the data has been written to temp files.

In flush(), we upload the buffered data if they are bigger than 5MB and wait for all uploads to finish. The temp files will be deleted after uploading successfully.

In close(), we upload the last part of data (if exists), wait for all uploads to finish, and complete the multipart upload.

close() will not be retried, but all the multipart upload related operations(init, upload, complete, and abort) will be retried.

If an error occurs and we have no way to recover, we abort the multipart uploads. Some multipart uploads may not be completed/aborted in normal ways and need periodical cleanup by enabling the PropertyKey.UNDERFS_CLEANUP_ENABLED. When a leader master starts or a cleanup interval is reached, all the multipart uploads older than clean age will be cleaned.

Field Summary

Fields

Modifier and Type	Field and Description
protected static org.slf4j.Logger	LOG
protected String	mBucketName Bucket name of the object storage bucket.
protected boolean	mClosed Flag to indicate this stream has been closed, to ensure close is only done once.
protected File	mFile The local temp file that will be uploaded when reaches the partition size or when flush() is called and this file is bigger than UPLOAD_THRESHOLD.
protected MessageDigest	mHash The MD5 hash of the file.
protected String	mKey Key of the file when it is uploaded to object storage.
protected OutputStream	mLocalOutputStream The output stream to the local temp file.
protected long	mPartitionOffset When the offset reaches the partition size, we upload the temp file.
protected long	mPartitionSize The maximum allowed size of a partition.
protected RetryPolicy	mRetryPolicy The retry policy of this multipart upload.
protected byte[]	mSingleCharWrite Pre-allocated byte buffer for writing single characters.
protected List<String>	mTmpDirs
protected static long	UPLOAD_THRESHOLD Only parts bigger than 5MB could be uploaded through multipart upload, except the last part.

Constructor Summary

Constructors

Constructor and Description
ObjectLowLevelOutputStream(String bucketName, String key, com.google.common.util.concurrent.ListeningExecutorService executor, long streamingUploadPartitionSize, AlluxioConfiguration ufsConf) Constructs a new stream for writing a file.

Method Summary

Modifier and Type	Method and Description
protected void	abortMultiPartUpload()
protected abstract void	abortMultiPartUploadInternal()
void	close()
protected abstract void	completeMultiPartUploadInternal()
protected abstract void	createEmptyObject(String key)
void	flush()
int	getPartNumber() Get the part number.
protected abstract void	initMultiPartUploadInternal()
protected abstract void	putObject(String key, File file, String md5)
protected void	uploadPart() Uploads part async.
protected void	uploadPart(File file, int partNumber, boolean lastPart)
protected abstract void	uploadPartInternal(File file, int partNumber, boolean isLastPart, String md5)
protected void	waitForAllPartsUpload()
void	write(byte[] b)
void	write(byte[] b, int off, int len)
void	write(int b)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface alluxio.underfs.ContentHashable

getContentHash

Field Detail

LOG

protected static final org.slf4j.Logger LOG

mTmpDirs

protected final List<String> mTmpDirs

UPLOAD_THRESHOLD

protected static final long UPLOAD_THRESHOLD

Only parts bigger than 5MB could be uploaded through multipart upload, except the last part.

See Also:

Constant Field Values

mBucketName

protected final String mBucketName

Bucket name of the object storage bucket.

mKey

protected final String mKey

Key of the file when it is uploaded to object storage.

mRetryPolicy

protected final RetryPolicy mRetryPolicy

The retry policy of this multipart upload.

mSingleCharWrite

protected final byte[] mSingleCharWrite

Pre-allocated byte buffer for writing single characters.

mHash

@Nullable
protected MessageDigest mHash

The MD5 hash of the file.

mClosed

protected boolean mClosed

Flag to indicate this stream has been closed, to ensure close is only done once.

mPartitionOffset

protected long mPartitionOffset

When the offset reaches the partition size, we upload the temp file.

mPartitionSize

protected final long mPartitionSize

The maximum allowed size of a partition.

mFile

@Nullable
protected File mFile

The local temp file that will be uploaded when reaches the partition size or when flush() is called and this file is bigger than UPLOAD_THRESHOLD.

mLocalOutputStream

@Nullable
protected OutputStream mLocalOutputStream

The output stream to the local temp file.

Constructor Detail

ObjectLowLevelOutputStream

public ObjectLowLevelOutputStream(String bucketName,
                                  String key,
                                  com.google.common.util.concurrent.ListeningExecutorService executor,
                                  long streamingUploadPartitionSize,
                                  AlluxioConfiguration ufsConf)

Constructs a new stream for writing a file.

Parameters:

bucketName - the name of the bucket

key - the key of the file

streamingUploadPartitionSize - the size in bytes for partitions of streaming uploads

executor - executor

ufsConf - the object store under file system configuration

Method Detail

write

public void write(int b)
           throws IOException

Specified by:

write in class OutputStream

Throws:

IOException

write

public void write(byte[] b)
           throws IOException

Overrides:

write in class OutputStream

Throws:

IOException

write

public void write(byte[] b,
                  int off,
                  int len)
           throws IOException

Overrides:

write in class OutputStream

Throws:

IOException

flush

public void flush()
           throws IOException

Specified by:

flush in interface Flushable

Overrides:

flush in class OutputStream

Throws:

IOException

close

public void close()
           throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Overrides:

close in class OutputStream

Throws:

IOException

uploadPart

protected void uploadPart()
                   throws IOException

Uploads part async.

Throws:

IOException

uploadPart

protected void uploadPart(File file,
                          int partNumber,
                          boolean lastPart)

abortMultiPartUpload

protected void abortMultiPartUpload()
                             throws IOException

Throws:

IOException

waitForAllPartsUpload

protected void waitForAllPartsUpload()
                              throws IOException

Throws:

IOException

getPartNumber

public int getPartNumber()

Get the part number.

Returns:

the part number

uploadPartInternal

protected abstract void uploadPartInternal(File file,
                                           int partNumber,
                                           boolean isLastPart,
                                           @Nullable
                                           String md5)
                                    throws IOException

Throws:

IOException

initMultiPartUploadInternal

protected abstract void initMultiPartUploadInternal()
                                             throws IOException

Throws:

IOException

completeMultiPartUploadInternal

protected abstract void completeMultiPartUploadInternal()
                                                 throws IOException

Throws:

IOException

abortMultiPartUploadInternal

protected abstract void abortMultiPartUploadInternal()
                                              throws IOException

Throws:

IOException

createEmptyObject

protected abstract void createEmptyObject(String key)
                                   throws IOException

Throws:

IOException

putObject

protected abstract void putObject(String key,
                                  File file,
                                  @Nullable
                                  String md5)
                           throws IOException

Throws:

IOException