[FLINK-2097] Implement job session management #858

mxm · 2015-06-22T14:31:23Z

This is a joint effort by @StephanEwen and me to introduce a session management in Flink. Session are used to keep a copy of the ExecutionGraph in the job manager for the session lifetime. It is important that the ExecutionGraph is not kept around longer because it consumes a lot of memory. Its intermediate results can also be freed. To integrate sessions properly into Flink, some refactoring was necessary. In particular these are:

JobId is created through the ExecutionEnvironment and passed through
Sessions can be termined by the ExecutionEnvironment or directly through the executor
Session are cancelled implicitly through "reapers" or shutdown hooks in the ExecutionEnvironment, otherwise they time out
LocalExecutor and RemoteExecutor manage sessions
The Client only deals with the communication with the job manager and is agnostic of session management

With the session management, we will be able to properly support backtracking of produced intermediate results. This makes calls to count()/collect()/print() efficient and enables to write incremental/interactive jobs.

uce · 2015-07-22T13:42:50Z

I just had a look at the JobManager in a different context and thought about the following, which might be relevant here: when submitting a new JobGraph, which is attached to an existing ExecutionGraph, some ExecutionGraph state is overwritten by the new JobGraph. With some you might run into (maybe) unexpected behaviour like resetting number of left execution retries or creating a new CheckpointCoordinator for the ExecutionGraph.

What's the intended behaviour of attaching to an existing ExecutionGraph? Is there an implicit assumption that the existing ExecutionGraph needs to be finished already?

StephanEwen · 2015-07-22T15:13:49Z

I think right now, it pretty much behaves as if someone started a new job, with the "grown" execution graph.

mxm · 2015-07-22T15:18:49Z

It should just add more nodes to the ExecutionGraph. Existing ones should not be modified. For batch, I think the assumption is that it needs to be finished. For streaming, I could also picture attaching nodes at runtime but this has to be carefully implemented..

Sessions make sure that the JobManager does not immediately discard a JobGraph after execution, but keeps it around for further operations to be attached to the graph. That is the basis for interactive sessions. This pull request implements a rudimentary session management. Together with the backtracking apache#640, this will enable users to submit jobs to the cluster and access intermediate results. Session handling ensures that the results are cleared eventually. ExecutionGraphs are kept as long as - no timeout occurred or - the session has not been explicitly ended The following changes have also been made in this pull request: - The Job ID is created through the ExecutionEnvironment and passed through - Sessions can be termined by the ExecutionEnvironment or directly through the executor - The environments use reapers (local) and shutdown hooks (remote) to ensure session termination when the environment runs out of scope - The Client manages only connections to the JobManager, it is not job specific This closes apache#858.

mxm · 2015-09-08T16:40:45Z

I've ported this pull request to the latest master. It was a lot more work than I anticipated because some classes had diverged significantly and merging them was a bit hard.

Due to some refactoring, the changes have grown quite large again and I know that makes reviewing hard. Despite that, I wouldn't delay merging this pull request much further. We can disable the session management until it is integrated with the rest of the system (intermediate results) by throwing an exception on the interface methods. If we decide later, that we want to delay this feature, we could also remove the session code. In that case, it would still make sense to merge this pull request because it contains a lot of nice refactoring.

With the session management in place, we can reuse already computed intermediate results with not too much effort. Actually, only some API changes are remaining to expose the session management to the user in production.

rmetzger · 2015-09-08T17:03:32Z

...sts/src/test/scala/org/apache/flink/api/scala/runtime/jobmanager/JobManagerFailsITCase.scala

- cluster.stop()
- }
- }
+// "detect a lost connection to the JobManager and try to reconnect to it" in {


Re-enable the test?

Sessions make sure that the JobManager does not immediately discard a JobGraph after execution, but keeps it around for further operations to be attached to the graph. That is the basis for interactive sessions. This pull request implements a rudimentary session management. Together with the backtracking apache#640, this will enable users to submit jobs to the cluster and access intermediate results. Session handling ensures that the results are cleared eventually. ExecutionGraphs are kept as long as - no timeout occurred or - the session has not been explicitly ended The following changes have also been made in this pull request: - The Job ID is created through the ExecutionEnvironment and passed through - Sessions can be termined by the ExecutionEnvironment or directly through the executor - The environments use reapers (local) and shutdown hooks (remote) to ensure session termination when the environment runs out of scope - The Client manages only connections to the JobManager, it is not job specific This closes apache#858.

tillrohrmann · 2015-09-09T08:32:33Z

Could you elaborate a little bit on what you refactored and which components would be important to review?

