public Runner run() throws Exception {
		// Set up the execution environment
		env = ExecutionEnvironment.getExecutionEnvironment();
		ExecutionConfig config = env.getConfig();

		// should not have any non-Flink data types
		config.disableForceAvro();
		config.disableForceKryo();

		config.setGlobalJobParameters(parameters);
		parameterize(this);

		// configure local parameters and throw proper exception on error
		try {
			this.configure(parameters);
		} catch (RuntimeException ex) {
			throw new ProgramParametrizationException(ex.getMessage());
		}

		// integration tests run with with object reuse both disabled and enabled
		if (disableObjectReuse.getValue()) {
			config.disableObjectReuse();
		} else {
			config.enableObjectReuse();
		}

		// ----------------------------------------------------------------------------------------
		// Usage and configuration
		// ----------------------------------------------------------------------------------------

		// algorithm and usage
		if (!parameters.has(ALGORITHM)) {
			throw new ProgramParametrizationException(getAlgorithmsListing());
		}

		String algorithmName = parameters.get(ALGORITHM);
		algorithm = driverFactory.get(algorithmName);

		if (algorithm == null) {
			throw new ProgramParametrizationException("Unknown algorithm name: " + algorithmName);
		}

		// input and usage
		if (!parameters.has(INPUT)) {
			if (!parameters.has(OUTPUT)) {
				// if neither input nor output is given then print algorithm usage
				throw new ProgramParametrizationException(getAlgorithmUsage(algorithmName));
			}
			throw new ProgramParametrizationException("No input given");
		}

		parameterize(algorithm);

		String inputName = parameters.get(INPUT);
		Input input = inputFactory.get(inputName);

		if (input == null) {
			throw new ProgramParametrizationException("Unknown input type: " + inputName);
		}

		parameterize(input);

		// output and usage
		if (!parameters.has(OUTPUT)) {
			throw new ProgramParametrizationException("No output given");
		}

		String outputName = parameters.get(OUTPUT);
		output = outputFactory.get(outputName);

		if (output == null) {
			throw new ProgramParametrizationException("Unknown output type: " + outputName);
		}

		parameterize(output);

		// ----------------------------------------------------------------------------------------
		// Create list of input and algorithm transforms
		// ----------------------------------------------------------------------------------------

		List<Transform> transforms = new ArrayList<>();

		if (input instanceof Transformable) {
			transforms.addAll(((Transformable) input).getTransformers());
		}

		if (algorithm instanceof Transformable) {
			transforms.addAll(((Transformable) algorithm).getTransformers());
		}

		for (Transform transform : transforms) {
			parameterize(transform);
		}

		// unused parameters
		if (parameters.getUnrequestedParameters().size() > 0) {
			throw new ProgramParametrizationException("Unrequested parameters: " + parameters.getUnrequestedParameters());
		}

		// ----------------------------------------------------------------------------------------
		// Execute
		// ----------------------------------------------------------------------------------------

		// Create input
		Graph graph = input.create(env);

		// Transform input
		for (Transform transform : transforms) {
			graph = (Graph) transform.transformInput(graph);
		}

		// Run algorithm
		result = algorithm.plan(graph);

		// Output
		executionName = jobName.getValue() != null ? jobName.getValue() + ": " : "";

		executionName += input.getIdentity() + " ⇨ " + algorithmName + " ⇨ " + output.getName();

		if (transforms.size() > 0) {
			// append identifiers to job name
			StringBuffer buffer = new StringBuffer(executionName).append(" [");

			for (Transform transform : transforms) {
				buffer.append(transform.getIdentity());
			}

			executionName = buffer.append("]").toString();
		}

		if (output == null) {
			throw new ProgramParametrizationException("Unknown output type: " + outputName);
		}

		try {
			output.configure(parameters);
		} catch (RuntimeException ex) {
			throw new ProgramParametrizationException(ex.getMessage());
		}

		if (result != null) {
			// Transform output if algorithm returned result DataSet
			if (transforms.size() > 0) {
				Collections.reverse(transforms);
				for (Transform transform : transforms) {
					result = (DataSet) transform.transformResult(result);
				}
			}
		}

		return this;
	}
