Merge branch 'master' of github.com:ldbc/ldbc_socialnet_bm

Author: agubichev

src/main/java/ldbc/socialnet/dbgen/generator/MRGenerateUsers.java

@@ -46,6 +46,7 @@
 import ldbc.socialnet.dbgen.util.*;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -98,17 +99,18 @@ protected void setup(Context context){
                 FileSystem fs = FileSystem.get(conf);
                 String strTaskId = context.getTaskAttemptID().getTaskID().toString();
                 int attempTaskId = Integer.parseInt(strTaskId.substring(strTaskId.length() - 3));
+                int reducerId = conf.getInt("reducerId",0);
                 int partitionId = conf.getInt("partitionId",0);
                 String streamType = conf.get("streamType");
                 if( Boolean.parseBoolean(conf.get("compressed")) == true ) {
-                    Path outFile = new Path(context.getConfiguration().get("outputDir")+"/social_network/updateStream_"+attempTaskId+"_"+partitionId+"_"+streamType+".csv.gz");
+                    Path outFile = new Path(context.getConfiguration().get("outputDir")+"/social_network/updateStream_"+reducerId+"_"+partitionId+"_"+streamType+".csv.gz");
                     out = new GZIPOutputStream( fs.create(outFile));
                 } else {
-                    Path outFile = new Path(context.getConfiguration().get("outputDir")+"/social_network/updateStream_"+attempTaskId+"_"+partitionId+"_"+streamType+".csv");
+                    Path outFile = new Path(context.getConfiguration().get("outputDir")+"/social_network/updateStream_"+reducerId+"_"+partitionId+"_"+streamType+".csv");
                     out = fs.create(outFile);
                 }
                 if (conf.getBoolean("updateStreams",false)) {
-                    properties = fs.create(new Path(context.getConfiguration().get("outputDir")+"/social_network/updateStream_"+attempTaskId+"_"+partitionId+"_"+streamType+".properties"));
+                    properties = fs.create(new Path(context.getConfiguration().get("outputDir")+"/social_network/updateStream_"+reducerId+"_"+partitionId+"_"+streamType+".properties"));
                 }
             } catch (IOException e) {
                 e.printStackTrace();
@@ -552,6 +554,7 @@ public int runGenerateJob(Configuration conf) throws Exception {
             for( int j = 0; j < numPartitions; ++j ) {
                 /// --------------- Fifth job: Sort update streams ----------------
                 conf.setInt("mapred.line.input.format.linespermap", 1000000);
+                conf.setInt("reducerId",i);
                 conf.setInt("partitionId",j);
                 conf.set("streamType","forum");
                 Job jobForum = new Job(conf, "Soring update streams "+j+" of reducer "+i);
@@ -574,6 +577,7 @@ public int runGenerateJob(Configuration conf) throws Exception {
                 fs.delete(new Path(hadoopDir + "/sibEnd"), true);
 
                 conf.setInt("mapred.line.input.format.linespermap", 1000000);
+                conf.setInt("reducerId",i);
                 conf.setInt("partitionId",j);
                 conf.set("streamType","person");
                 Job jobPerson = new Job(conf, "Soring update streams "+j+" of reducer "+i);
@@ -597,21 +601,26 @@ public int runGenerateJob(Configuration conf) throws Exception {
 
                 if(conf.getBoolean("updateStreams",false)) {
                     Properties properties = new Properties();
-                    properties.load(fs.open(new Path(conf.get("outputDir") + "/social_network/updateStream_" + i + "_" + j + "_person.properties")));
-                    Long auxMin = Long.parseLong(properties.getProperty("min_write_event_start_time"));
-                    min = auxMin < min ? auxMin : min;
-                    Long auxMax = Long.parseLong(properties.getProperty("max_write_event_start_time"));
-                    max = auxMax > max ? auxMax : max;
-                    numEvents += Long.parseLong(properties.getProperty("num_events"));
-
-                    properties.load(fs.open(new Path(conf.get("outputDir") + "/social_network/updateStream_" + i + "_" + j + "_forum.properties")));
-
-                    auxMin = Long.parseLong(properties.getProperty("min_write_event_start_time"));
-                    min = auxMin < min ? auxMin : min;
-                    auxMax = Long.parseLong(properties.getProperty("max_write_event_start_time"));
-                    max = auxMax > max ? auxMax : max;
-                    numEvents += Long.parseLong(properties.getProperty("num_events"));
-
+                    FSDataInputStream file = fs.open(new Path(conf.get("outputDir") + "/social_network/updateStream_" + i + "_" + j + "_person.properties"));
+                    properties.load(file);
+                    if( properties.getProperty("min_write_event_start_time") != null ) {
+                        Long auxMin = Long.parseLong(properties.getProperty("min_write_event_start_time"));
+                        min = auxMin < min ? auxMin : min;
+                        Long auxMax = Long.parseLong(properties.getProperty("max_write_event_start_time"));
+                        max = auxMax > max ? auxMax : max;
+                        numEvents += Long.parseLong(properties.getProperty("num_events"));
+                    }
+                    file.close();
+                    file = fs.open(new Path(conf.get("outputDir") + "/social_network/updateStream_" + i + "_" + j + "_forum.properties"));
+                    properties.load(file);
+                    if( properties.getProperty("min_write_event_start_time") != null ) {
+                        Long auxMin = Long.parseLong(properties.getProperty("min_write_event_start_time"));
+                        min = auxMin < min ? auxMin : min;
+                        Long auxMax = Long.parseLong(properties.getProperty("max_write_event_start_time"));
+                        max = auxMax > max ? auxMax : max;
+                        numEvents += Long.parseLong(properties.getProperty("num_events"));
+                    }
+                    file.close();
                     fs.delete(new Path(conf.get("outputDir") + "/social_network/updateStream_" + i + "_" + j + "_person.properties"),true);
                     fs.delete(new Path(conf.get("outputDir") + "/social_network/updateStream_" + i + "_" + j + "_forum.properties"),true);
                 }

src/main/java/ldbc/socialnet/dbgen/serializer/UpdateEventSerializer.java

@@ -426,7 +426,8 @@ public void serialize(Friend friend) {
             data.add(Long.toString(friend.getFriendAcc()));
             date.setTimeInMillis(friend.getCreatedTime());
             data.add(Long.toString(date.getTime().getTime()));
-            endEvent(Stream.PERSON_STREAM);
+            //endEvent(Stream.PERSON_STREAM);
+            endEvent(Stream.FORUM_STREAM);
         }
     }