...

...

...

...

...

...

Please keep the discussion on the mailing list rather than commenting on the wiki.

...

This document proposes the introduction of the Executor interface. This interface will abstract the execution/submission logic from the environments. The executor will take as argument an abstract representation of the user’s job and a configuration object containing the parameters that are relevant for that specific execution of that specific job and, for now, return a JobExecutionResult, as shown below:

...

  JobExecutionResult execute(Pipeline pipeline, Configuration executionConfig) throws Exception;

}

This signature is similar to that of the execute() method of the ExecutionEnvironment and StreamExecutionEnvironment with the difference that it takes a Pipeline as an argument. This is not coincidental.

...

Flink will use Service Discovery to discover the available factories. Then, based on the provided Configuration, it will decide which executor to use. . This configuration object will contain all the relevant parameters needed to instantiate the required Executor.

In terms of code, the above means that, for example, the execute() method of the ExecutionEnvironment will become a concrete method with the following body:

...

For the ExecutorFactory we were considering an alternative interface where the isCompatibleWith(Configuration config) is replaced by the String getExecutorName(). The downside with the approach we advocate here is that the user has to provide the configuration twice, once in the isCompatibleWith() and once in the getExecutor(). On the flip-side, the plus is that we have more flexibility to change the "matching" logic between configuration parameters and executors in the future, as this is hidden from the user. For example, in a first version we may want to have one executor for session-cluster deployment, but in the future we may want to split it in yarn-session, standalone-session, etc.

User-Facing Implications

With So far, the above changes introduce no user-facing changes. But with the transition to a configuration-based execution process, where all possible execution parameters will have matching configuration parameters, one can imagine that now the main() of a user’s program can include everything a user needs to specify in order to run his job on any target environment. To make this more explicit, we are also planning to add a method getExecutionEnvironment(Configuration config) in the available Environments.

...