code: protected GenericDataSinkBase<T> translateToDataFlow(Operator<T> input) {
  		// select the name (or create a default one)
  		String name = this.name != null ? this.name : this.format.toString();
  		GenericDataSinkBase<T> sink = new GenericDataSinkBase<>(this.format, new UnaryOperatorInformation<>(this.type, new NothingTypeInfo()), name);
  		// set input
  		sink.setInput(input);
  		// set parameters
  		if (this.parameters != null) {
  			sink.getParameters().addAll(this.parameters);
  		}
  		// set parallelism
  		if (this.parallelism > 0) {
  			// use specified parallelism
  			sink.setParallelism(this.parallelism);
  		} else {
  			// if no parallelism has been specified, use parallelism of input operator to enable chaining
  			sink.setParallelism(input.getParallelism());
  		}
  
  		if (this.sortKeyPositions != null) {
  			// configure output sorting
  			Ordering ordering = new Ordering();
  			for (int i = 0; i < this.sortKeyPositions.length; i++) {
  				ordering.appendOrdering(this.sortKeyPositions[i], null, this.sortOrders[i]);
  			}
  			sink.setLocalOrder(ordering);
  		}
  
  		return sink;
  	}
docs: --------------------------------------------------------------------------------------------