package es.tid.cosmos.mobility.outpois;

import java.io.IOException;

import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import es.tid.cosmos.mobility.data.MobProtocol.Poi;
import es.tid.cosmos.mobility.data.MobProtocol.TwoInt;
import es.tid.cosmos.mobility.data.PoiUtil;

/**
 *
 * @author dmicol
 */
public class VectorPoisOutReducer extends Reducer<ProtobufWritable<TwoInt>,
        ProtobufWritable<Poi>, NullWritable, Text> {
    @Override
    protected void reduce(ProtobufWritable<TwoInt> key,
            Iterable<ProtobufWritable<Poi>> values, Context context)
            throws IOException, InterruptedException {
        for (ProtobufWritable<Poi> value : values) {
            value.setConverter(Poi.class);
            final Poi poi = value.get();
            Poi.Builder poiBuilder = Poi.newBuilder(poi);
            poiBuilder.setLabelgroupnodebts(poi.getConfidentnodebts() == 0 ?
                    0 : poi.getLabelgroupnodebts());
            poiBuilder.setLabelgroupnode(poi.getConfidentnode() == 0 ?
                    0 : poi.getLabelgroupnode());
            poiBuilder.setLabelgroupbts(poi.getConfidentbts() == 0 ?
                    0 : poi.getLabelgroupbts());
            context.write(NullWritable.get(),
                          new Text(PoiUtil.toString(poiBuilder.build())));
        }
    }
}

@@ -1331,114 +1331,1050 @@
     
-context.write(NullWritable.get(),%0A                          new Text(PoiUtil.toString(poiBuilder.build()))
+final Poi outputPoi = poiBuilder.build();%0A            Text output = new Text(outputPoi.getId() + PoiUtil.DELIMITER%0A                    + outputPoi.getNode() + PoiUtil.DELIMITER%0A                    + outputPoi.getBts() + PoiUtil.DELIMITER%0A                    + outputPoi.getLabelnodebts() + PoiUtil.DELIMITER%0A                    + outputPoi.getLabelgroupnodebts() + PoiUtil.DELIMITER%0A                    + outputPoi.getConfidentnodebts() + PoiUtil.DELIMITER%0A                    + outputPoi.getInoutWeek() + PoiUtil.DELIMITER%0A                    + outputPoi.getInoutWend() + PoiUtil.DELIMITER%0A                    + outputPoi.getLabelnode() + PoiUtil.DELIMITER%0A                    + outputPoi.getLabelgroupnode() + PoiUtil.DELIMITER%0A                    + outputPoi.getConfidentnode() + PoiUtil.DELIMITER%0A                    + outputPoi.getLabelbts() + PoiUtil.DELIMITER%0A                    + outputPoi.getLabelgroupbts() + PoiUtil.DELIMITER%0A                    + outputPoi.getConfidentbts());%0A            context.write(NullWritable.get(), output
 );%0A 

Fix VectorPoisOutReducer.