Implemented split of edge generation into multiple disjoint files and then merging of them

Author: ArnauPrat

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

@@ -18,21 +18,21 @@ public DistanceKnowsGenerator() {
         this.randomFarm = new RandomGeneratorFarm();
     }
 
-    public void generateKnows( ArrayList<Person> persons, int seed, float upperBound )  {
+    public void generateKnows( ArrayList<Person> persons, int seed, float percentage )  {
         randomFarm.resetRandomGenerators(seed);
         for( int i = 0; i < persons.size(); ++i ) {
             Person p = persons.get(i);
-           for( int j = i+1; (p.maxNumKnows()*upperBound > p.knows().size()) /*&& (j < (i + 1000))*/ && (j < persons.size()); ++j  ) {
-                if( know(p, persons.get(j), j - i, upperBound)) {
+           for( int j = i+1; (p.maxNumKnows()*percentage > p.knows().size()) && (j < persons.size()); ++j  ) {
+                if( know(p, persons.get(j), j - i, percentage)) {
                    createKnow(p, persons.get(j));
                 }
            }
         }
     }
 
-    boolean know( Person personA, Person personB, int dist, float upperBound ) {
-        if((float)(personA.knows().size()) >= (float)(personA.maxNumKnows())*upperBound ||
-           personB.knows().size() >= (float)(personB.maxNumKnows())*upperBound ) return false;
+    boolean know( Person personA, Person personB, int dist, float percentage ) {
+        if((float)(personA.knows().size()) >= (float)(personA.maxNumKnows())*percentage ||
+           personB.knows().size() >= (float)(personB.maxNumKnows())*percentage ) return false;
         double randProb = randomFarm.get(RandomGeneratorFarm.Aspect.UNIFORM).nextDouble();
         double prob = Math.pow(DatagenParams.baseProbCorrelated, dist);
         if ((randProb < prob) || (randProb < DatagenParams.limitProCorrelated)) {

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

@@ -48,6 +48,7 @@
 import org.apache.hadoop.io.Text;
 
 import java.io.OutputStream;
+import java.util.ArrayList;
 import java.util.Properties;
 
 public class LDBCDatagen {
@@ -70,39 +71,45 @@ private void printProgress(String message) {
 
     public int runGenerateJob(Configuration conf) throws Exception {
 
-        String personsFileName1 = conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/persons1";
-        String personsFileName2 = conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/persons2";
+        String hadoopPrefix = conf.get("ldbc.snb.datagen.serializer.hadoopDir");
         FileSystem fs = FileSystem.get(conf);
 
         long start = System.currentTimeMillis();
         printProgress("Starting: Person generation");
         long startPerson = System.currentTimeMillis();
         HadoopPersonGenerator personGenerator = new HadoopPersonGenerator( conf );
-        personGenerator.run(personsFileName1, "ldbc.snb.datagen.hadoop.UniversityKeySetter");
+        personGenerator.run(hadoopPrefix+"/persons", "ldbc.snb.datagen.hadoop.UniversityKeySetter");
         long endPerson = System.currentTimeMillis();
 
-
         printProgress("Creating university location correlated edges");
         long startUniversity = System.currentTimeMillis();
-        HadoopKnowsGenerator knowsGenerator = new HadoopKnowsGenerator(conf,null, "ldbc.snb.datagen.hadoop.InterestKeySetter", 0.45f);
-        knowsGenerator.run(personsFileName1,personsFileName2);
-        fs.delete(new Path(personsFileName1), true);
+        HadoopKnowsGenerator knowsGenerator = new HadoopKnowsGenerator(conf,"ldbc.snb.datagen.hadoop.UniversityKeySetter", "ldbc.snb.datagen.hadoop.RandomKeySetter", 0.45f);
+        knowsGenerator.run(hadoopPrefix+"/persons",hadoopPrefix+"/universityEdges");
         long endUniversity = System.currentTimeMillis();
 
         printProgress("Creating main interest correlated edges");
         long startInterest= System.currentTimeMillis();
-        knowsGenerator = new HadoopKnowsGenerator(conf,null, "ldbc.snb.datagen.hadoop.RandomKeySetter", 0.90f);
-        knowsGenerator.run(personsFileName2,personsFileName1);
-        fs.delete(new Path(personsFileName2), true);
+        knowsGenerator = new HadoopKnowsGenerator(conf,"ldbc.snb.datagen.hadoop.InterestKeySetter", "ldbc.snb.datagen.hadoop.RandomKeySetter", 0.45f);
+        knowsGenerator.run(hadoopPrefix+"/persons",hadoopPrefix+"/interestEdges");
         long endInterest = System.currentTimeMillis();
 
         printProgress("Creating random correlated edges");
         long startRandom= System.currentTimeMillis();
-        knowsGenerator = new HadoopKnowsGenerator(conf,null, "ldbc.snb.datagen.hadoop.RandomKeySetter", 1.0f);
-        knowsGenerator.run(personsFileName1,personsFileName2);
-        fs.delete(new Path(personsFileName1), true);
+        knowsGenerator = new HadoopKnowsGenerator(conf,"ldbc.snb.datagen.hadoop.RandomKeySetter", "ldbc.snb.datagen.hadoop.RandomKeySetter", 0.1f);
+        knowsGenerator.run(hadoopPrefix+"/persons",hadoopPrefix+"/randomEdges");
         long endRandom= System.currentTimeMillis();
 
+
+        fs.delete(new Path(DatagenParams.hadoopDir + "/persons"), true);
+        printProgress("Merging the different edge files");
+        ArrayList<String> edgeFileNames = new ArrayList<String>();
+        edgeFileNames.add(hadoopPrefix+"/universityEdges");
+        edgeFileNames.add(hadoopPrefix+"/interestEdges");
+        edgeFileNames.add(hadoopPrefix+"/randomEdges");
+        long startMerge = System.currentTimeMillis();
+        HadoopMergeFriendshipFiles merger = new HadoopMergeFriendshipFiles(conf,"ldbc.snb.datagen.hadoop.RandomKeySetter");
+        merger.run(hadoopPrefix+"/mergedPersons", edgeFileNames);
+        long endMerge = System.currentTimeMillis();
         /*printProgress("Creating edges to fill the degree gap");
         long startGap = System.currentTimeMillis();
         knowsGenerator = new HadoopKnowsGenerator(conf,null, "ldbc.snb.datagen.hadoop.DegreeGapKeySetter", 1.0f);
@@ -114,14 +121,14 @@ public int runGenerateJob(Configuration conf) throws Exception {
         printProgress("Serializing persons");
         long startPersonSerializing= System.currentTimeMillis();
         HadoopPersonSerializer serializer = new HadoopPersonSerializer(conf);
-        serializer.run(personsFileName2);
+        serializer.run(hadoopPrefix+"/mergedPersons");
         long endPersonSerializing= System.currentTimeMillis();
 
         long startPersonActivity= System.currentTimeMillis();
         if(conf.getBoolean("ldbc.snb.datagen.generator.activity", true)) {
             printProgress("Generating and serializing person activity");
             HadoopPersonActivityGenerator activityGenerator = new HadoopPersonActivityGenerator(conf);
-            activityGenerator.run(personsFileName2);
+            activityGenerator.run(hadoopPrefix+"/mergedPersons");
 
             int numThreads = DatagenParams.numThreads;
             int blockSize = DatagenParams.blockSize;
@@ -134,41 +141,41 @@ public int runGenerateJob(Configuration conf) throws Exception {
                 }
             }
         }
-        fs.delete(new Path(personsFileName2), true);
         long endPersonActivity= System.currentTimeMillis();
 
         long startSortingUpdateStreams= System.currentTimeMillis();
-        if(conf.getBoolean("ldbc.snb.datagen.serializer.updateStreams", false)) {
-            printProgress("Sorting update streams ");
 
-        int blockSize = DatagenParams.blockSize;
-        int numBlocks = (int)Math.ceil(DatagenParams.numPersons / (double)blockSize);
-
-        for( int i = 0; i < DatagenParams.numThreads; ++i) {
-            int numPartitions = conf.getInt("ldbc.snb.datagen.serializer.numUpdatePartitions", 1);
-            if( i < numBlocks ) {
-                for (int j = 0; j < numPartitions; ++j) {
-                    HadoopFileSorter updateStreamSorter = new HadoopFileSorter(conf, LongWritable.class, Text.class);
-                    updateStreamSorter.run(DatagenParams.hadoopDir + "/temp_updateStream_person_" + i + "_" + j, DatagenParams.hadoopDir + "/updateStream_person_" + i + "_" + j);
-                    updateStreamSorter.run(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + "_" + j, DatagenParams.hadoopDir + "/updateStream_forum_" + i + "_" + j);
+        if(conf.getBoolean("ldbc.snb.datagen.serializer.updateStreams", false)) {
 
-                    fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_person_" + i + "_" + j), true);
-                    fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + "_" + j), true);
+            printProgress("Sorting update streams ");
 
-                    HadoopUpdateStreamSerializer updateSerializer = new HadoopUpdateStreamSerializer(conf);
-                    updateSerializer.run(DatagenParams.hadoopDir + "/updateStream_person_" + i + "_" + j, i, j, "person");
-                    updateSerializer.run(DatagenParams.hadoopDir + "/updateStream_forum_" + i + "_" + j, i, j, "forum");
+            int blockSize = DatagenParams.blockSize;
+            int numBlocks = (int)Math.ceil(DatagenParams.numPersons / (double)blockSize);
 
-                    fs.delete(new Path(DatagenParams.hadoopDir + "/updateStream_person_" + i + "_" + j), true);
-                    fs.delete(new Path(DatagenParams.hadoopDir + "/updateStream_forum_" + i + "_" + j), true);
-                }
-            } else {
-                for (int j = 0; j < numPartitions; ++j) {
-                    fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_person_" + i + "_" + j), true);
-                    fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + "_" + j), true);
+            for( int i = 0; i < DatagenParams.numThreads; ++i) {
+                int numPartitions = conf.getInt("ldbc.snb.datagen.serializer.numUpdatePartitions", 1);
+                if( i < numBlocks ) {
+                    for (int j = 0; j < numPartitions; ++j) {
+                        HadoopFileSorter updateStreamSorter = new HadoopFileSorter(conf, LongWritable.class, Text.class);
+                        HadoopUpdateStreamSerializer updateSerializer = new HadoopUpdateStreamSerializer(conf);
+                        updateStreamSorter.run(DatagenParams.hadoopDir + "/temp_updateStream_person_" + i + "_" + j, DatagenParams.hadoopDir + "/updateStream_person_" + i + "_" + j);
+                        fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_person_" + i + "_" + j), true);
+                        updateSerializer.run(DatagenParams.hadoopDir + "/updateStream_person_" + i + "_" + j, i, j, "person");
+                        fs.delete(new Path(DatagenParams.hadoopDir + "/updateStream_person_" + i + "_" + j), true);
+                        if( conf.getBoolean("ldbc.snb.datagen.generator.activity", false)) {
+                            updateStreamSorter.run(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + "_" + j, DatagenParams.hadoopDir + "/updateStream_forum_" + i + "_" + j);
+                            fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + "_" + j), true);
+                            updateSerializer.run(DatagenParams.hadoopDir + "/updateStream_forum_" + i + "_" + j, i, j, "forum");
+                            fs.delete(new Path(DatagenParams.hadoopDir + "/updateStream_forum_" + i + "_" + j), true);
+                        }
+                    }
+                } else {
+                    for (int j = 0; j < numPartitions; ++j) {
+                        fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_person_" + i + "_" + j), true);
+                        fs.delete(new Path(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + "_" + j), true);
+                    }
                 }
             }
-        }
 
             long minDate = Long.MAX_VALUE;
             long maxDate = Long.MIN_VALUE;
@@ -188,18 +195,20 @@ public int runGenerateJob(Configuration conf) throws Exception {
                 file.close();
                 fs.delete(propertiesFile,true);
 
-                propertiesFile = new Path(DatagenParams.hadoopDir+"/temp_updateStream_forum_"+i+".properties");
-                file = fs.open(propertiesFile);
-                properties = new Properties();
-                properties.load(file);
-                aux = Long.parseLong(properties.getProperty("ldbc.snb.interactive.min_write_event_start_time"));
-                minDate = aux < minDate ? aux : minDate;
-                aux = Long.parseLong(properties.getProperty("ldbc.snb.interactive.max_write_event_start_time"));
-                maxDate = aux > maxDate ? aux : maxDate;
-                aux = Long.parseLong(properties.getProperty("ldbc.snb.interactive.num_events"));
-                count += aux;
-                file.close();
-                fs.delete(propertiesFile,true);
+                if( conf.getBoolean("ldbc.snb.datagen.generator.activity", false)) {
+                    propertiesFile = new Path(DatagenParams.hadoopDir + "/temp_updateStream_forum_" + i + ".properties");
+                    file = fs.open(propertiesFile);
+                    properties = new Properties();
+                    properties.load(file);
+                    aux = Long.parseLong(properties.getProperty("ldbc.snb.interactive.min_write_event_start_time"));
+                    minDate = aux < minDate ? aux : minDate;
+                    aux = Long.parseLong(properties.getProperty("ldbc.snb.interactive.max_write_event_start_time"));
+                    maxDate = aux > maxDate ? aux : maxDate;
+                    aux = Long.parseLong(properties.getProperty("ldbc.snb.interactive.num_events"));
+                    count += aux;
+                    file.close();
+                    fs.delete(propertiesFile, true);
+                }
             }
 
             OutputStream output = fs.create(new Path(DatagenParams.socialNetworkDir+"/updateStream"+".properties"),true);
@@ -229,8 +238,9 @@ public int runGenerateJob(Configuration conf) throws Exception {
                 + " total seconds");
         System.out.println("Person generation time: "+((endPerson - startPerson) / 1000));
         System.out.println("University correlated edge generation time: "+((endUniversity - startUniversity) / 1000));
-//        System.out.println("Interest correlated edge generation time: "+((endInterest - startInterest) / 1000));
-//        System.out.println("Random correlated edge generation time: "+((endRandom - startRandom) / 1000));
+        System.out.println("Interest correlated edge generation time: "+((endInterest - startInterest) / 1000));
+        System.out.println("Random correlated edge generation time: "+((endRandom - startRandom) / 1000));
+        System.out.println("Edges merge time: "+((endMerge - startMerge) / 1000));
         System.out.println("Person serialization time: "+((endPersonSerializing - startPersonSerializing) / 1000));
         System.out.println("Person activity generation and serialization time: "+((endPersonActivity - startPersonActivity) / 1000));
         System.out.println("Sorting update streams time: "+((endSortingUpdateStreams - startSortingUpdateStreams) / 1000));

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

@@ -88,7 +88,9 @@ public void run( String inputFileName, String outputFileName ) throws Exception
         String rankedFileName = conf.get("ldbc.snb.datagen.serializer.hadoopDir") + "/ranked";
         HadoopFileRanker hadoopFileRanker = new HadoopFileRanker( conf, TupleKey.class, Person.class );
         hadoopFileRanker.run(keyChangedFileName,rankedFileName);
-        fs.delete(new Path(keyChangedFileName), true);
+        if(preKeySetterName != null ) {
+            fs.delete(new Path(keyChangedFileName), true);
+        }
         System.out.println("... Time to rank persons: "+ (System.currentTimeMillis() - start)+" ms");
 
         conf.set("upperBound",Double.toString(upperBound));

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

@@ -0,0 +1,99 @@
+package ldbc.snb.datagen.hadoop;
+
+import ldbc.snb.datagen.generator.KnowsGenerator;
+import ldbc.snb.datagen.generator.LDBCDatagen;
+import ldbc.snb.datagen.objects.Knows;
+import ldbc.snb.datagen.objects.Person;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+/**
+ * Created by aprat on 29/07/15.
+ */
+public class HadoopMergeFriendshipFiles {
+
+    public static class HadoopMergeFriendshipFilesReducer  extends Reducer<TupleKey, Person, TupleKey, Person> {
+
+        private Configuration conf;
+        private HadoopFileKeyChanger.KeySetter<TupleKey> keySetter = null;
+
+        protected void setup(Context context) {
+            this.conf = context.getConfiguration();
+            try {
+                this.keySetter = (HadoopFileKeyChanger.KeySetter) Class.forName(conf.get("postKeySetterName")).newInstance();
+            }catch(Exception e) {
+                System.out.println(e.getMessage());
+            }
+            LDBCDatagen.init(conf);
+        }
+
+        @Override
+        public void reduce(TupleKey key, Iterable<Person> valueSet,Context context)
+                throws IOException, InterruptedException {
+
+            Person person = null;
+            int index = 0;
+            for ( Person p : valueSet) {
+                if( index == 0 ) {
+                    person = new Person(p);
+                } else {
+                    for ( Knows k : p.knows() ) {
+                        person.knows().add(k);
+                    }
+                }
+                index++;
+            }
+            //System.out.println("Num persons "+index);
+            context.write(keySetter.getKey(person),person);
+        }
+    }
+
+    private Configuration conf;
+    private String postKeySetterName;
+
+    public HadoopMergeFriendshipFiles( Configuration conf,  String postKeySetterName ) {
+
+        this.conf = conf;
+        this.postKeySetterName = postKeySetterName;
+    }
+
+    public void run( String outputFileName, ArrayList<String> friendshipFileNames ) throws Exception {
+
+        conf.set("postKeySetterName",postKeySetterName);
+        int numThreads = Integer.parseInt(conf.get("ldbc.snb.datagen.generator.numThreads"));
+        Job job = Job.getInstance(conf, "Edges merger generator");
+        job.setMapOutputKeyClass(TupleKey.class);
+        job.setMapOutputValueClass(Person.class);
+        job.setOutputKeyClass(TupleKey.class);
+        job.setOutputValueClass(Person.class);
+        //job.setJarByClass(HadoopBlockMapper.class);
+        //job.setMapperClass(HadoopBlockMapper.class);
+        job.setReducerClass(HadoopMergeFriendshipFilesReducer.class);
+        job.setNumReduceTasks(numThreads);
+        job.setInputFormatClass(SequenceFileInputFormat.class);
+        job.setOutputFormatClass(SequenceFileOutputFormat.class);
+
+        for ( String s : friendshipFileNames ) {
+            FileInputFormat.addInputPath(job, new Path(s));
+        }
+        FileOutputFormat.setOutputPath(job, new Path(outputFileName));
+
+        System.out.println("Merging edges");
+        long start = System.currentTimeMillis();
+        if(!job.waitForCompletion(true) ){
+            throw new Exception();
+        }
+        System.out.println("... time to merge edges: "+ (System.currentTimeMillis() - start)+" ms");
+
+
+    }
+}

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

@@ -51,7 +51,7 @@ public void reduce(BlockKey key, Iterable<Person> valueSet,Context context)
 			throws IOException, InterruptedException {
 			personSerializer_.reset();
 			for( Person p : valueSet ) {
-				if(p.creationDate()< Dictionaries.dates.getUpdateThreshold() || !DatagenParams.updateStreams ) {
+				if(p.creationDate()< Dictionaries.dates.getUpdateThreshold() || !DatagenParams.updateStreams  ) {
 					personSerializer_.export(p);
 				} else {
 					updateSerializer_.export(p);