code: @SuppressWarnings("unchecked")
  	protected void executeOnCollections(List<IN> inputData, RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		OutputFormat<IN> format = this.formatWrapper.getUserCodeObject();
  		TypeInformation<IN> inputType = getInput().getOperatorInfo().getOutputType();
  
  		if (this.localOrdering != null) {
  			int[] sortColumns = this.localOrdering.getFieldPositions();
  			boolean[] sortOrderings = this.localOrdering.getFieldSortDirections();
  
  			final TypeComparator<IN> sortComparator;
  			if (inputType instanceof CompositeType) {
  				sortComparator = ((CompositeType<IN>) inputType).createComparator(sortColumns, sortOrderings, 0, executionConfig);
  			} else if (inputType instanceof AtomicType) {
  				sortComparator = ((AtomicType<IN>) inputType).createComparator(sortOrderings[0], executionConfig);
  			} else {
  				throw new UnsupportedOperationException("Local output sorting does not support type "+inputType+" yet.");
  			}
  
  			Collections.sort(inputData, new Comparator<IN>() {
  				@Override
  				public int compare(IN o1, IN o2) {
  					return sortComparator.compare(o1, o2);
  				}
  			});
  		}
  
  		if(format instanceof InitializeOnMaster) {
  			((InitializeOnMaster)format).initializeGlobal(1);
  		}
  		format.configure(this.parameters);
  
  		if(format instanceof RichOutputFormat){
  			((RichOutputFormat<?>) format).setRuntimeContext(ctx);
  		}
  		format.open(0, 1);
  		for (IN element : inputData) {
  			format.writeRecord(element);
  		}
  		
  		format.close();
  		
  		if(format instanceof FinalizeOnMaster) {
  			((FinalizeOnMaster)format).finalizeGlobal(1);
  		}
  	}
docs: --------------------------------------------------------------------------------------------