org.apache.flink.runtime.executiongraph

Class ExecutionGraph

java.lang.Object

org.apache.flink.runtime.executiongraph.ExecutionGraph

All Implemented Interfaces:

Serializable

public class ExecutionGraph
extends Object
implements Serializable

The execution graph is the central data structure that coordinates the distributed execution of a data flow. It keeps representations of each parallel task, each intermediate result, and the communication between them. The execution graph consists of the following constructs:

• The ExecutionJobVertex represents one vertex from the JobGraph (usually one operation like "map" or "join") during execution. It holds the aggregated state of all parallel subtasks. The ExecutionJobVertex is identified inside the graph by the JobVertexID, which it takes from the JobGraph's corresponding JobVertex.
• The ExecutionVertex represents one parallel subtask. For each ExecutionJobVertex, there are as many ExecutionVertices as the parallelism. The ExecutionVertex is identified by the ExecutionJobVertex and the number of the parallel subtask
• The Execution is one attempt to execute a ExecutionVertex. There may be multiple Executions for the ExecutionVertex, in case of a failure, or in the case where some data needs to be recomputed because it is no longer available when requested by later operations. An Execution is always identified by an ExecutionAttemptID. All messages between the JobManager and the TaskManager about deployment of tasks and updates in the task status always use the ExecutionAttemptID to address the message receiver.

The ExecutionGraph implements Serializable, because it can be archived by sending it to an archive actor via an actor message. The execution graph does contain some non-serializable fields. These fields are not required in the archived form and are cleared in the prepareForArchiving() method.

See Also:

Serialized Form

Constructor Summary

Constructors

Constructor and Description
ExecutionGraph(scala.concurrent.ExecutionContext executionContext, JobID jobId, String jobName, Configuration jobConfig, scala.concurrent.duration.FiniteDuration timeout, RestartStrategy restartStrategy, List<BlobKey> requiredJarFiles, List<URL> requiredClasspaths, ClassLoader userClassLoader)

Method Summary

Modifier and Type	Method and Description
Map<String,Accumulator<?,?>>	aggregateUserAccumulators() Merges all accumulator results from the tasks previously executed in the Executions.
void	attachJobGraph(List<JobVertex> topologiallySorted)
void	cancel()
void	disableSnaphotCheckpointing() Disables checkpointing.
void	enableSnapshotCheckpointing(long interval, long checkpointTimeout, long minPauseBetweenCheckpoints, int maxConcurrentCheckpoints, List<ExecutionJobVertex> verticesToTrigger, List<ExecutionJobVertex> verticesToWaitFor, List<ExecutionJobVertex> verticesToCommitTo, akka.actor.ActorSystem actorSystem, UUID leaderSessionID, CheckpointIDCounter checkpointIDCounter, CompletedCheckpointStore completedCheckpointStore, RecoveryMode recoveryMode, StateStore<CompletedCheckpoint> savepointStore, CheckpointStatsTracker statsTracker)
void	fail(Throwable t)
StringifiedAccumulatorResult[]	getAccumulatorResultsStringified() Returns the a stringified version of the user-defined accumulators.
Map<String,SerializedValue<Object>>	getAccumulatorsSerialized() Gets a serialized accumulator map.
Iterable<ExecutionVertex>	getAllExecutionVertices()
Map<IntermediateDataSetID,IntermediateResult>	getAllIntermediateResults()
Map<JobVertexID,ExecutionJobVertex>	getAllVertices()
CheckpointCoordinator	getCheckpointCoordinator()
CheckpointStatsTracker	getCheckpointStatsTracker()
ExecutionConfig	getExecutionConfig()
scala.concurrent.ExecutionContext	getExecutionContext() Returns the ExecutionContext associated with this ExecutionGraph.
Throwable	getFailureCause()
Map<ExecutionAttemptID,Map<AccumulatorRegistry.Metric,Accumulator<?,?>>>	getFlinkAccumulators() Gets the internal flink accumulator map of maps which contains some metrics.
Configuration	getJobConfiguration()
JobID	getJobID()
String	getJobName()
ExecutionJobVertex	getJobVertex(JobVertexID id)
String	getJsonPlan()
int	getNumberOfExecutionJobVertices() Gets the number of job vertices currently held by this execution graph.
Map<ExecutionAttemptID,Execution>	getRegisteredExecutions()
List<URL>	getRequiredClasspaths() Returns a list of classpaths referring to the directories/JAR files required to run this job
List<BlobKey>	getRequiredJarFiles() Returns a list of BLOB keys referring to the JAR files required to run this job
RestartStrategy	getRestartStrategy()
SavepointCoordinator	getSavepointCoordinator()
ScheduleMode	getScheduleMode()
Scheduler	getScheduler()
JobStatus	getState()
long	getStatusTimestamp(JobStatus status)
ClassLoader	getUserClassLoader()
Iterable<ExecutionJobVertex>	getVerticesTopologically()
boolean	isArchived()
boolean	isQueuedSchedulingAllowed()
boolean	isStoppable()
void	prepareForArchiving() This method cleans fields that are irrelevant for the archived execution attempt.
void	registerExecutionListener(ActorGateway listener)
void	registerJobStatusListener(ActorGateway listener)
void	restart()
void	restoreLatestCheckpointedState() Restores the latest checkpointed state.
void	restoreSavepoint(String savepointPath) Restores the execution state back to a savepoint.
void	scheduleForExecution(Scheduler scheduler)
void	scheduleOrUpdateConsumers(ResultPartitionID partitionId)
void	setJsonPlan(String jsonPlan)
void	setQueuedSchedulingAllowed(boolean allowed)
void	setScheduleMode(ScheduleMode scheduleMode)
void	stop()
void	updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) Updates the accumulators during the runtime of a job.
boolean	updateState(TaskExecutionState state) Updates the state of one of the ExecutionVertex's Execution attempts.
void	waitUntilFinished() For testing: This waits until the job execution has finished.