public Runner run ( ) throws Exception { env = ExecutionEnvironment . getExecutionEnvironment ( ) ; ExecutionConfig config = env . getConfig ( ) ; config . disableForceAvro ( ) ; config . disableForceKryo ( ) ; config . setGlobalJobParameters ( parameters ) ; parameterize ( this ) ; try { this . configure ( parameters ) ; } catch ( RuntimeException ex ) { throw new ProgramParametrizationException ( ex . getMessage ( ) ) ; } if ( disableObjectReuse . getValue ( ) ) { config . disableObjectReuse ( ) ; } else { config . enableObjectReuse ( ) ; } if ( ! parameters . has ( ALGORITHM ) ) { throw new ProgramParametrizationException ( getAlgorithmsListing ( ) ) ; } String algorithmName = parameters . get ( ALGORITHM ) ; algorithm = driverFactory . get ( algorithmName ) ; if ( algorithm == null ) { throw new ProgramParametrizationException ( "Unknown algorithm name: " + algorithmName ) ; } if ( ! parameters . has ( INPUT ) ) { if ( ! parameters . has ( OUTPUT ) ) { throw new ProgramParametrizationException ( getAlgorithmUsage ( algorithmName ) ) ; } throw new ProgramParametrizationException ( "No input given" ) ; } parameterize ( algorithm ) ; String inputName = parameters . get ( INPUT ) ; Input input = inputFactory . get ( inputName ) ; if ( input == null ) { throw new ProgramParametrizationException ( "Unknown input type: " + inputName ) ; } parameterize ( input ) ; if ( ! parameters . has ( OUTPUT ) ) { throw new ProgramParametrizationException ( "No output given" ) ; } String outputName = parameters . get ( OUTPUT ) ; output = outputFactory . get ( outputName ) ; if ( output == null ) { throw new ProgramParametrizationException ( "Unknown output type: " + outputName ) ; } parameterize ( output ) ; List < Transform > transforms = new ArrayList < > ( ) ; if ( input instanceof Transformable ) { transforms . addAll ( ( ( Transformable ) input ) . getTransformers ( ) ) ; } if ( algorithm instanceof Transformable ) { transforms . addAll ( ( ( Transformable ) algorithm ) . getTransformers ( ) ) ; } for ( Transform transform : transforms ) { parameterize ( transform ) ; } if ( parameters . getUnrequestedParameters ( ) . size ( ) > 0 ) { throw new ProgramParametrizationException ( "Unrequested parameters: " + parameters . getUnrequestedParameters ( ) ) ; } Graph graph = input . create ( env ) ; for ( Transform transform : transforms ) { graph = ( Graph ) transform . transformInput ( graph ) ; } result = algorithm . plan ( graph ) ; executionName = jobName . getValue ( ) != null ? jobName . getValue ( ) + ": " : "" ; executionName += input . getIdentity ( ) + " ⇨ " + algorithmName + " ⇨ " + output . getName ( ) ; if ( transforms . size ( ) > 0 ) { StringBuffer buffer = new StringBuffer ( executionName ) . append ( " [" ) ; for ( Transform transform : transforms ) { buffer . append ( transform . getIdentity ( ) ) ; } executionName = buffer . append ( "]" ) . toString ( ) ; } if ( output == null ) { throw new ProgramParametrizationException ( "Unknown output type: " + outputName ) ; } try { output . configure ( parameters ) ; } catch ( RuntimeException ex ) { throw new ProgramParametrizationException ( ex . getMessage ( ) ) ; } if ( result != null ) { if ( transforms . size ( ) > 0 ) { Collections . reverse ( transforms ) ; for ( Transform transform : transforms ) { result = ( DataSet ) transform . transformResult ( result ) ; } } } return this ; }
Setup the Flink job with the graph input, algorithm, and output.

<p>To then execute the job call {@link #execute}.

@return this
@throws Exception on error