Merge branch 'new_execution_flow'

Author: ArnauPrat

params.ini

@@ -11,5 +11,5 @@ ldbc.snb.datagen.serializer.personActivitySerializer:ldbc.snb.datagen.serializer
 
 ldbc.snb.datagen.generator.numThreads:1
 
-ldbc.snb.datagen.serializer.updateStreams:true
+
 

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

@@ -1,8 +1,8 @@
 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.Knows.FullComparator;
 import ldbc.snb.datagen.objects.Person;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
@@ -15,16 +15,19 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 
 /**
  * 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;
+        private int numRepeated = 0;
 
         protected void setup(Context context) {
             this.conf = context.getConfiguration();
@@ -40,21 +43,41 @@ protected void setup(Context context) {
         public void reduce(TupleKey key, Iterable<Person> valueSet,Context context)
                 throws IOException, InterruptedException {
 
+            ArrayList<Knows> knows = new ArrayList<Knows>();
             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);
-                    }
+                }
+                for(Knows k : p.knows()) {
+                    knows.add(k);
                 }
                 index++;
             }
+            person.knows().clear();
+            Knows.FullComparator comparator = new Knows.FullComparator();
+            Collections.sort(knows, comparator);
+            if(knows.size() > 0 ) {
+                long currentTo = knows.get(0).to().accountId();
+                person.knows().add(knows.get(0));
+                for (index = 1; index < knows.size(); ++index) {
+                    Knows nextKnows = knows.get(index);
+                    if(currentTo != knows.get(index).to().accountId()) {
+                        person.knows().add(nextKnows);
+                        currentTo = nextKnows.to().accountId();
+                    } else {
+                        numRepeated++;
+                    }
+                }
+            }
+
             //System.out.println("Num persons "+index);
             context.write(keySetter.getKey(person),person);
         }
+        protected void cleanup(Context context){
+            System.out.println("Number of repeated edges: "+numRepeated);
+		}
     }
 
     private Configuration conf;

src/main/java/ldbc/snb/datagen/objects/Knows.java

@@ -36,12 +36,17 @@
  */
 package ldbc.snb.datagen.objects;
 
+import ldbc.snb.datagen.dictionary.Dictionaries;
+import ldbc.snb.datagen.generator.DatagenParams;
+import ldbc.snb.datagen.util.RandomGeneratorFarm;
 import org.apache.hadoop.io.Writable;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
-
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Random;
 
 
 public class Knows implements Writable, Comparable<Knows> {
@@ -103,11 +108,55 @@ public void write(DataOutput arg0) throws IOException {
     }
 
     public int compareTo(Knows k) {
-        int res =  (int)(to_.accountId() - k.to().accountId());
-        if( res != 0 ) return res;
-        long res2 = creationDate_ - k.creationDate();
+        long res =  (to_.accountId() - k.to().accountId());
+        if( res > 0 ) return 1;
+        if( res < 0 ) return -1;
+        /*long res2 = creationDate_ - k.creationDate();
         if( res2 > 0 ) return 1;
         if( res2 < 0 ) return -1;
+        */
         return 0;
     }
+
+    static public class FullComparator implements Comparator<Knows> {
+
+        public int compare(Knows a, Knows b) {
+            long res =  (a.to_.accountId() - b.to().accountId());
+            if( res > 0 ) return 1;
+            if( res < 0 ) return -1;
+            long res2 = a.creationDate_ - b.creationDate();
+            if( res2 > 0 ) return 1;
+            if( res2 < 0 ) return -1;
+            return 0;
+        }
+
+    }
+
+    public static int num = 0;
+
+    public static boolean createKnow( Random random, Person personA, Person personB ) {
+        long  creationDate = Dictionaries.dates.randomKnowsCreationDate(
+                random,
+                personA,
+                personB);
+        creationDate = creationDate - personA.creationDate() >= DatagenParams.deltaTime ? creationDate : creationDate + (DatagenParams.deltaTime - (creationDate - personA.creationDate()));
+        creationDate = creationDate - personB.creationDate() >= DatagenParams.deltaTime ? creationDate : creationDate + (DatagenParams.deltaTime - (creationDate - personB.creationDate()));
+        if( creationDate <= Dictionaries.dates.getEndDateTime() ) {
+            float similarity = Person.Similarity(personA,personB);
+            if(!personB.knows().add(new Knows(personA, creationDate, similarity))) return false;
+            if(!personA.knows().add(new Knows(personB, creationDate, similarity))) return false;
+            return true;
+        }
+        return false;
+    }
+
+    public static long target_edges(Person person, ArrayList<Float> percentages, int step_index ) {
+        int generated_edges = 0;
+        for (int i = 0; i < step_index; ++i) {
+            generated_edges += Math.ceil(percentages.get(i)*person.maxNumKnows());
+        }
+        generated_edges = Math.min(generated_edges, (int)person.maxNumKnows());
+        int to_generate = Math.min( (int)person.maxNumKnows() - generated_edges, (int)Math.ceil(percentages.get(step_index)*person.maxNumKnows()));
+        return  to_generate;
+    }
 }