Methods inherited from class java.lang.Object

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

Constructor Detail

ExecutionGraph

public ExecutionGraph(scala.concurrent.ExecutionContext executionContext,
                      JobID jobId,
                      String jobName,
                      Configuration jobConfig,
                      scala.concurrent.duration.FiniteDuration timeout,
                      RestartStrategy restartStrategy,
                      List<BlobKey> requiredJarFiles,
                      List<URL> requiredClasspaths,
                      ClassLoader userClassLoader)

Method Detail

getNumberOfExecutionJobVertices

public int getNumberOfExecutionJobVertices()

Gets the number of job vertices currently held by this execution graph.

Returns:

The current number of job vertices.

isQueuedSchedulingAllowed

public boolean isQueuedSchedulingAllowed()

setQueuedSchedulingAllowed

public void setQueuedSchedulingAllowed(boolean allowed)

setScheduleMode

public void setScheduleMode(ScheduleMode scheduleMode)

getScheduleMode

public ScheduleMode getScheduleMode()

isArchived

public boolean isArchived()

enableSnapshotCheckpointing

public void enableSnapshotCheckpointing(long interval,
                                        long checkpointTimeout,
                                        long minPauseBetweenCheckpoints,
                                        int maxConcurrentCheckpoints,
                                        List<ExecutionJobVertex> verticesToTrigger,
                                        List<ExecutionJobVertex> verticesToWaitFor,
                                        List<ExecutionJobVertex> verticesToCommitTo,
                                        akka.actor.ActorSystem actorSystem,
                                        UUID leaderSessionID,
                                        CheckpointIDCounter checkpointIDCounter,
                                        CompletedCheckpointStore completedCheckpointStore,
                                        RecoveryMode recoveryMode,
                                        StateStore<CompletedCheckpoint> savepointStore,
                                        CheckpointStatsTracker statsTracker)
                                 throws Exception

Throws:

Exception

disableSnaphotCheckpointing

public void disableSnaphotCheckpointing()
                                 throws Exception

Disables checkpointing.

The shutdown of the checkpoint coordinator might block. Make sure that calls to this method don't block the job manager actor and run asynchronously.

Throws:

Exception

getCheckpointCoordinator

public CheckpointCoordinator getCheckpointCoordinator()

getSavepointCoordinator

public SavepointCoordinator getSavepointCoordinator()

getRestartStrategy

public RestartStrategy getRestartStrategy()

getCheckpointStatsTracker

public CheckpointStatsTracker getCheckpointStatsTracker()

getRequiredJarFiles

public List<BlobKey> getRequiredJarFiles()

Returns a list of BLOB keys referring to the JAR files required to run this job

Returns:

list of BLOB keys referring to the JAR files required to run this job

getRequiredClasspaths

public List<URL> getRequiredClasspaths()

Returns a list of classpaths referring to the directories/JAR files required to run this job

Returns:

list of classpaths referring to the directories/JAR files required to run this job

setJsonPlan

public void setJsonPlan(String jsonPlan)

getJsonPlan

