code: public JavaPlan createProgramPlan(String jobName) {
  		if (this.sinks.isEmpty()) {
  			throw new RuntimeException("No data sinks have been created yet. A program needs at least one sink that consumes data. Examples are writing the data set or printing it.");
  		}
  		
  		if (jobName == null) {
  			jobName = getDefaultName();
  		}
  		
  		OperatorTranslation translator = new OperatorTranslation();
  		JavaPlan plan = translator.translateToPlan(this.sinks, jobName);
  		
  		if (getDegreeOfParallelism() > 0) {
  			plan.setDefaultParallelism(getDegreeOfParallelism());
  		}
  		
  		try {
  			registerCachedFilesWithPlan(plan);
  		} catch (Exception e) {
  			throw new RuntimeException("Error while registering cached files: " + e.getMessage(), e);
  		}
  		
  		return plan;
  	}
docs: Creates the program's {@link Plan}. The plan is a description of all data sources, data sinks,
  and operations and how they interact, as an isolated unit that can be executed with a
  {@link eu.stratosphere.api.common.PlanExecutor}. Obtaining a plan and starting it with an
  executor is an alternative way to run a program and is only possible if the program consists
  only of distributed operations.
  
  @param jobName The name attached to the plan (displayed in logs and monitoring).
  @return The program's plan.