Implemented keychanger into ranking

Author: ArnauPrat

src/main/java/ldbc/snb/datagen/generator/LDBCDatagen.java

@@ -278,6 +278,7 @@ public static void main(String[] args) /*throws Exception*/ {
         }catch(Exception e ) {
             System.err.println("Error during execution");
             System.err.println(e.getMessage());
+            e.printStackTrace();
             System.exit(1);
         }
     }

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

@@ -6,6 +6,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.Partitioner;
 import org.apache.hadoop.mapreduce.Reducer;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
@@ -23,19 +24,53 @@ public class HadoopFileRanker {
     private Configuration conf;
     private Class<?> K;
     private Class<?> V;
+    private String keySetterName;
 
     /**
      *
      * @param conf The configuration object.
      * @param K The Key class of the hadoop sequence file.
      * @param V The Value class of the hadoop sequence file.
      */
-    public HadoopFileRanker( Configuration conf, Class<?> K, Class<?> V) {
+    public HadoopFileRanker( Configuration conf, Class<?> K, Class<?> V, String keySetter ) {
         this.conf  = new Configuration(conf);
+        this.keySetterName = keySetter;
         this.K = K;
         this.V = V;
     }
 
+    public static class HadoopFileRankerSortMapper<K, V>  extends Mapper<K, V, K, V> {
+
+        HadoopFileKeyChanger.KeySetter<TupleKey> keySetter;
+        private long counter = 0;       /** Counter of the number of elements received by this reducer.*/
+
+        @Override
+        public void setup( Context context )  {
+
+            try {
+                LDBCDatagen.init(context.getConfiguration());
+                String className = context.getConfiguration().get("keySetterClassName");
+                keySetter = (HadoopFileKeyChanger.KeySetter) Class.forName(className).newInstance();
+            } catch(ClassNotFoundException e) {
+                System.out.print(e.getMessage());
+            } catch(IllegalAccessException e) {
+                System.out.print(e.getMessage());
+            } catch(InstantiationException e) {
+                System.out.print(e.getMessage());
+            }
+        }
+
+        @Override
+        public void map(K key, V value,
+                           Context context) throws IOException, InterruptedException {
+            context.write((K)keySetter.getKey(value), value);
+        }
+
+        @Override
+        public void cleanup(Context context) {
+        }
+    }
+
     public static class HadoopFileRankerSortReducer<K, V, T extends BlockKey>  extends Reducer<K, V, BlockKey, V> {
 
         private int reducerId;          /** The id of the reducer.**/
@@ -136,12 +171,19 @@ public void reduce(BlockKey key, Iterable<V> valueSet,
     public void run( String inputFileName, String outputFileName ) throws Exception {
         int numThreads = conf.getInt("ldbc.snb.datagen.generator.numThreads",1);
 
+        if( keySetterName != null ) {
+            conf.set("keySetterClassName", keySetterName);
+        }
+
         /** First Job to sort the key-value pairs and to count the number of elements processed by each reducer.**/
         Job jobSort = Job.getInstance(conf, "Sorting "+inputFileName);
 
         FileInputFormat.setInputPaths(jobSort, new Path(inputFileName));
-        FileOutputFormat.setOutputPath(jobSort, new Path(conf.get("ldbc.snb.datagen.serializer.hadoopDir")+"/rankIntermediate"));
+        FileOutputFormat.setOutputPath(jobSort, new Path(conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/rankIntermediate"));
 
+        if( keySetterName != null ) {
+            jobSort.setMapperClass(HadoopFileRankerSortMapper.class);
+        }
         jobSort.setMapOutputKeyClass(K);
         jobSort.setMapOutputValueClass(V);
         jobSort.setOutputKeyClass(BlockKey.class);

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

@@ -9,6 +9,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.Reducer;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -86,7 +87,7 @@ public void run( String inputFileName, String outputFileName ) throws Exception
 
         FileSystem fs = FileSystem.get(conf);
 
-        String keyChangedFileName = inputFileName;
+        /*String keyChangedFileName = inputFileName;
         if(preKeySetterName != null) {
             System.out.println("Changing key of persons");
             long start = System.currentTimeMillis();
@@ -95,18 +96,20 @@ public void run( String inputFileName, String outputFileName ) throws Exception
             keyChanger.run(inputFileName, keyChangedFileName);
             System.out.println("... Time to change keys: "+ (System.currentTimeMillis() - start)+" ms");
         }
+        */
 
         System.out.println("Ranking persons");
         long start = System.currentTimeMillis();
         String rankedFileName = conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/ranked";
-        HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class );
-        hadoopFileRanker.run(keyChangedFileName,rankedFileName);
-        if(preKeySetterName != null ) {
+        HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class , preKeySetterName);
+        hadoopFileRanker.run(inputFileName,rankedFileName);
+/*        if(preKeySetterName != null ) {
             fs.delete(new Path(keyChangedFileName), true);
         }
-        System.out.println("... Time to rank persons: "+ (System.currentTimeMillis() - start)+" ms");
+        */
+        System.out.println("... Time to rank persons: " + (System.currentTimeMillis() - start) + " ms");
 
-        conf.setInt("stepIndex",step_index);
+        conf.setInt("stepIndex", step_index);
         int index = 0;
         for( float p : percentages ) {
             conf.setFloat("percentage"+index, p);

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

@@ -62,7 +62,7 @@ public void reduce(TupleKey key, Iterable<Person> valueSet,Context context)
 
     public HadoopMergeFriendshipFiles( Configuration conf,  String postKeySetterName ) {
 
-        this.conf = conf;
+        this.conf = new Configuration(conf);
         this.postKeySetterName = postKeySetterName;
     }
 

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

@@ -116,7 +116,7 @@ public void run( String inputFileName ) throws Exception {
         FileSystem fs = FileSystem.get(conf);
 
         String rankedFileName = conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/ranked";
-        HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class );
+        HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class, null );
         hadoopFileRanker.run(inputFileName,rankedFileName);
 
         int numThreads = Integer.parseInt(conf.get("ldbc.snb.datagen.generator.numThreads"));

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

@@ -86,7 +86,7 @@ public void run( String inputFileName ) throws Exception {
 		FileSystem fs = FileSystem.get(conf);
 
 		String rankedFileName = conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/ranked";
-		HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class );
+		HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class, null );
         hadoopFileRanker.run(inputFileName,rankedFileName);
 
 		int numThreads = Integer.parseInt(conf.get("ldbc.snb.datagen.generator.numThreads"));