package es.tid.bdp.profile.export.ps;

import java.io.IOException;
import java.util.Iterator;

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

import es.tid.bdp.profile.data.ProfileProtocol.CategoryCount;
import es.tid.bdp.profile.data.ProfileProtocol.UserProfile;

/**
 *
 * @author dmicol, sortega
 */
public class PSExporterReducer extends Reducer<Text,
                                               ProtobufWritable<UserProfile>,
                                               NullWritable, Text> {
    private StringBuilder builder;
    private Text record;

    @Override
    public void setup(Context context) throws IOException {
        this.builder = new StringBuilder();
        this.record = new Text();
    }
    
    @Override
    public void reduce(Text userId,
                       Iterable<ProtobufWritable<UserProfile>> profiles,
                       Context context) throws IOException,
                                               InterruptedException {
        Counter recordCount = context.getCounter(PSExporterCounter.NUM_RECORDS);
        for (Iterator<ProtobufWritable<UserProfile>> it = profiles.iterator();
                it.hasNext();) {
            final ProtobufWritable<UserProfile> wrappedProfile = it.next();
            wrappedProfile.setConverter(UserProfile.class);
            UserProfile profile = wrappedProfile.get();

            if (recordCount.getValue() == 0L) {
                this.builder.setLength(0);
                this.builder.append("User");
                this.builder.append("|");
                for (CategoryCount count : profile.getCountsList()) {
                    this.builder.append(count.getName());
                    this.builder.append("|");
                }
            }
            this.record.set(this.builder.toString());
            context.write(NullWritable.get(), this.record);

            this.builder.setLength(0);
            String userIdAndDate = userId + "_" + profile.getDate();
            this.builder.append(userIdAndDate);
            for (CategoryCount count : profile.getCountsList()) {
                this.builder.append(count.getCount());
                this.builder.append("|");
            }
            this.record.set(this.builder.toString());
            context.write(NullWritable.get(), this.record);

            recordCount.increment(1L);
        }
    }
}

@@ -736,16 +736,51 @@
  record;
+%0A    private Counter recordCounter;
 %0A%0A    @O
@@ -921,24 +921,104 @@
 new Text();%0A
+        this.recordCounter = context.getCounter(PSExporterCounter.NUM_RECORDS);%0A
     %7D%0A    %0A 
@@ -1273,89 +1273,8 @@
 n %7B%0A
-        Counter recordCount = context.getCounter(PSExporterCounter.NUM_RECORDS);%0A
     
@@ -1590,16 +1590,21 @@
     if (
+this.
 recordCo
@@ -1602,24 +1602,26 @@
 .recordCount
+er
 .getValue() 
@@ -2540,16 +2540,21 @@
         
+this.
 recordCo
@@ -2556,16 +2556,18 @@
 ordCount
+er
 .increme
@@ -2590,10 +2590,372 @@
 %7D%0A    %7D%0A
+        %0A    @Override%0A    public void cleanup(Context context) throws IOException,%0A                                                InterruptedException %7B%0A        this.builder.setLength(0);%0A        this.builder.append(this.recordCounter.getValue());%0A        this.record.set(this.builder.toString());%0A        context.write(NullWritable.get(), this.record);%0A    %7D%0A
 %7D%0A

Add entry count at the end of the PS data file.