org.apache.flink.fs.s3.common.writer

Class S3RecoverableWriter

java.lang.Object

org.apache.flink.fs.s3.common.writer.S3RecoverableWriter

All Implemented Interfaces:

RecoverableWriter

@PublicEvolving
public class S3RecoverableWriter
extends Object
implements RecoverableWriter

An implementation of the RecoverableWriter against S3.

This implementation makes heavy use of MultiPart Uploads in S3 to persist intermediate data as soon as possible.

This class partially reuses utility classes and implementations from the Hadoop project, specifically around configuring S3 requests and handling retries.

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.core.fs.RecoverableWriter

RecoverableWriter.CommitRecoverable, RecoverableWriter.ResumeRecoverable

Method Summary

Modifier and Type	Method and Description
boolean	cleanupRecoverableState(RecoverableWriter.ResumeRecoverable resumable) Frees up any resources that were previously occupied in order to be able to recover from a (potential) failure.
SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable>	getCommitRecoverableSerializer() The serializer for the CommitRecoverable types created in this writer.
SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable>	getResumeRecoverableSerializer() The serializer for the ResumeRecoverable types created in this writer.
RecoverableFsDataOutputStream	open(Path path) Opens a new recoverable stream to write to the given path.
S3RecoverableFsDataOutputStream	recover(RecoverableWriter.ResumeRecoverable recoverable) Resumes a recoverable stream consistently at the point indicated by the given ResumeRecoverable.
RecoverableFsDataOutputStream.Committer	recoverForCommit(RecoverableWriter.CommitRecoverable recoverable) Recovers a recoverable stream consistently at the point indicated by the given CommitRecoverable for finalizing and committing.
boolean	requiresCleanupOfRecoverableState() Marks if the writer requires to do any additional cleanup/freeing of resources occupied as part of a RecoverableWriter.ResumeRecoverable, e.g. temporarily files created or objects uploaded to external systems.
boolean	supportsResume() Checks whether the writer and its streams support resuming (appending to) files after recovery (via the RecoverableWriter.recover(ResumeRecoverable) method).
static S3RecoverableWriter	writer(org.apache.hadoop.fs.FileSystem fs, FunctionWithException<File,RefCountedFileWithStream,IOException> tempFileCreator, S3AccessHelper s3AccessHelper, Executor uploadThreadPool, long userDefinedMinPartSize, int maxConcurrentUploadsPerStream)

Methods inherited from class java.lang.Object

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

Method Detail

open

public RecoverableFsDataOutputStream open(Path path)
                                   throws IOException

Description copied from interface: RecoverableWriter

Opens a new recoverable stream to write to the given path. Whether existing files will be overwritten is implementation specific and should not be relied upon.

Specified by:

open in interface RecoverableWriter

Parameters:

path - The path of the file/object to write to.

Returns:

A new RecoverableFsDataOutputStream writing a new file/object.

Throws:

IOException - Thrown if the stream could not be opened/initialized.

recoverForCommit

public RecoverableFsDataOutputStream.Committer recoverForCommit(RecoverableWriter.CommitRecoverable recoverable)
                                                         throws IOException

Description copied from interface: RecoverableWriter

Recovers a recoverable stream consistently at the point indicated by the given CommitRecoverable for finalizing and committing. This will publish the target file with exactly the data that was written up to the point then the CommitRecoverable was created.

Specified by:

recoverForCommit in interface RecoverableWriter

Parameters:

recoverable - The opaque handle with the recovery information.

Returns:

A committer that publishes the target file.

Throws:

IOException - Thrown, if recovery fails.

recover

public S3RecoverableFsDataOutputStream recover(RecoverableWriter.ResumeRecoverable recoverable)
                                        throws IOException

Description copied from interface: RecoverableWriter

Resumes a recoverable stream consistently at the point indicated by the given ResumeRecoverable. Future writes to the stream will continue / append the file as of that point.

This method is optional and whether it is supported is indicated through the RecoverableWriter.supportsResume() method.

Specified by:

recover in interface RecoverableWriter

Parameters:

recoverable - The opaque handle with the recovery information.

Returns:

A recoverable stream writing to the file/object as it was at the point when the ResumeRecoverable was created.

Throws:

IOException - Thrown, if resuming fails.

requiresCleanupOfRecoverableState

public boolean requiresCleanupOfRecoverableState()

Description copied from interface: RecoverableWriter

Marks if the writer requires to do any additional cleanup/freeing of resources occupied as part of a RecoverableWriter.ResumeRecoverable, e.g. temporarily files created or objects uploaded to external systems.

In case cleanup is required, then RecoverableWriter.cleanupRecoverableState(ResumeRecoverable) should be called.

Specified by:

requiresCleanupOfRecoverableState in interface RecoverableWriter

Returns:

true if cleanup is required, false otherwise.

cleanupRecoverableState

public boolean cleanupRecoverableState(RecoverableWriter.ResumeRecoverable resumable)
                                throws IOException

Description copied from interface: RecoverableWriter

Frees up any resources that were previously occupied in order to be able to recover from a (potential) failure. These can be temporary files that were written to the filesystem or objects that were uploaded to S3.

NOTE: This operation should not throw an exception, but return false if the cleanup did not happen for any reason.

Specified by:

cleanupRecoverableState in interface RecoverableWriter

Parameters:

resumable - The RecoverableWriter.ResumeRecoverable whose state we want to clean-up.

Returns:

true if the resources were successfully freed, false otherwise (e.g. the file to be deleted was not there for any reason - already deleted or never created).

Throws:

IOException

getCommitRecoverableSerializer

public SimpleVersionedSerializer<RecoverableWriter.CommitRecoverable> getCommitRecoverableSerializer()

Description copied from interface: RecoverableWriter

The serializer for the CommitRecoverable types created in this writer. This serializer should be used to store the CommitRecoverable in checkpoint state or other forms of persistent state.

Specified by:

getCommitRecoverableSerializer in interface RecoverableWriter

getResumeRecoverableSerializer

public SimpleVersionedSerializer<RecoverableWriter.ResumeRecoverable> getResumeRecoverableSerializer()

Description copied from interface: RecoverableWriter

The serializer for the ResumeRecoverable types created in this writer. This serializer should be used to store the ResumeRecoverable in checkpoint state or other forms of persistent state.

Specified by:

getResumeRecoverableSerializer in interface RecoverableWriter

supportsResume

public boolean supportsResume()

Description copied from interface: RecoverableWriter

Checks whether the writer and its streams support resuming (appending to) files after recovery (via the RecoverableWriter.recover(ResumeRecoverable) method).

If true, then this writer supports the RecoverableWriter.recover(ResumeRecoverable) method. If false, then that method may not be supported and streams can only be recovered via RecoverableWriter.recoverForCommit(CommitRecoverable).

Specified by:

supportsResume in interface RecoverableWriter

writer

public static S3RecoverableWriter writer(org.apache.hadoop.fs.FileSystem fs,
                                         FunctionWithException<File,RefCountedFileWithStream,IOException> tempFileCreator,
                                         S3AccessHelper s3AccessHelper,
                                         Executor uploadThreadPool,
                                         long userDefinedMinPartSize,
                                         int maxConcurrentUploadsPerStream)