Number of records logged, small fixes

Author: Maxim Moinat

src/main/java/org/radarcns/restructureAvroRecords.java

@@ -31,15 +31,15 @@ public class restructureAvroRecords {
     private final static SimpleDateFormat dateFormatFileName = new SimpleDateFormat("yyyyMMdd_HH");
 
     private int processedFileCount;
+    private int processedRecordsCount;
 
     public static void main(String [] args) throws Exception {
-
         restructureAvroRecords restr = new restructureAvroRecords(args[0], args[2]);
         long time1 = System.currentTimeMillis();
         restr.start(args[1]);
+        System.out.printf("Processed %d files and %,d records\n", restr.getProcessedFileCount(), restr.getProcessedRecordsCount());
         System.out.printf("Time taken: %.2f seconds\n",(System.currentTimeMillis() - time1)/1000d);
 
-
 //        restructureAvroRecords restr = new restructureAvroRecords("webhdfs://radar-test.thehyve.net:50070", "output4/");
 //        restr.start("/topicE4/");
 
@@ -63,6 +63,14 @@ public void setOutputPath(String path) {
         offsetsPath = outputPath + "/" + OFFSETS_FILE_NAME;
     }
 
+    public int getProcessedFileCount() {
+        return processedFileCount;
+    }
+
+    public int getProcessedRecordsCount() {
+        return processedRecordsCount;
+    }
+
     public void start(String directoryName) throws IOException {
         // Get files and directories
         Path path = new Path(directoryName);
@@ -86,8 +94,6 @@ public void start(String directoryName) throws IOException {
                 processTopic(filePath);
             }
         }
-
-        System.out.printf("%d files processed\n", processedFileCount);
     }
 
     private void processTopic(Path topicPath) throws IOException {
@@ -133,6 +139,7 @@ record = dataFileReader.next(record);
             this.writeRecord(record, topicName);
         }
 
+        // Write which file has been processed and update bins
         this.writeSeenOffsets(fileName);
         bins.writeBins();
         processedFileCount++;
@@ -154,8 +161,9 @@ private void writeRecord(GenericRecord record, String topicName) throws IOExcept
         String data = record.toString(); // TODO: check whether this indeed always creates valid JSON
         this.appendToFile(dirName, outputFileName, data);
 
-        // Count data
+        // Count data (binned and total)
         bins.addToBin(topicName, keyField.get("sourceId").toString(), (Double) valueField.get("time"));
+        processedRecordsCount++;
     }
 
     public static String createHourTimestamp(Double time) {
@@ -230,8 +238,7 @@ private void readSeenOffsets() {
             }
 
         } catch (IOException e) {
-            // TODO
-            e.printStackTrace();
+            System.out.println("No files processed yet.");
         }
     }
 }