Sessions make sure that the JobManager does not immediately discard a JobGraph after execution, but keeps it around for further operations to be attached to the graph. That is the basis for interactive sessions. This pull request implements a rudimentary session management. Together with the backtracking apache#640, this will enable users to submit jobs to the cluster and access intermediate results. Session handling ensures that the results are cleared eventually. ExecutionGraphs are kept as long as - no timeout occurred or - the session has not been explicitly ended The following changes have also been made in this pull request: - The Job ID is created through the ExecutionEnvironment and passed through - Sessions can be termined by the ExecutionEnvironment or directly through the executor - The environments use reapers (local) and shutdown hooks (remote) to ensure session termination when the environment runs out of scope - The Client manages only connections to the JobManager, it is not job specific This closes apache#858.

mxm · 2015-09-09T09:12:06Z

Of course! The following classes have been refactored in the course of integrating them with the session management:

Client

Establish connection to JobManager on creation
Refactor run method into runBlocking and runDetached
Extract helper classes to generate the Plan
Make Optimizer and JobGraph generation methods static
Pass ClassLoader correctly (do not keep one per Client but rather let it be passed before submission)

CliFrontend

runBlocking and runDetached methods by analogy with the Client class

ExecutionEnvironment, LocalEnvironment, RemoteEnvironment

modified abstract class to support sessions (timeout and jobID generation)
handle session management via Reapers and ShutdownHooks

PlanExecutor, LocalExecutor, RemoteExecutor

modified interface
support session termination
set JobID on Plan

JobManager

keep ExecutionGraph as long as session has not expired

Future issues:

Support for sessions in streaming. Currently streaming jobs are agnostic of sessions.
Representation of sessions in the JobManager web frontend. How do we represent updates to the ExecutionGraph in sessions?
Build features on top of session management (e.g. intermediate results)

Sessions make sure that the JobManager does not immediately discard a JobGraph after execution, but keeps it around for further operations to be attached to the graph. That is the basis for interactive sessions. This pull request implements a rudimentary session management. Together with the backtracking apache#640, this will enable users to submit jobs to the cluster and access intermediate results. Session handling ensures that the results are cleared eventually. ExecutionGraphs are kept as long as - no timeout occurred or - the session has not been explicitly ended The following changes have also been made in this pull request: - The Job ID is created through the ExecutionEnvironment and passed through - Sessions can be termined by the ExecutionEnvironment or directly through the executor - The environments use reapers (local) and shutdown hooks (remote) to ensure session termination when the environment runs out of scope - The Client manages only connections to the JobManager, it is not job specific This closes apache#858.

tillrohrmann · 2015-09-09T09:56:24Z

Thanks Max for the detailed description.

On Wed, Sep 9, 2015 at 11:12 AM, Max [email protected] wrote:

Of course! The following classes have been refactored in the course of
integrating them with the session management:

Client

Establish connection to JobManager on creation

Refactor run method into runBlocking and runDetached

Extract helper classes to generate the Plan

Make Optimizer and JobGraph generation methods static

Pass ClassLoader correctly (do not keep one per Client but rather
let it be passed before submission)

CliFrontend

runBlocking and runDetached methods by analogy with the Client class

ExecutionEnvironment, LocalEnvironment, RemoteEnvironment

modified abstract class to support sessions (timeout and jobID
generation)

handle session management via Reapers and ShutdownHooks

PlanExecutor, LocalExecutor, RemoteExecutor

modified interface

support session termination

set JobID on Plan

JobManager

keep ExecutionGraph as long as session has not expired

Future issues:

Support for sessions in streaming. Currently streaming jobs are
agnostic of sessions.

Representation of sessions in the JobManager web frontend. How do we
represent updates to the ExecutionGraph in sessions?

Build features on top of session management (e.g. intermediate
results)

—
Reply to this email directly or view it on GitHub
#858 (comment).

Sessions make sure that the JobManager does not immediately discard a JobGraph after execution, but keeps it around for further operations to be attached to the graph. That is the basis for interactive sessions. This pull request implements a rudimentary session management. Together with the backtracking apache#640, this will enable users to submit jobs to the cluster and access intermediate results. Session handling ensures that the results are cleared eventually. ExecutionGraphs are kept as long as - no timeout occurred or - the session has not been explicitly ended The following changes have also been made in this pull request: - The Job ID is created through the ExecutionEnvironment and passed through - Sessions can be termined by the ExecutionEnvironment or directly through the executor - The environments use reapers (local) and shutdown hooks (remote) to ensure session termination when the environment runs out of scope - The Client manages only connections to the JobManager, it is not job specific This closes apache#858.

