code: public JobExecutionResult execute(Plan program) throws Exception {
  		long startTime = System.currentTimeMillis();
  
  		initCache(program.getCachedFiles());
  		Collection<? extends GenericDataSinkBase<?>> sinks = program.getDataSinks();
  		for (Operator<?> sink : sinks) {
  			execute(sink);
  		}
  		
  		long endTime = System.currentTimeMillis();
  		Map<String, OptionalFailure<Object>> accumulatorResults = AccumulatorHelper.toResultMap(accumulators);
  		return new JobExecutionResult(null, endTime - startTime, accumulatorResults);
  	}
docs: --------------------------------------------------------------------------------------------