Refactored CSVCompositeMergeForeignDynamicPersonSerializer and fixed typo in LDBCDatagen

Author: miratepuffin

params-csv.ini

@@ -5,4 +5,4 @@ ldbc.snb.datagen.serializer.staticSerializer:ldbc.snb.datagen.serializer.snb.csv
 ldbc.snb.datagen.serializer.dynamicActivitySerializer:ldbc.snb.datagen.serializer.snb.csv.dynamicserializer.activity.CSVDynamicActivitySerializer
 
 # To generate RFC 3339-compliant timestamps (https://tools.ietf.org/html/rfc3339), uncomment the following line:
-#ldbc.snb.datagen.serializer.formatter.StringDateFormatter.dateTimeFormat:yyyy-MM-dd'T'HH:mm:ss.SSS+00:00
+#ldbc.snb.datagen.util.formatter.StringDateFormatter.dateTimeFormat:yyyy-MM-dd'T'HH:mm:ss.SSS+00:00

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

@@ -40,6 +40,7 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import ldbc.snb.datagen.hadoop.generator.HadoopPersonActivityGenerator;
 import ldbc.snb.datagen.hadoop.generator.HadoopPersonGenerator;
 import ldbc.snb.datagen.hadoop.serializer.HadoopPersonSerializer;
+import ldbc.snb.datagen.hadoop.serializer.HadoopPersonSortAndSerializer;
 import ldbc.snb.datagen.hadoop.serializer.HadoopStaticSerializer;
 import ldbc.snb.datagen.hadoop.serializer.HadoopUpdateStreamSorterAndSerializer;
 import ldbc.snb.datagen.hadoop.miscjob.HadoopMergeFriendshipFiles;
@@ -157,7 +158,7 @@ public int runGenerateJob(Configuration conf) throws Exception {
             HadoopPersonSortAndSerializer serializer = new HadoopPersonSortAndSerializer(conf);
             serializer.run(hadoopPrefix + "/mergedPersons");
         } else {
-            HadoopDynamicPersonSerializer serializer = new HadoopDynamicPersonSerializer(conf);
+            HadoopPersonSerializer serializer = new HadoopPersonSerializer(conf);
             serializer.run(hadoopPrefix + "/mergedPersons");
         }
         long endPersonSerializing = System.currentTimeMillis();

src/main/java/ldbc/snb/datagen/serializer/snb/csv/dynamicserializer/person/CSVCompositeMergeForeignDynamicPersonSerializer.java

@@ -35,6 +35,7 @@ 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.snb.csv.dynamicserializer.person;
 
+import com.google.common.collect.ImmutableList;
 import ldbc.snb.datagen.dictionary.Dictionaries;
 import ldbc.snb.datagen.entities.dynamic.relations.Knows;
 import ldbc.snb.datagen.entities.dynamic.person.Person;
@@ -47,183 +48,67 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import ldbc.snb.datagen.serializer.snb.csv.FileName;
 
