Refactored PersonSerializer interface to pass person to export knows relationship api.

Added graphalytics person serializer.

Author: ArnauPrat

src/main/java/ldbc/snb/datagen/dictionary/FlashmobTagDictionary.java

@@ -123,6 +123,7 @@ private void initialize() {
             flashmobTag.tag = tags[i];
             instances.add(flashmobTag);
             flashmobTagCumDist[i] = flashmobTag;
+//            if(tags[i] == 1761) System.out.println(flashmobTag);
         }
         Arrays.sort(flashmobTagCumDist);
         int size = flashmobTagCumDist.length;

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

@@ -43,7 +43,6 @@ public Forum createWall(RandomGeneratorFarm randomFarm, long forumId, Person per
 		return forum;
 	}
 
-
 	public Forum createGroup(RandomGeneratorFarm randomFarm, long forumId, Person person, ArrayList<Person> persons){
 		long date = Dictionaries.dates.randomDate(randomFarm.get(RandomGeneratorFarm.Aspect.DATE), person.creationDate() + DatagenParams.deltaTime);
 		if( date > Dictionaries.dates.getEndDateTime() )  return null;

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

@@ -57,7 +57,7 @@ public void reduce(BlockKey key, Iterable<Person> valueSet,Context context)
 
 				for( Knows k : p.knows() ) {
 					if( k.creationDate() < Dictionaries.dates.getUpdateThreshold() || !DatagenParams.updateStreams ) {
-						personSerializer_.export(p.accountId(), k);
+						personSerializer_.export(p, k);
 					} 
 				}
 			}

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

@@ -36,6 +36,8 @@
  */
 package ldbc.snb.datagen.objects;
 
+import ldbc.snb.datagen.dictionary.Dictionaries;
+
 public class FlashmobTag implements Comparable<FlashmobTag> {
     public int level;
     public long date;
@@ -55,4 +57,8 @@ public void copyTo(FlashmobTag t) {
         t.prob = this.prob;
         t.tag = this.tag;
     }
+
+    public String toString(){
+        return "Level: "+level+" Date: "+Dictionaries.dates.formatDateDetail(date)+" Tag:"+ Dictionaries.tags.getName(tag);
+    }
 }

src/main/java/ldbc/snb/datagen/serializer/PersonSerializer.java

@@ -45,9 +45,9 @@ public void export(Person person) {
         }
     }
 
