@@ -0,0 +1,2127 @@
+package es.tid.cosmos.mobility.clientlabelling;%0A%0Aimport java.io.IOException;%0A%0Aimport com.twitter.elephantbird.mapreduce.io.ProtobufWritable;%0Aimport org.apache.hadoop.mapreduce.Reducer;%0Aimport org.apache.hadoop.mapreduce.Reducer.Context;%0A%0Aimport es.tid.cosmos.mobility.data.MobProtocol.ClusterVector;%0Aimport es.tid.cosmos.mobility.data.MobProtocol.NodeBts;%0A%0A/**%0A *%0A * @author dmicol%0A */%0Apublic class VectorNormalizedReducer extends Reducer%3CProtobufWritable%3CNodeBts%3E,%0A        ProtobufWritable%3CClusterVector%3E, ProtobufWritable%3CNodeBts%3E,%0A        ProtobufWritable%3CClusterVector%3E%3E %7B%0A    @Override%0A    protected void reduce(ProtobufWritable%3CNodeBts%3E key,%0A                          Iterable%3CProtobufWritable%3CClusterVector%3E%3E values,%0A            Context context) throws IOException, InterruptedException %7B%0A        ClusterVector.Builder vectorNormBuilder = ClusterVector.newBuilder();%0A        ClusterVector.Builder divBuilder = ClusterVector.newBuilder();%0A        final NodeBts bts = key.get();%0A        double elem;%0A        for (ProtobufWritable%3CClusterVector%3E value : values) %7B%0A            final ClusterVector clusterVector = value.get();%0A            double sumvalues = 0;%0A%0A            for (int j = 0; j %3C clusterVector.getComsCount(); j++) %7B%0A                if (j %3C 24) %7B%0A                    // Mondays, Tuesday, Wednesday and Thursday. Total: 121 days%0A                    elem = clusterVector.getComs(j) / 121.0D;%0A                %7D else %7B%0A                    elem = clusterVector.getComs(j) / 31.0D;%0A                %7D%0A                sumvalues += elem;%0A                divBuilder.addComs(elem);%0A            %7D%0A%0A            ClusterVector div = divBuilder.build();%0A            for (double comm : div.getComsList()) %7B%0A                elem = comm / sumvalues;%0A                vectorNormBuilder.addComs(elem);%0A            %7D%0A            ClusterVector vectorNorm = vectorNormBuilder.build();%0A            ProtobufWritable%3CClusterVector%3E vectorNormWrapper =%0A                    ProtobufWritable.newInstance(ClusterVector.class);%0A            vectorNormWrapper.set(vectorNorm);%0A            context.write(key, vectorNormWrapper);%0A        %7D%0A    %7D%0A%7D%0A

Add VectorNormalizedReducer.