org.apache.flink.runtime.blob

Class PermanentBlobCache

java.lang.Object

org.apache.flink.runtime.blob.AbstractBlobCache

org.apache.flink.runtime.blob.PermanentBlobCache

All Implemented Interfaces:

Closeable, AutoCloseable, PermanentBlobService

public class PermanentBlobCache
extends AbstractBlobCache
implements PermanentBlobService

Provides a cache for permanent BLOB files including a per-job ref-counting and a staged cleanup.

When requesting BLOBs via getFile(JobID, PermanentBlobKey), the cache will first attempt to serve the file from its local cache. Only if the local cache does not contain the desired BLOB, it will try to download it from a distributed HA file system (if available) or the BLOB server.

If files for a job are not needed any more, they will enter a staged, i.e. deferred, cleanup. Files may thus still be be accessible upon recovery and do not need to be re-downloaded.

Field Summary

Fields inherited from class org.apache.flink.runtime.blob.AbstractBlobCache

blobClientConfig, blobView, log, numFetchRetries, readWriteLock, serverAddress, shutdownHook, shutdownRequested, storageDir, tempFileCounter

Constructor Summary

Constructors

Constructor and Description
PermanentBlobCache(Configuration blobClientConfig, BlobView blobView, InetSocketAddress serverAddress) Instantiates a new cache for permanent BLOBs which are also available in an HA store.

Method Summary

Modifier and Type	Method and Description
protected void	cancelCleanupTask() Cancels any cleanup task that subclasses may be executing.
File	getFile(JobID jobId, PermanentBlobKey key) Returns the path to a local copy of the file associated with the provided job ID and blob key.
int	getNumberOfReferenceHolders(JobID jobId)
File	getStorageLocation(JobID jobId, BlobKey key) Returns a file handle to the file associated with the given blob key on the blob server.
void	registerJob(JobID jobId) Registers use of job-related BLOBs.
void	releaseJob(JobID jobId) Unregisters use of job-related BLOBs and allow them to be released.

Methods inherited from class org.apache.flink.runtime.blob.AbstractBlobCache

close, getFileInternal, getPort, setBlobServerAddress

Methods inherited from class java.lang.Object

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

Methods inherited from interface java.io.Closeable

close

Constructor Detail

PermanentBlobCache

public PermanentBlobCache(Configuration blobClientConfig,
                          BlobView blobView,
                          @Nullable
                          InetSocketAddress serverAddress)
                   throws IOException

Instantiates a new cache for permanent BLOBs which are also available in an HA store.

Parameters:

blobClientConfig - global configuration

blobView - (distributed) HA blob store file system to retrieve files from first

serverAddress - address of the BlobServer to use for fetching files from or null if none yet

Throws:

IOException - thrown if the (local or distributed) file storage cannot be created or is not usable

Method Detail

registerJob

public void registerJob(JobID jobId)

Registers use of job-related BLOBs.

Using any other method to access BLOBs, e.g. getFile(org.apache.flink.api.common.JobID, org.apache.flink.runtime.blob.PermanentBlobKey), is only valid within calls to registerJob(JobID) and releaseJob(JobID).

Parameters:

jobId - ID of the job this blob belongs to

See Also:

releaseJob(JobID)

releaseJob

public void releaseJob(JobID jobId)

Unregisters use of job-related BLOBs and allow them to be released.

Parameters:

jobId - ID of the job this blob belongs to

See Also:

registerJob(JobID)

getNumberOfReferenceHolders

public int getNumberOfReferenceHolders(JobID jobId)

getFile

public File getFile(JobID jobId,
                    PermanentBlobKey key)
             throws IOException

Returns the path to a local copy of the file associated with the provided job ID and blob key.

We will first attempt to serve the BLOB from the local storage. If the BLOB is not in there, we will try to download it from the HA store, or directly from the BlobServer.

Specified by:

getFile in interface PermanentBlobService

Parameters:

jobId - ID of the job this blob belongs to

key - blob key associated with the requested file

Returns:

The path to the file.

Throws:

FileNotFoundException - if the BLOB does not exist;

IOException - if any other error occurs when retrieving the file

getStorageLocation

@VisibleForTesting
public File getStorageLocation(JobID jobId,
                                                  BlobKey key)
                                           throws IOException

Returns a file handle to the file associated with the given blob key on the blob server.

Parameters:

jobId - ID of the job this blob belongs to (or null if job-unrelated)

key - identifying the file

Returns:

file handle to the file

Throws:

IOException - if creating the directory fails

cancelCleanupTask

protected void cancelCleanupTask()

Description copied from class: AbstractBlobCache

Cancels any cleanup task that subclasses may be executing.

This is called during AbstractBlobCache.close().

Specified by:

cancelCleanupTask in class AbstractBlobCache