Disabled update stream sorting when update stream generationis not enabled.

Improved ranking

Author: ArnauPrat

src/main/java/ldbc/socialnet/dbgen/generator/MRGenerateUsers.java

@@ -43,6 +43,7 @@
 import ldbc.socialnet.dbgen.util.MapReduceKey;
 
 import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.Reducer;
 
 
@@ -65,11 +66,12 @@ public void writeReducedUserProfiles(int from, int to, int pass,
         try {
             to = to % windowSize;
             for (int i = from; i != to; i = (i+1)%windowSize) {
-                int key = userProfiles[i].getDicElementId(pass);
+             /*   int key = userProfiles[i].getDicElementId(pass);
                 int block = 0;
                 long id = userProfiles[i].getAccountId();
                 MapReduceKey mpk = new MapReduceKey(block,key,id);
-                context.write(mpk, userProfiles[i]);
+                context.write(mpk, userProfiles[i]);*/
+                context.write(new LongWritable(userProfiles[i].getDicElementId(pass)), userProfiles[i]);
                 numberSerializedObject++;
             }
         }

src/main/java/ldbc/socialnet/dbgen/generator/MRWriter.java

@@ -58,6 +58,7 @@
 import ldbc.socialnet.dbgen.objects.*;
 import ldbc.socialnet.dbgen.serializer.*;
 import ldbc.socialnet.dbgen.storage.StorageManager;
+import ldbc.socialnet.dbgen.util.ComposedKey;
 import ldbc.socialnet.dbgen.util.ScaleFactor;
 import ldbc.socialnet.dbgen.vocabulary.SN;
 import ldbc.socialnet.dbgen.util.MapReduceKey;
@@ -66,6 +67,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.Mapper.Context;
 import org.apache.hadoop.mapreduce.Reducer;
 
@@ -781,7 +783,7 @@ public void closeSerializer() {
         System.out.println("Writing the data for test driver ");
     }
 
-    public void generateUserActivity( ReducedUserProfile userProfile, Reducer<MapReduceKey, ReducedUserProfile,MapReduceKey, ReducedUserProfile>.Context context) {
+    public void generateUserActivity( ReducedUserProfile userProfile, Reducer<ComposedKey, ReducedUserProfile,LongWritable, ReducedUserProfile>.Context context) {
         int index = numUserProfilesRead%windowSize;
         numUserProfilesRead++;
         reducedUserProfiles[index] = userProfile;
@@ -829,7 +831,7 @@ public void resetWindow() {
         mrCurCellPost = 0;
     }
 
-    public void pushUserProfile(ReducedUserProfile reduceUser, int pass, int outputDimension, Reducer<MapReduceKey, ReducedUserProfile,MapReduceKey, ReducedUserProfile>.Context context){
+    public void pushUserProfile(ReducedUserProfile reduceUser, int pass, int outputDimension, Reducer<ComposedKey, ReducedUserProfile,LongWritable, ReducedUserProfile>.Context context){
         ReducedUserProfile userObject = new ReducedUserProfile();
         userObject.copyFields(reduceUser);
         totalNumUserProfilesRead++;
@@ -851,7 +853,7 @@ public void pushUserProfile(ReducedUserProfile reduceUser, int pass, int outputD
         }
     }
 
-    public void pushAllRemainingUser(int pass, int outputDimension, Reducer<MapReduceKey, ReducedUserProfile,MapReduceKey, ReducedUserProfile>.Context context){
+    public void pushAllRemainingUser(int pass, int outputDimension, Reducer<ComposedKey, ReducedUserProfile,LongWritable, ReducedUserProfile>.Context context){
 
         // For each remianing cell in the window, we create the edges.
         for (int numLeftCell = Math.min(numberOfCellPerWindow, numUserProfilesRead/cellSize); numLeftCell > 0; --numLeftCell, ++mrCurCellPost) {
@@ -910,11 +912,7 @@ public void mrGenerateUserInfo(int pass, Context context){
                 ReducedUserProfile reduceUserProf = generateGeneralInformation(j);
                 ++numUsersToGenerate;
                 try {
-                    int block =  0;                                                                  // The mapreduce group this university will be assigned.
-                    int key = reduceUserProf.getDicElementId(pass);                                  // The key used to sort within the block.
-                    long id = reduceUserProf.getAccountId();                                         // The id used to sort within the key, to guarantee determinism.
-                    MapReduceKey mpk = new MapReduceKey( block, key, id );
-                    context.write(mpk, reduceUserProf);
+                    context.write(new LongWritable(reduceUserProf.getDicElementId(pass)), reduceUserProf);
                 } catch (IOException e) {
                     e.printStackTrace();
                 } catch (InterruptedException e) {

src/main/java/ldbc/socialnet/dbgen/generator/ScalableGenerator.java

@@ -0,0 +1,49 @@
+
+package ldbc.socialnet.dbgen.util;
+
+import org.apache.hadoop.io.WritableComparable;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Created by aprat on 11/9/14.
+ */
+
+    public class ComposedKey implements WritableComparable<ComposedKey> {
+        public long block;
+        public long key;
+
+        public ComposedKey( ) {
+        }
+
+        public ComposedKey(ComposedKey cK) {
+            this.block = cK.block;
+            this.key = cK.key;
+        }
+
+        public ComposedKey( long block, long key) {
+            this.block = block;
+            this.key = key;
+        }
+
+        @Override
+        public void write(DataOutput out) throws IOException {
+            out.writeLong(block);
+            out.writeLong(key);
+        }
+
+        @Override
+        public void readFields(DataInput in) throws IOException {
+            block = in.readLong();
+            key = in.readLong();
+        }
+
+        @Override
+        public int compareTo( ComposedKey mpk) {
+            if (block < mpk.block) return -1;
+            if (block > mpk.block) return 1;
+            return 0;
+        }
+    }

src/main/java/ldbc/socialnet/dbgen/util/ComposedKey.java

@@ -0,0 +1,26 @@
+package ldbc.socialnet.dbgen.util;
+
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+
+/**
+ * Created by aprat on 11/9/14.
+ */
+public class ComposedKeyComparator extends WritableComparator {
+
+    protected ComposedKeyComparator() {
+        super(ComposedKey.class,true);
+    }
+
+    @Override
+    public int compare(WritableComparable a, WritableComparable b) {
+        //return a.compareTo(b);
+        ComposedKey keyA = (ComposedKey)a;
+        ComposedKey keyB = (ComposedKey)b;
+        if (keyA.block < keyB.block) return -1;
+        if (keyA.block > keyB.block) return 1;
+        if (keyA.key < keyB.key) return -1;
+        if (keyA.key > keyB.key) return 1;
+        return 0;
+    }
+}

src/main/java/ldbc/socialnet/dbgen/util/ComposedKeyComparator.java

@@ -0,0 +1,25 @@
+package ldbc.socialnet.dbgen.util;
+
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+
+/**
+ * Created by aprat on 11/17/14.
+ */
+
+public class ComposedKeyGroupComparator extends WritableComparator {
+
+    protected ComposedKeyGroupComparator() {
+        super(ComposedKey.class,true);
+    }
+
+    @Override
+    public int compare(WritableComparable a, WritableComparable b) {
+        //return a.compareTo(b);
+        ComposedKey keyA = (ComposedKey)a;
+        ComposedKey keyB = (ComposedKey)b;
+        if (keyA.block < keyB.block) return -1;
+        if (keyA.block > keyB.block) return 1;
+        return 0;
+    }
+}

src/main/java/ldbc/socialnet/dbgen/util/ComposedKeyGroupComparator.java

@@ -0,0 +1,27 @@
+package ldbc.socialnet.dbgen.util;
+
+import ldbc.socialnet.dbgen.objects.ReducedUserProfile;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.mapreduce.Mapper;
+
+import java.io.IOException;
+
+/**
+ * Created by aprat on 11/17/14.
+ */
+public class HadoopBlockMapper extends Mapper<LongWritable, ReducedUserProfile, ComposedKey, ReducedUserProfile> {
+    int mapId;
+
+    @Override
+    public void setup(Mapper.Context context) {
+        Configuration conf = context.getConfiguration();
+        mapId = context.getTaskAttemptID().getId();
+    }
+
+    @Override
+    public void map(LongWritable key, ReducedUserProfile value, Mapper.Context context)
+            throws IOException, InterruptedException {
+        context.write(new ComposedKey(key.get() / 10000, key.get()), value);
+    }
+}

src/main/java/ldbc/socialnet/dbgen/util/HadoopBlockMapper.java

@@ -0,0 +1,19 @@
+package ldbc.socialnet.dbgen.util;
+
+import ldbc.socialnet.dbgen.objects.ReducedUserProfile;
+import org.apache.hadoop.mapreduce.Partitioner;
+
+/**
+ * Created by aprat on 11/17/14.
+ */
+public class HadoopBlockPartitioner extends Partitioner<ComposedKey, ReducedUserProfile> {
+
+    public HadoopBlockPartitioner() {
+        super();
+    }
+
+    @Override
+    public int getPartition(ComposedKey key, ReducedUserProfile person, int numReduceTasks) {
+        return (int)(key.block % numReduceTasks);
+    }
+}