-    public void export(long personId, Knows k ) {
-        if( personId < k.to().accountId())
-            serialize(personId, k);
+    public void export(Person p, Knows k ) {
+        if( p.accountId() < k.to().accountId())
+            serialize(p, k);
     }
 
     abstract public void initialize(Configuration conf, int reducerId);
@@ -60,5 +60,5 @@ public void export(long personId, Knows k ) {
 
     abstract protected void serialize(WorkAt workAt);
 
-    abstract protected void serialize(long personId, Knows knows);
+    abstract protected void serialize(Person p, Knows knows);
 }

src/main/java/ldbc/snb/datagen/serializer/empty/EmptyPersonSerializer.java

@@ -32,7 +32,7 @@ protected void serialize(WorkAt workAt) {
 
     }
 
-    protected void serialize(long p, Knows knows) {
+    protected void serialize(Person p, Knows knows) {
 
     }
 

src/main/java/ldbc/snb/datagen/serializer/graphalytics/CSVPersonSerializer.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright (c) 2013 LDBC
+ * Linked Data Benchmark Council (http://ldbc.eu)
+ *
+ * This file is part of ldbc_socialnet_dbgen.
+ *
+ * ldbc_socialnet_dbgen is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * ldbc_socialnet_dbgen is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with ldbc_socialnet_dbgen.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * Copyright (C) 2011 OpenLink Software <[email protected]>
+ * All Rights Reserved.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation;  only Version 2 of the License dated
+ * June 1991.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+
+package ldbc.snb.datagen.serializer.graphalytics;
+
+import ldbc.snb.datagen.dictionary.Dictionaries;
+import ldbc.snb.datagen.objects.Knows;
+import ldbc.snb.datagen.objects.Person;
+import ldbc.snb.datagen.objects.StudyAt;
+import ldbc.snb.datagen.objects.WorkAt;
+import ldbc.snb.datagen.serializer.HDFSCSVWriter;
+import ldbc.snb.datagen.serializer.PersonSerializer;
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+
+public class CSVPersonSerializer extends PersonSerializer {
+
+    private HDFSCSVWriter [] writers;
+
+    private enum FileNames {
+        PERSON_KNOWS_PERSON("person_knows_person");
+
+        private final String name;
+
+        private FileNames( String name ) {
+            this.name = name;
+        }
+        public String toString() {
+            return name;
+        }
+    }
+
+    public CSVPersonSerializer() {
+    }
+
+    public void initialize(Configuration conf, int reducerId) {
+        int numFiles = FileNames.values().length;
+        writers = new HDFSCSVWriter[numFiles];
+        for( int i = 0; i < numFiles; ++i) {
+            writers[i] = new HDFSCSVWriter(conf.get("ldbc.snb.datagen.serializer.socialNetworkDir"), FileNames.values()[i].toString()+"_"+reducerId,conf.getInt("ldbc.snb.datagen.numPartitions",1),conf.getBoolean("ldbc.snb.datagen.serializer.compressed",false),"|");
+        }
+
+        ArrayList<String> arguments = new ArrayList<String>();
+        arguments.clear();
+        arguments.add("Person.id");
+        arguments.add("Person.id");
+        arguments.add("Weight");
+        writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeEntry(arguments);
+
+    }
+
+    @Override
+    public void close() {
+        int numFiles = FileNames.values().length;
+        for(int i = 0; i < numFiles; ++i) {
+            writers[i].close();
+        }
+    }
+
+    @Override
+    protected void serialize(Person p) {
+
+    }
+
+    @Override
+    protected void serialize(StudyAt studyAt) {
+    }
+
+    @Override
+    protected void serialize(WorkAt workAt) {
+    }
+
+    protected void serialize(Person p, Knows knows) {
+        ArrayList<String> arguments = new ArrayList<String>();
+        long denominator = knows.creationDate() - Math.min(p.creationDate(),knows.to().creationDate());
+        long numerator = knows.creationDate() - Math.max(p.creationDate(),knows.to().creationDate());
+        double weight = 1.0 - numerator/(double)denominator;
+        arguments.add(Long.toString(p.accountId()));
+        arguments.add(Long.toString(knows.to().accountId()));
+        arguments.add(Double.toString(weight));
+        writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeEntry(arguments);
+    }
+}

src/main/java/ldbc/snb/datagen/serializer/small/CSVPersonSerializer.java

@@ -118,10 +118,11 @@ protected void serialize(StudyAt studyAt) {
     protected void serialize(WorkAt workAt) {
     }
 
-    protected void serialize(long personId, Knows knows) {
+    @Override
+    protected void serialize(Person p, Knows knows) {
         ArrayList<String> arguments = new ArrayList<String>();
 
-        arguments.add(Long.toString(personId));
+        arguments.add(Long.toString(p.accountId()));
         arguments.add(Long.toString(knows.to().accountId()));
         writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeEntry(arguments);
     }

src/main/java/ldbc/snb/datagen/serializer/snb/interactive/CSVMergeForeignPersonSerializer.java

@@ -174,10 +174,11 @@ protected void serialize(WorkAt workAt) {
         writers[FileNames.PERSON_WORK_AT.ordinal()].writeEntry(arguments);
     }
 
-    protected void serialize(long personId, Knows knows) {
+    @Override
+    protected void serialize( Person p, Knows knows) {
         ArrayList<String> arguments = new ArrayList<String>();
         String dateString = Dictionaries.dates.formatDateDetail(knows.creationDate());
-        arguments.add(Long.toString(personId));
+        arguments.add(Long.toString(p.accountId()));
         arguments.add(Long.toString(knows.to().accountId()));
         arguments.add(dateString);
         writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeEntry(arguments);

src/main/java/ldbc/snb/datagen/serializer/snb/interactive/CSVPersonSerializer.java

@@ -218,10 +218,11 @@ protected void serialize(WorkAt workAt) {
         writers[FileNames.PERSON_WORK_AT.ordinal()].writeEntry(arguments);
     }
 
-    protected void serialize(long personId, Knows knows) {
+    @Override
+    protected void serialize(Person p, Knows knows) {
         ArrayList<String> arguments = new ArrayList<String>();
         String dateString = Dictionaries.dates.formatDateDetail(knows.creationDate());
-        arguments.add(Long.toString(personId));
+        arguments.add(Long.toString(p.accountId()));
         arguments.add(Long.toString(knows.to().accountId()));
         arguments.add(dateString);
         writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeEntry(arguments);