public String getJsonPlan()

getScheduler

public Scheduler getScheduler()

getJobID

public JobID getJobID()

getJobName

public String getJobName()

isStoppable

public boolean isStoppable()

getJobConfiguration

public Configuration getJobConfiguration()

getUserClassLoader

public ClassLoader getUserClassLoader()

getState

public JobStatus getState()

getFailureCause

public Throwable getFailureCause()

getJobVertex

public ExecutionJobVertex getJobVertex(JobVertexID id)

getAllVertices

public Map<JobVertexID,ExecutionJobVertex> getAllVertices()

getVerticesTopologically

public Iterable<ExecutionJobVertex> getVerticesTopologically()

getAllIntermediateResults

public Map<IntermediateDataSetID,IntermediateResult> getAllIntermediateResults()

getAllExecutionVertices

public Iterable<ExecutionVertex> getAllExecutionVertices()

getStatusTimestamp

public long getStatusTimestamp(JobStatus status)

getExecutionContext

public scala.concurrent.ExecutionContext getExecutionContext()

Returns the ExecutionContext associated with this ExecutionGraph.

Returns:

ExecutionContext associated with this ExecutionGraph

getFlinkAccumulators

public Map<ExecutionAttemptID,Map<AccumulatorRegistry.Metric,Accumulator<?,?>>> getFlinkAccumulators()

Gets the internal flink accumulator map of maps which contains some metrics.

Returns:

A map of accumulators for every executed task.

aggregateUserAccumulators

public Map<String,Accumulator<?,?>> aggregateUserAccumulators()

Merges all accumulator results from the tasks previously executed in the Executions.

Returns:

The accumulator map

getAccumulatorsSerialized

public Map<String,SerializedValue<Object>> getAccumulatorsSerialized()
                                                              throws IOException

Gets a serialized accumulator map.

Returns:

The accumulator map with serialized accumulator values.

Throws:

IOException

getAccumulatorResultsStringified

public StringifiedAccumulatorResult[] getAccumulatorResultsStringified()

Returns the a stringified version of the user-defined accumulators.

Returns:

an Array containing the StringifiedAccumulatorResult objects

attachJobGraph

public void attachJobGraph(List<JobVertex> topologiallySorted)
                    throws JobException

Throws:

JobException

scheduleForExecution

public void scheduleForExecution(Scheduler scheduler)
                          throws JobException

Throws:

JobException

cancel

public void cancel()

stop

public void stop()
          throws StoppingException

Throws:

StoppingException

fail

public void fail(Throwable t)

restart

public void restart()

restoreLatestCheckpointedState

public void restoreLatestCheckpointedState()
                                    throws Exception

Restores the latest checkpointed state.

The recovery of checkpoints might block. Make sure that calls to this method don't block the job manager actor and run asynchronously.

Throws:

Exception

restoreSavepoint

public void restoreSavepoint(String savepointPath)
                      throws Exception

Restores the execution state back to a savepoint.

The execution vertices need to be in state ExecutionState.CREATED when calling this method. The operation might block. Make sure that calls don't block the job manager actor.

Parameters:

savepointPath - The path of the savepoint to rollback to.

Throws:

IllegalStateException - If checkpointing is disabled

IllegalStateException - If checkpoint coordinator is shut down

Exception - If failure during rollback

prepareForArchiving

public void prepareForArchiving()

This method cleans fields that are irrelevant for the archived execution attempt.

getExecutionConfig

public ExecutionConfig getExecutionConfig()

waitUntilFinished

public void waitUntilFinished()
                       throws InterruptedException

For testing: This waits until the job execution has finished.

Throws:

InterruptedException

updateState

public boolean updateState(TaskExecutionState state)

Updates the state of one of the ExecutionVertex's Execution attempts. If the new status if "FINISHED", this also updates the

Parameters:

state - The state update.

Returns:

True, if the task update was properly applied, false, if the execution attempt was not found.

scheduleOrUpdateConsumers

public void scheduleOrUpdateConsumers(ResultPartitionID partitionId)

getRegisteredExecutions

public Map<ExecutionAttemptID,Execution> getRegisteredExecutions()

updateAccumulators

public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot)

Updates the accumulators during the runtime of a job. Final accumulator results are transferred through the UpdateTaskExecutionState message.

Parameters:

accumulatorSnapshot - The serialized flink and user-defined accumulators

registerJobStatusListener

public void registerJobStatusListener(ActorGateway listener)

registerExecutionListener

public void registerExecutionListener(ActorGateway listener)