@Override
public void sink( FlowProcess<JobConf> flowProcess, SinkCall<Object[], OutputCollector> sinkCall ) throws IOException {
// it's ok to use NULL here so the collector does not write anything
TupleEntry tupleEntry = sinkCall.getOutgoingEntry();
OutputCollector outputCollector = sinkCall.getOutput();
if( updateBy != null )
{
Tuple allValues = tupleEntry.selectTuple( updateValueFields );
Tuple updateValues = tupleEntry.selectTuple( updateByFields );
allValues = cleanTuple( allValues );
TupleRecord key = new TupleRecord( allValues );
if( updateValues.equals( updateIfTuple ) )
outputCollector.collect( key, null );
else
outputCollector.collect( key, key );
return;
}
Tuple result = tupleEntry.selectTuple( getSinkFields() );
result = cleanTuple( result );
outputCollector.collect( new TupleRecord( result ), null );
}