mxm · 2015-09-15T14:11:27Z

I've rebased again...If nobody objects, I will merge this soon. The new API-facing methods on ExecutionEnvironment will be disabled until we implement first applications of session management. I've added a separate commit that does that.

Sessions make sure that the JobManager does not immediately discard a JobGraph after execution, but keeps it around for further operations to be attached to the graph. That is the basis for interactive sessions. This pull request implements a rudimentary session management. Together with the backtracking apache#640, this will enable users to submit jobs to the cluster and access intermediate results. Session handling ensures that the results are cleared eventually. ExecutionGraphs are kept as long as - no timeout occurred or - the session has not been explicitly ended The following changes have also been made in this pull request: - The Job ID is created through the ExecutionEnvironment and passed through - Sessions can be termined by the ExecutionEnvironment or directly through the executor - The environments use reapers (local) and shutdown hooks (remote) to ensure session termination when the environment runs out of scope - The Client manages only connections to the JobManager, it is not job specific This closes apache#858.

mxm force-pushed the session-dev branch 3 times, most recently from 783a72a to b43b9f6 Compare June 23, 2015 11:42

mxm mentioned this pull request Jun 23, 2015

implement a simple session management #681

Closed

mxm force-pushed the session-dev branch from b43b9f6 to ceb2d57 Compare June 24, 2015 07:46

StephanEwen mentioned this pull request Sep 3, 2015

[FLINK-2615]Preserve executors for the entire run of program. #1088

Closed

mxm force-pushed the session-dev branch from ceb2d57 to 74e6e12 Compare September 8, 2015 15:49

mxm force-pushed the session-dev branch from 74e6e12 to 8418b10 Compare September 8, 2015 16:31

rmetzger reviewed Sep 8, 2015
View reviewed changes

mxm force-pushed the session-dev branch from 8418b10 to 6c92015 Compare September 9, 2015 08:22

mxm force-pushed the session-dev branch from 6c92015 to 0f2ad87 Compare September 9, 2015 08:26

mxm force-pushed the session-dev branch from 0f2ad87 to 7ea22e9 Compare September 9, 2015 08:33

mxm force-pushed the session-dev branch from 7ea22e9 to 798e734 Compare September 9, 2015 09:17

fhueske mentioned this pull request Sep 9, 2015

[FLINK-1730]Persist operator on Data Sets #1083

Closed

mxm force-pushed the session-dev branch from 798e734 to 2d1985f Compare September 10, 2015 13:41

mxm force-pushed the session-dev branch from 2d1985f to 0c605fc Compare September 10, 2015 15:30

mxm force-pushed the session-dev branch from 0c605fc to 6558a21 Compare September 15, 2015 14:09

mxm force-pushed the session-dev branch from 6558a21 to 1db6248 Compare September 15, 2015 15:32

rmetzger mentioned this pull request Sep 21, 2015

[FLINK-1789] [core] [runtime] [java-api] Allow adding of URLs to the usercode class loader #593

Closed

mxm added 2 commits September 21, 2015 20:53

[FLINK-2097] temporarily disable session management API

57cf958

mxm force-pushed the session-dev branch from 1db6248 to 57cf958 Compare September 21, 2015 21:10

asfgit closed this in 71bf2f5 Sep 22, 2015

rmetzger added the component=Runtime/Coordination label Mar 14, 2019

pnowojski pushed a commit that referenced this pull request Jul 3, 2024

[FRT-468] Adds Async Calc support to QuerySummary (#858)

4c13c88

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-2097] Implement job session management #858

[FLINK-2097] Implement job session management #858

mxm commented Jun 22, 2015

uce commented Jul 22, 2015

StephanEwen commented Jul 22, 2015

mxm commented Jul 22, 2015

mxm commented Sep 8, 2015

rmetzger Sep 8, 2015

mxm Sep 9, 2015

tillrohrmann commented Sep 9, 2015

mxm commented Sep 9, 2015

tillrohrmann commented Sep 9, 2015

mxm commented Sep 15, 2015

[FLINK-2097] Implement job session management #858

[FLINK-2097] Implement job session management #858

Conversation

mxm commented Jun 22, 2015

uce commented Jul 22, 2015

StephanEwen commented Jul 22, 2015

mxm commented Jul 22, 2015

mxm commented Sep 8, 2015

rmetzger Sep 8, 2015

Choose a reason for hiding this comment

mxm Sep 9, 2015

Choose a reason for hiding this comment

tillrohrmann commented Sep 9, 2015

mxm commented Sep 9, 2015

tillrohrmann commented Sep 9, 2015

mxm commented Sep 15, 2015