+import static ldbc.snb.datagen.serializer.snb.csv.FileName.*;
 /**
  * Created by aprat on 17/02/15.
  */
 public class CSVCompositeMergeForeignDynamicPersonSerializer extends DynamicPersonSerializer {
-
-    private HDFSCSVWriter[] writers;
-
-    private enum FileNames {
-        PERSON("person"),
-        PERSON_HAS_INTEREST_TAG("person_hasInterest_tag"),
-        PERSON_WORK_AT("person_workAt_organisation"),
-        PERSON_STUDY_AT("person_studyAt_organisation"),
-        PERSON_KNOWS_PERSON("person_knows_person");
-
-        private final String name;
-
-        private FileNames(String name) {
-            this.name = name;
-        }
-
-        public String toString() {
-            return name;
-        }
-    }
-
     @Override
     public List<FileName> getFileNames() {
-        return null;
+        return Arrays.asList(PERSON,PERSON_HAS_INTEREST_TAG,PERSON_WORK_AT,PERSON_STUDY_AT,PERSON_KNOWS_PERSON);
     }
 
     @Override
     public void writeFileHeaders() {
-
-    }
-
-    public void initialize(Configuration conf, int reducerId) throws IOException {
-        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")+"/dynamic/", FileNames
-                    .values()[i].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));
-        }
-
-        ArrayList<String> arguments = new ArrayList<String>();
-        arguments.add("id");
-        arguments.add("firstName");
-        arguments.add("lastName");
-        arguments.add("gender");
-        arguments.add("birthday");
-        arguments.add("creationDate");
-        arguments.add("locationIP");
-        arguments.add("browserUsed");
-        arguments.add("place");
-        arguments.add("language");
-        arguments.add("email");
-        writers[FileNames.PERSON.ordinal()].writeHeader(arguments);
-
-        arguments.clear();
-        arguments.add("Person.id");
-        arguments.add("Tag.id");
-        writers[FileNames.PERSON_HAS_INTEREST_TAG.ordinal()].writeHeader(arguments);
-
-        arguments.clear();
-        arguments.add("Person.id");
-        arguments.add("Organisation.id");
-        arguments.add("workFrom");
-        writers[FileNames.PERSON_WORK_AT.ordinal()].writeHeader(arguments);
-
-        arguments.clear();
-        arguments.add("Person.id");
-        arguments.add("Organisation.id");
-        arguments.add("classYear");
-        writers[FileNames.PERSON_STUDY_AT.ordinal()].writeHeader(arguments);
-
-        arguments.clear();
-        arguments.add("Person.id");
-        arguments.add("Person.id");
-        arguments.add("creationDate");
-        writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeHeader(arguments);
-
-    }
-
-    @Override
-    public void close() {
-        int numFiles = FileNames.values().length;
-        for (int i = 0; i < numFiles; ++i) {
-            writers[i].close();
-        }
+        writers.get(PERSON).writeHeader(ImmutableList.of("id","firstName","lastName","gender","birthday","creationDate","locationIP","browserUsed","place","language","email"));
+        writers.get(PERSON_HAS_INTEREST_TAG).writeHeader(ImmutableList.of("Person.id","Tag.id"));
+        writers.get(PERSON_WORK_AT).writeHeader(ImmutableList.of("Person.id","Organisation.id","workFrom"));
+        writers.get(PERSON_STUDY_AT).writeHeader(ImmutableList.of("Person.id","Organisation.id","classYear"));
+        writers.get(PERSON_KNOWS_PERSON).writeHeader(ImmutableList.of("Person.id","Person.id","creationDate"));
     }
 
     @Override
     protected void serialize(final Person p) {
 
-        ArrayList<String> arguments = new ArrayList<String>();
-
-        arguments.add(Long.toString(p.accountId()));
-        arguments.add(p.firstName());
-        arguments.add(p.lastName());
-        if (p.gender() == 1) {
-            arguments.add("male");
-        } else {
-            arguments.add("female");
-        }
-
-        String dateString = Dictionaries.dates.formatDate(p.birthday());
-        arguments.add(dateString);
-
-        dateString = Dictionaries.dates.formatDateTime(p.creationDate());
-        arguments.add(dateString);
-        arguments.add(p.ipAddress().toString());
-        arguments.add(Dictionaries.browsers.getName(p.browserId()));
-        arguments.add(Integer.toString(p.cityId()));
-
-        ArrayList<Integer> languages = p.languages();
-        StringBuilder languagesBuilder = new StringBuilder();
-        for (int i = 0; i < languages.size()-1; i++) {
-            languagesBuilder.append(Dictionaries.languages.getLanguageName(languages.get(i))+";");
-        }
-        if(languages.size() > 0) {
-            languagesBuilder.append(Dictionaries.languages.getLanguageName(languages.get(languages.size()-1)));
-        }
-        arguments.add(languagesBuilder.toString());
-
-        StringBuilder emailsBuilder = new StringBuilder();
-        Iterator<String> itString = p.emails().iterator();
-        for (int i = 0; i < p.emails().size()-1; i++) {
-            emailsBuilder.append(itString.next()+";");
-        }
-        if(itString.hasNext()) {
-            emailsBuilder.append(itString.next());
-        }
-        arguments.add(emailsBuilder.toString());
-        writers[FileNames.PERSON.ordinal()].writeEntry(arguments);
+        writers.get(PERSON).writeEntry(ImmutableList.of(Long.toString(
+                p.accountId()),
+                p.firstName(),
+                p.lastName(),
+                getGender(p.gender()),
+                Dictionaries.dates.formatDate(p.birthday()),
+                Dictionaries.dates.formatDateTime(p.creationDate()),
+                p.ipAddress().toString(),
+                Dictionaries.browsers.getName(p.browserId()),
+                Integer.toString(p.cityId()),
+                buildLanguages(p.languages()),
+                buildEmail(p.emails())
+                ));
 
         Iterator<Integer> itInteger = p.interests().iterator();
         while (itInteger.hasNext()) {
-            arguments.clear();
             Integer interestIdx = itInteger.next();
-            arguments.add(Long.toString(p.accountId()));
-            arguments.add(Integer.toString(interestIdx));
-            writers[FileNames.PERSON_HAS_INTEREST_TAG.ordinal()].writeEntry(arguments);
+            writers.get(PERSON_HAS_INTEREST_TAG).writeEntry(ImmutableList.of(Long.toString(p.accountId()),Integer.toString(interestIdx)));
         }
     }
 
     @Override
     protected void serialize(final StudyAt studyAt) {
-        ArrayList<String> arguments = new ArrayList<String>();
-        String dateString = Dictionaries.dates.formatYear(studyAt.year);
-        arguments.add(Long.toString(studyAt.user));
-        arguments.add(Long.toString(studyAt.university));
-        arguments.add(dateString);
-        writers[FileNames.PERSON_STUDY_AT.ordinal()].writeEntry(arguments);
+        writers.get(PERSON_STUDY_AT).writeEntry(ImmutableList.of(Long.toString(studyAt.user),Long.toString(studyAt.university),Dictionaries.dates.formatYear(studyAt.year)));
     }
 
     @Override
     protected void serialize(final WorkAt workAt) {
-        ArrayList<String> arguments = new ArrayList<String>();
-        String dateString = Dictionaries.dates.formatYear(workAt.year);
-        arguments.add(Long.toString(workAt.user));
-        arguments.add(Long.toString(workAt.company));
-        arguments.add(dateString);
-        writers[FileNames.PERSON_WORK_AT.ordinal()].writeEntry(arguments);
+        writers.get(PERSON_WORK_AT).writeEntry(ImmutableList.of(Long.toString(workAt.user), Long.toString(workAt.company), Dictionaries.dates.formatYear(workAt.year)));
     }
 
     @Override
     protected void serialize(final Person p, Knows knows) {
-        ArrayList<String> arguments = new ArrayList<String>();
-        String dateString = Dictionaries.dates.formatDateTime(knows.creationDate());
-        arguments.add(Long.toString(p.accountId()));
-        arguments.add(Long.toString(knows.to().accountId()));
-        arguments.add(dateString);
-        writers[FileNames.PERSON_KNOWS_PERSON.ordinal()].writeEntry(arguments);
+        writers.get(PERSON_KNOWS_PERSON).writeEntry(ImmutableList.of(Long.toString(p.accountId()),Long.toString(knows.to().accountId()),Dictionaries.dates.formatDateTime(knows.creationDate())));
     }
 
 }

