code: @SuppressWarnings("unchecked")
  	void addStreamOperationFunction(String operationName, SerFunction<?,?> function){
  		this.operationNames.add(operationName);
  		this.streamOperationFunction = this.streamOperationFunction != null
  				? this.streamOperationFunction.andThen(function)
  						: function;
  		
  		if (function instanceof KeyValueMappingFunction){
  			if ( ((KeyValueMappingFunction<?,?,?>)function).aggregatesValues() ) {
  				String lastOperationName = this.operationNames.get(this.operationNames.size()-1);
  				lastOperationName = lastOperationName + "{reducingValues}";  
  				this.operationNames.set(this.operationNames.size()-1, lastOperationName);
  			}
  		}
  	}
docs: Will add the given {@link SerFunction} to this {@link DStreamOperation} by
  composing it with the previous function. If previous function is <i>null</i>
  the given function becomes the root function of this operation.<br>
  It also adds the given <i>operationName</i> to the list of operation names
  which composes this {@link DStreamOperation}.<br>
  The final (composed) function represents the function to applied on the
  localized {@link Stream} of data processed by a target task.