package es.tid.cosmos.mobility.mapreduce;

import java.io.IOException;

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

import es.tid.cosmos.mobility.data.BtsCounterUtil;
import es.tid.cosmos.mobility.data.MobProtocol.BtsCounter;
import es.tid.cosmos.mobility.data.MobProtocol.NodeBts;

/**
 *
 * @author dmicol
 */
public class NodeBtsCounterReducer extends Reducer<
        ProtobufWritable<NodeBts>, NullWritable, LongWritable,
        ProtobufWritable<BtsCounter>> {
    @Override
    protected void reduce(ProtobufWritable<NodeBts> key,
                          Iterable<NullWritable> values, Context context)
            throws IOException, InterruptedException {
        int count = 0;
        for (NullWritable unused : values) {
            count++;
        }
        NodeBts node = key.get();
        BtsCounter counter = BtsCounterUtil.create(node.getPlaceId(),
                                                   node.getWeekday(),
                                                   node.getRange(), count);
        ProtobufWritable<BtsCounter> wrapper = ProtobufWritable.newInstance(
                BtsCounter.class);
        wrapper.set(counter);
        context.write(new LongWritable(node.getUserId()), wrapper);
    }
}

@@ -608,16 +608,162 @@
 ter%3E%3E %7B%0A
+    private LongWritable userId;%0A    %0A    @Override%0A    public void setup(Context context) %7B%0A        this.userId = new LongWritable();%0A    %7D%0A    %0A
     @Ove
@@ -764,24 +764,24 @@
   @Override%0A
-
     protecte
@@ -1061,16 +1061,22 @@
         
+final 
 NodeBts 
@@ -1105,355 +1105,196 @@
     
-BtsCounter counter = BtsCounterUtil.create(node.getPlaceId(),%0A                                                   node.getWeekday(),%0A                                                   node.getRange(), count);%0A        ProtobufWritable%3CBtsCounter%3E wrapper = ProtobufWritable.newInstance(%0A                BtsCounter.class);%0A        wrapper.set(
+this.userId.set(node.getUserId());%0A        ProtobufWritable%3CBtsCounter%3E counter = BtsCounterUtil.createAndWrap(%0A                node.getPlaceId(), node.getWeekday(), node.getRange(), 
 count
-er
 );%0A 
@@ -1318,49 +1318,26 @@
 ite(
-new LongWritable(node.getUserId()), wrapp
+this.userId, count
 er);

Improve the NodeBtsCounterReducer.