src/main/java/ldbc/snb/datagen/util/ConfigParser.java

@@ -82,9 +82,9 @@ public static Configuration initialize() throws Exception {
         conf.set("ldbc.snb.datagen.serializer.endlineSeparator", Boolean.toString(false));
         conf.set("ldbc.snb.datagen.generator.deltaTime", "10000");
         conf.set("ldbc.snb.datagen.generator.activity", "true");
-        conf.set("ldbc.snb.datagen.serializer.dateFormatter", "ldbc.snb.datagen.serializer.formatter.StringDateFormatter");
-        conf.set("ldbc.snb.datagen.serializer.formatter.StringDateFormatter.dateTimeFormat", "yyyy-MM-dd'T'HH:mm:ss.SSSZ");
-        conf.set("ldbc.snb.datagen.serializer.formatter.StringDateFormatter.dateFormat", "yyyy-MM-dd");
+        conf.set("ldbc.snb.datagen.serializer.dateFormatter", "ldbc.snb.datagen.util.formatter.StringDateFormatter");
+        conf.set("ldbc.snb.datagen.util.formatter.StringDateFormatter.dateTimeFormat", "yyyy-MM-dd'T'HH:mm:ss.SSSZ");
+        conf.set("ldbc.snb.datagen.util.formatter.StringDateFormatter.dateFormat", "yyyy-MM-dd");
         conf.set("ldbc.snb.datagen.generator.person.similarity", "ldbc.snb.datagen.entities.dynamic.person.similarity.GeoDistanceSimilarity");
         conf.set("ldbc.snb.datagen.parametergenerator.python", "python");
         conf.set("ldbc.snb.datagen.parametergenerator.parameters", "true");

src/main/java/ldbc/snb/datagen/util/DateUtils.java

@@ -38,7 +38,7 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
 import ldbc.snb.datagen.DatagenParams;
 import ldbc.snb.datagen.generator.tools.PowerDistribution;
 import ldbc.snb.datagen.entities.dynamic.person.Person;
-import ldbc.snb.datagen.serializer.formatter.DateFormatter;
+import ldbc.snb.datagen.util.formatter.DateFormatter;
 import org.apache.hadoop.conf.Configuration;
 
 import java.util.*;

src/main/java/ldbc/snb/datagen/util/formatter/DateFormatter.java

@@ -33,7 +33,7 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
  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.formatter;
+package ldbc.snb.datagen.util.formatter;
 
 import org.apache.hadoop.conf.Configuration;
 

src/main/java/ldbc/snb/datagen/util/formatter/LongDateFormatter.java

@@ -33,7 +33,7 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
  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.formatter;
+package ldbc.snb.datagen.util.formatter;
 
 import org.apache.hadoop.conf.Configuration;
 

src/main/java/ldbc/snb/datagen/util/formatter/StringDateFormatter.java

@@ -33,7 +33,7 @@ Linked Data Benchmark Council (http://www.ldbcouncil.org)
  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.formatter;
+package ldbc.snb.datagen.util.formatter;
 
 import org.apache.hadoop.conf.Configuration;
 
@@ -56,11 +56,11 @@ public class StringDateFormatter implements DateFormatter {
     public void initialize(Configuration conf) {
 
         formatDateTimeString_ = conf
-                .get("ldbc.snb.datagen.serializer.formatter.StringDateFormatter.dateTimeFormat", formatDateTimeString_);
+                .get("ldbc.snb.datagen.util.formatter.StringDateFormatter.dateTimeFormat", formatDateTimeString_);
         gmtDateTimeFormatter_ = new SimpleDateFormat(formatDateTimeString_);
         gmtDateTimeFormatter_.setTimeZone(TimeZone.getTimeZone("GMT"));
         formatDateString_ = conf
-                .get("ldbc.snb.datagen.serializer.formatter.StringDateFormatter.dateFormat", formatDateString_);
+                .get("ldbc.snb.datagen.util.formatter.StringDateFormatter.dateFormat", formatDateString_);
         gmtDateFormatter_ = new SimpleDateFormat(formatDateString_);
         gmtDateFormatter_.setTimeZone(TimeZone.getTimeZone("GMT"));
         date_ = new Date();