Merge branch 'master' of https://github.com/ldbc/ldbc_snb_datagen

Author: Hadoop user

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

@@ -26,13 +26,13 @@
  */
 public class HadoopPersonSerializer {
 
-//	public static class HadoopPersonSerializerReducer  extends Reducer<BlockKey, Person, LongWritable, Person> {
 	public static class HadoopPersonSerializerReducer  extends Reducer<TupleKey, Person, LongWritable, Person> {
 
 		private int reducerId;                          /** The id of the reducer.**/
 		private PersonSerializer personSerializer_;   /** The person serializer **/
-		private UpdateEventSerializer updateSerializer_;   
-		
+		private UpdateEventSerializer updateSerializer_;
+
+		@Override
 		protected void setup(Context context) {
 			Configuration conf = context.getConfiguration();
 			reducerId = context.getTaskAttemptID().getTaskID().getId();
@@ -49,7 +49,6 @@ protected void setup(Context context) {
 		}
 
 		@Override
-	//	public void reduce(BlockKey key, Iterable<Person> valueSet,Context context)
 	public void reduce(TupleKey key, Iterable<Person> valueSet,Context context)
 			throws IOException, InterruptedException {
 //			SN.machineId = key.block;
@@ -70,6 +69,8 @@ public void reduce(TupleKey key, Iterable<Person> valueSet,Context context)
 			}
 
 		}
+
+		@Override
 		protected void cleanup(Context context){
 			personSerializer_.close();
 			if (DatagenParams.updateStreams) {

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

@@ -32,6 +32,7 @@ public static class HadoopPersonSerializerReducer  extends Reducer<BlockKey, Per
 		private PersonSerializer personSerializer_;   /** The person serializer **/
 		private UpdateEventSerializer updateSerializer_;
 
+		@Override
 		protected void setup(Context context) {
 			Configuration conf = context.getConfiguration();
 			reducerId = context.getTaskAttemptID().getTaskID().getId();
@@ -68,6 +69,7 @@ public void reduce(BlockKey key, Iterable<Person> valueSet,Context context)
 			}
 
 		}
+		@Override
 		protected void cleanup(Context context){
 			personSerializer_.close();
 			if (DatagenParams.updateStreams) {

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

@@ -75,6 +75,7 @@ public void writeAllPartitions( String entry ) {
     public void close() {
         try {
             for (int i = 0; i < numPartitions; ++i) {
+                fileOutputStream[i].flush();
                 fileOutputStream[i].close();
             }
         } catch (IOException e) {