Fixed critial bug when more than one updatePartitions were set

Author: ArnauPrat

src/main/java/ldbc/snb/datagen/hadoop/HadoopUpdateStreamSorterAndSerializer.java

@@ -55,7 +55,9 @@ public void reduce(UpdateEventKey key, Iterable<Text> valueSet,Context context)
 					Path outFile = new Path(context.getConfiguration().get("ldbc.snb.datagen.serializer.socialNetworkDir")+"/updateStream_"+key.reducerId+"_"+key.partition+"_"+streamType+".csv");
 					out = fs.create(outFile);
 				}
+				int counter = 0;
 				for( Text t : valueSet ) {
+					counter++;
 					out.write(t.toString().getBytes("UTF8"));
 				}
 				out.close();
@@ -95,6 +97,7 @@ public void run(List<String> inputFileNames, String type ) throws Exception {
 		job.setOutputFormatClass(SequenceFileOutputFormat.class);
 		job.setPartitionerClass(HadoopUpdateEventKeyPartitioner.class);
 		job.setGroupingComparatorClass(UpdateEventKeyGroupComparator.class);
+		//job.setSortComparatorClass(UpdateEventKeySortComparator.class);
 
 		for(String s : inputFileNames) {
 			FileInputFormat.addInputPath(job, new Path(s));

src/main/java/ldbc/snb/datagen/hadoop/UpdateEventKey.java

@@ -46,10 +46,10 @@ public void readFields(DataInput in) throws IOException {
     }
 
     public int compareTo( UpdateEventKey key) {
-        if( date < key.date) return -1;
-        if( date > key.date) return 1;
         if (reducerId != key.reducerId) return reducerId - key.reducerId;
         if (partition != key.partition) return partition - key.partition;
+        if( date < key.date) return -1;
+        if( date > key.date) return 1;
         return 0;
     }
 }

src/main/java/ldbc/snb/datagen/hadoop/UpdateEventKeySortComparator.java

@@ -0,0 +1,26 @@
+package ldbc.snb.datagen.hadoop;
+
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+
+/**
+ * Created by aprat on 11/17/14.
+ */
+
+public class UpdateEventKeySortComparator extends WritableComparator {
+
+    protected UpdateEventKeySortComparator() {
+        super(UpdateEventKey.class,true);
+    }
+
+    @Override
+    public int compare(WritableComparable a, WritableComparable b) {
+        UpdateEventKey keyA = (UpdateEventKey)a;
+        UpdateEventKey keyB = (UpdateEventKey)b;
+        if (keyA.reducerId != keyB.reducerId) return keyA.reducerId - keyB.reducerId;
+        if (keyA.partition != keyB.partition) return keyA.partition - keyB.partition;
+        if( keyA.date < keyB.date) return -1;
+        if( keyA.date > keyB.date) return 1;
+        return 0;
+    }
+}