Further refactoring, fix Empty and Turtle serializers

Author: szarnyasg

check-md5sums-ttl.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+set -e
+
+[[ `md5sum social_network/social_network_activity_0_0.ttl | cut -d' ' -f1` == '52a5356ecc757a8e3e5bce2c8ea79557' ]]
+[[ `md5sum social_network/social_network_person_0_0.ttl   | cut -d' ' -f1` == '30973a3dc339617773651b425f7441e4' ]]
+[[ `md5sum social_network/social_network_static_0_0.ttl   | cut -d' ' -f1` == '3c4f4120a2ea1e101cf7d72fbfe30c48' ]]
+
+[[ `md5sum social_network/updateStream_0_0_forum.csv      | cut -d' ' -f1` == '7e00243f68a8171974eabe4ac37df86b' ]]
+[[ `md5sum social_network/updateStream_0_0_person.csv     | cut -d' ' -f1` == '2e1f44e6d48112a9fd87092206153b57' ]]

src/main/java/ldbc/snb/datagen/hadoop/writer/HDFSCSVWriter.java

@@ -45,7 +45,6 @@ public class HDFSCSVWriter extends HDFSWriter {
     private StringBuffer buffer;
     private boolean endLineSeparator = true;
 
-
     public HDFSCSVWriter(String outputDir, String prefix, int numPartitions, boolean compressed, String separator, boolean endLineSeparator) throws IOException {
         super(outputDir, prefix, numPartitions, compressed, "csv");
         this.separator = separator;

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

@@ -41,11 +41,24 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import ldbc.snb.datagen.entities.dynamic.messages.Post;
 import ldbc.snb.datagen.entities.dynamic.relations.ForumMembership;
 import ldbc.snb.datagen.entities.dynamic.relations.Like;
+import ldbc.snb.datagen.hadoop.writer.HDFSWriter;
 
 /**
  * @author aprat
  */
-abstract public class DynamicActivitySerializer extends LDBCSerializer {
+abstract public class DynamicActivitySerializer<TWriter extends HDFSWriter> extends LDBCSerializer<TWriter> {
+
+    abstract protected void serialize(final Forum forum);
+
+    abstract protected void serialize(final Post post);
+
+    abstract protected void serialize(final Comment comment);
+
+    abstract protected void serialize(final Photo photo);
+
+    abstract protected void serialize(final ForumMembership membership);
+
+    abstract protected void serialize(final Like like);
 
     public void export(final Forum forum) {
         serialize(forum);
@@ -59,32 +72,11 @@ public void export(final Post post) {
         serialize(post);
     }
 
-    public void export(Comment comment) {
-        serialize(comment);
-
-    }
-
-    public void export(Photo photo) {
-        serialize(photo);
+    public void export(Comment comment) { serialize(comment); }
 
-    }
-
-    public void export(Like like) {
-        serialize(like);
-
-    }
-
-    abstract protected void serialize(final Forum forum);
-
-    abstract protected void serialize(final Post post);
+    public void export(Photo photo) { serialize(photo); }
 
-    abstract protected void serialize(final Comment comment);
-
-    abstract protected void serialize(final Photo photo);
-
-    abstract protected void serialize(final ForumMembership membership);
-
-    abstract protected void serialize(final Like like);
+    public void export(Like like) { serialize(like); }
 
     @Override
     protected boolean isDynamic() {

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

@@ -41,6 +41,7 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import ldbc.snb.datagen.entities.dynamic.relations.Knows;
 import ldbc.snb.datagen.entities.dynamic.relations.StudyAt;
 import ldbc.snb.datagen.entities.dynamic.relations.WorkAt;
+import ldbc.snb.datagen.hadoop.writer.HDFSWriter;
 
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -50,10 +51,17 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 /**
  * Created by aprat on 10/15/14.
  */
-abstract public class DynamicPersonSerializer extends LDBCSerializer {
+abstract public class DynamicPersonSerializer<TWriter extends HDFSWriter> extends LDBCSerializer<TWriter> {
 
-    public void export(final Person person) {
+    abstract protected void serialize(final Person p);
 
+    abstract protected void serialize(final StudyAt studyAt);
+
+    abstract protected void serialize(final WorkAt workAt);
+
+    abstract protected void serialize(final Person p, final Knows knows);
+
+    public void export(final Person person) {
         serialize(person);
 
         long universityId = Dictionaries.universities.getUniversityFromLocation(person.universityLocationId());
@@ -99,14 +107,6 @@ public String buildEmail(TreeSet<String> emails) {
         return Joiner.on(";").join(emails);
     }
 
-    abstract protected void serialize(final Person p);
-
-    abstract protected void serialize(final StudyAt studyAt);
-
-    abstract protected void serialize(final WorkAt workAt);
-
-    abstract protected void serialize(final Person p, final Knows knows);
-
     @Override
     protected boolean isDynamic() {
         return true;

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

@@ -1,32 +1,23 @@
 package ldbc.snb.datagen.serializer;
 
-import ldbc.snb.datagen.hadoop.writer.HDFSCSVWriter;
+import ldbc.snb.datagen.hadoop.writer.HDFSWriter;
 import ldbc.snb.datagen.serializer.snb.csv.FileName;
 import org.apache.hadoop.conf.Configuration;
 
 import java.io.IOException;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-abstract public class LDBCSerializer {
+abstract public class LDBCSerializer<TWriter extends HDFSWriter> implements Serializer<TWriter> {
 
-    protected Map<FileName, HDFSCSVWriter> writers = new HashMap<>();
+    protected Map<FileName, TWriter> writers;
 
     abstract public List<FileName> getFileNames();
 
     abstract public void writeFileHeaders();
 
     public void initialize(Configuration conf, int reducerId) throws IOException {
-        for (FileName f : getFileNames()) {
-            writers.put(f, new HDFSCSVWriter(
-                    conf.get("ldbc.snb.datagen.serializer.socialNetworkDir") + (isDynamic() ? "/dynamic/" : "/static/"),
-                    f.toString() + "_" + reducerId,
-                    conf.getInt("ldbc.snb.datagen.numPartitions", 1),
-                    conf.getBoolean("ldbc.snb.datagen.serializer.compressed", false), "|",
-                    conf.getBoolean("ldbc.snb.datagen.serializer.endlineSeparator", false))
-            );
-        }
+        writers = initialize(conf, reducerId, isDynamic(), getFileNames());
         writeFileHeaders();
     }
 

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

@@ -0,0 +1,15 @@
+package ldbc.snb.datagen.serializer;
+
+import ldbc.snb.datagen.hadoop.writer.HDFSWriter;
+import ldbc.snb.datagen.serializer.snb.csv.FileName;
+import org.apache.hadoop.conf.Configuration;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public interface Serializer<THDFSWriter extends HDFSWriter> {
+
+    Map<FileName, THDFSWriter> initialize(Configuration conf, int reducerId, boolean dynamic, List<FileName> fileNames) throws IOException;
+
+}

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

@@ -39,12 +39,20 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import ldbc.snb.datagen.entities.statictype.TagClass;
 import ldbc.snb.datagen.entities.statictype.place.Place;
 import ldbc.snb.datagen.entities.statictype.tag.Tag;
-
+import ldbc.snb.datagen.hadoop.writer.HDFSWriter;
 
 /**
  * Created by aprat on 12/17/14.
  */
-abstract public class StaticSerializer extends LDBCSerializer {
+abstract public class StaticSerializer<TWriter extends HDFSWriter> extends LDBCSerializer<TWriter> {
+
+    abstract protected void serialize(final Place place);
+
+    abstract protected void serialize(final Organisation organisation);
+
+    abstract protected void serialize(final TagClass tagClass);
+
+    abstract protected void serialize(final Tag tag);
 
     public void export(final TagClass tagclass) {
         serialize(tagclass);
@@ -62,14 +70,6 @@ public void export(final Tag tag) {
         serialize(tag);
     }
 
-    abstract protected void serialize(final Place place);
-
-    abstract protected void serialize(final Organisation organisation);
-
-    abstract protected void serialize(final TagClass tagClass);
-
-    abstract protected void serialize(final Tag tag);
-
     @Override
     protected boolean isDynamic() {
         return false;

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

@@ -46,7 +46,9 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import org.apache.hadoop.conf.Configuration;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Created by aprat on 30/01/15.
@@ -103,4 +105,9 @@ protected void serialize(final Like like) {
         //This is left intentionally blank
     }
 
+    @Override
+    public Map initialize(Configuration conf, int reducerId, boolean dynamic, List list) throws IOException {
+        return Collections.emptyMap();
+    }
+
 }

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

@@ -44,7 +44,9 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import org.apache.hadoop.conf.Configuration;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Created by aprat on 30/01/15.
@@ -91,4 +93,9 @@ protected void serialize(final Person p, final Knows knows) {
         //Intentionally left empty
     }
 
+    @Override
+    public Map initialize(Configuration conf, int reducerId, boolean dynamic, List list) throws IOException {
+        return Collections.emptyMap();
+    }
+
 }

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

@@ -35,14 +35,20 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.*/
 package ldbc.snb.datagen.serializer.empty;
 
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
 import ldbc.snb.datagen.entities.statictype.Organisation;
 import ldbc.snb.datagen.entities.statictype.TagClass;
 import ldbc.snb.datagen.entities.statictype.place.Place;
 import ldbc.snb.datagen.entities.statictype.tag.Tag;
 import ldbc.snb.datagen.serializer.StaticSerializer;
 import ldbc.snb.datagen.serializer.snb.csv.FileName;
+import org.apache.hadoop.conf.Configuration;
 
+import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Created by aprat on 30/01/15.
@@ -83,4 +89,9 @@ protected void serialize(final Tag tag) {
 
     }
 
+    @Override
+    public Map initialize(Configuration conf, int reducerId, boolean dynamic, List list) throws IOException {
+        return Collections.emptyMap();
+    }
+
 }