package es.tid.cosmos.mobility.mapreduce;

import java.io.IOException;

import com.twitter.elephantbird.mapreduce.io.ProtobufWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;

import es.tid.cosmos.mobility.data.MobProtocol.BtsCounter;
import es.tid.cosmos.mobility.data.MobProtocol.NodeBtsDay;
import es.tid.cosmos.mobility.data.MobProtocol.NodeMxCounter;
import es.tid.cosmos.mobility.data.NodeBtsDayUtil;

/**
 *
 * @author dmicol
 */
public class RepbtsSpreadNodebtsMapper extends Mapper<LongWritable,
        ProtobufWritable<NodeMxCounter>, ProtobufWritable<NodeBtsDay>,
        IntWritable> {
    @Override
    public void map(LongWritable key, ProtobufWritable<NodeMxCounter> value,
            Context context) throws IOException, InterruptedException {
        final NodeMxCounter counter = value.get();
        for (BtsCounter bts : counter.getBtsList()) {
            ProtobufWritable<NodeBtsDay> nodeWrapper = NodeBtsDayUtil.
                    createAndWrap(key.get(), bts.getPlaceId(), 0, 0);
            context.write(nodeWrapper, new IntWritable(bts.getCount()));
        }
    }
}

@@ -678,16 +678,174 @@
 able%3E %7B%0A
+    private IntWritable numberOfCalls;%0A    %0A    @Override%0A    public void setup(Context context) %7B%0A        this.numberOfCalls = new IntWritable();%0A    %7D%0A    %0A
     @Ove
@@ -1149,23 +1149,22 @@
 ay%3E node
-Wrapper
+BtsDay
  = NodeB
@@ -1260,66 +1260,104 @@
     
-context.write(nodeWrapper, new IntWritable(bts.getCount())
+this.numberOfCalls.set(bts.getCount());%0A            context.write(nodeBtsDay, this.numberOfCalls
 );%0A 

Improve RepbtsSpreadNodebtsMapper.