code: protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig) throws Exception {
  		@SuppressWarnings("unchecked")
  		InputFormat<OUT, InputSplit> inputFormat = (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
  		//configure the input format
  		inputFormat.configure(this.parameters);
  
  		//open the input format
  		if (inputFormat instanceof RichInputFormat) {
  			((RichInputFormat) inputFormat).setRuntimeContext(ctx);
  			((RichInputFormat) inputFormat).openInputFormat();
  		}
  
  		List<OUT> result = new ArrayList<OUT>();
  		
  		// splits
  		InputSplit[] splits = inputFormat.createInputSplits(1);
  		TypeSerializer<OUT> serializer = getOperatorInfo().getOutputType().createSerializer(executionConfig);
  		
  		for (InputSplit split : splits) {
  			inputFormat.open(split);
  			
  			while (!inputFormat.reachedEnd()) {
  				OUT next = inputFormat.nextRecord(serializer.createInstance());
  				if (next != null) {
  					result.add(serializer.copy(next));
  				}
  			}
  			
  			inputFormat.close();
  		}
  		
  		//close the input format
  		if (inputFormat instanceof RichInputFormat) {
  			((RichInputFormat) inputFormat).closeInputFormat();
  		}
  
  		return result;
  	}
docs: --------------------------------------------------------------------------------------------