CSVAssignRowIDReducer.java 文件源码

@Override
@SuppressWarnings("unchecked")
public void reduce(ByteWritable key, Iterator<OffsetCount> values,
        OutputCollector<ByteWritable, OffsetCount> out, Reporter report)
        throws IOException 
{   
    //need to sort the values by filename and fileoffset
    while(values.hasNext())
        list.add(new OffsetCount(values.next()));
    Collections.sort(list);

    long lineOffset=0;
    for(OffsetCount oc: list)
    {
        long count=oc.count;
        oc.count=lineOffset;
        out.collect(key, oc);
        lineOffset+=count;
    }
    report.incrCounter(CSVReblockMR.NUM_ROWS_IN_MATRIX, key.toString(), lineOffset);
    list.clear();
}