Skip already processed files

Author: Maxim Moinat

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

@@ -18,31 +18,44 @@
 
 public class restructureAvroRecords {
 
-    String outputPath = ".";
-    String OUTPUT_FILE_EXTENSION = "json";
-    String OFFSETS_FILE_NAME = "offsets.csv";
-    Configuration conf = new Configuration();
-    SimpleDateFormat dateFormatFileName = new SimpleDateFormat("yyyyMMdd_HH");
+    private final String OUTPUT_FILE_EXTENSION = "json";
+    private final String OFFSETS_FILE_NAME = "offsets.csv";
+
+    private String outputPath = ".";
+    private String offsetsPath = outputPath + "/" + OFFSETS_FILE_NAME;
+    private Set<String> seenFiles = new HashSet<>();
+
+    private Configuration conf = new Configuration();
+    private final SimpleDateFormat dateFormatFileName = new SimpleDateFormat("yyyyMMdd_HH");
+
+    private int processedFileCount;
 
     public static void main(String [] args) throws Exception {
 
-        restructureAvroRecords restr = new restructureAvroRecords();
-        restr.setInputWebHdfsURL(args[0]); //"webhdfs://radar-test.thehyve.net:50070");
-        restr.setOutputPath(args[2]); //"output3/");
+//        restructureAvroRecords restr = new restructureAvroRecords(args[0], args[2]);
+//        restr.start(args[1]);
+
+        restructureAvroRecords restr = new restructureAvroRecords("webhdfs://radar-test.thehyve.net:50070", "output3/");
+        restr.start("/topicAndroidPhoneNew/");
 
-        restr.start(args[1]); //"/topicAndroidPhoneNew/");
 //        restructureAvroRecords.processTopic("/topicE4/android_empatica_e4_inter_beat_interval/partition=0/");
 //        restructureAvroRecords.processAvroFile(new Path("/topicE4/android_empatica_e4_inter_beat_interval/partition=0/android_empatica_e4_inter_beat_interval+0+0000031485+0000031488.avro") );
 //        restructureAvroRecords.processAvroFile(new Path("/testE4Time/android_phone_acceleration/partition=0/android_phone_acceleration+0+0000590000+0000599999.avro"),"wazaa" );
     }
 
+    public restructureAvroRecords(String inputPath, String outputPath) {
+        this.setInputWebHdfsURL(inputPath);
+        this.setOutputPath(outputPath);
+    }
+
     public void setInputWebHdfsURL(String fileSystemURL) {
-        conf.set("fs.defaultFS",fileSystemURL);
+        conf.set("fs.defaultFS", fileSystemURL);
     }
 
     public void setOutputPath(String path) {
         // Remove trailing backslash
         outputPath = path.replaceAll("/$","");
+        offsetsPath = outputPath + "/" + OFFSETS_FILE_NAME;
     }
 
     public void start(String directoryName) throws IOException {
@@ -51,7 +64,11 @@ public void start(String directoryName) throws IOException {
         FileSystem fs = FileSystem.get(conf);
         RemoteIterator<LocatedFileStatus> files = fs.listLocatedStatus(path);
 
+        // Load seen offsets from file
+        readSeenOffsets();
+
         // Process the directories topics
+        processedFileCount = 0;
         while (files.hasNext()) {
             LocatedFileStatus locatedFileStatus = files.next();
             Path filePath = locatedFileStatus.getPath();
@@ -64,6 +81,7 @@ public void start(String directoryName) throws IOException {
                 processTopic(filePath);
             }
         }
+        System.out.printf("%d files processed\n", processedFileCount);
     }
 
     public void processTopic(Path topicPath) throws IOException {
@@ -80,6 +98,7 @@ public void processTopic(Path topicPath) throws IOException {
 
             if (locatedFileStatus.isFile())
                 this.processAvroFile( locatedFileStatus.getPath(), topicName );
+
         }
     }
 
@@ -88,7 +107,12 @@ public void processAvroFile(Path filePath, String topicName) throws IOException
 
         // Skip if extension is not .avro
         if (! FilenameUtils.getExtension(fileName).equals("avro")) {
-            System.out.printf("Skipped non avro file: %s\n", filePath.getName());
+            System.out.printf("Skipped non avro file: %s\n", fileName);
+            return;
+        }
+
+        // Skip already processed avro files
+        if (seenFiles.contains(fileName)) {
             return;
         }
 
@@ -106,6 +130,7 @@ record = dataFileReader.next(record);
         }
 
         this.writeSeenOffsets(fileName);
+        processedFileCount++;
     }
 
     public void writeRecord(GenericRecord record, String topicName) throws IOException {
@@ -139,8 +164,10 @@ public String createFilePathFromTimestamp(Double time) {
     public void appendToFile(String directoryName, String fileName, String data) {
         File directory = new File(directoryName);
         if (! directory.exists()){
-            directory.mkdirs();
-            System.out.printf("Created directory: %s\n", directory.getAbsolutePath());
+            if (directory.mkdirs())
+                System.out.printf("Created directory: %s\n", directory.getAbsolutePath());
+            else
+                System.out.printf("FAILED to create directory: %s\n", directory.getAbsolutePath());
         }
 
         String filePath = directoryName + "/" + fileName;
@@ -175,9 +202,9 @@ private void writeSeenOffsets(String fileName) {
             return;
         }
 
-        String data = String.join(",", topicName, partition, fromOffset.toString(), toOffset.toString());
+        String data = String.join(",", fileName, topicName, partition, fromOffset.toString(), toOffset.toString());
 
-        try(FileWriter fw = new FileWriter(outputPath + "/" + OFFSETS_FILE_NAME, true);
+        try(FileWriter fw = new FileWriter(offsetsPath, true);
             BufferedWriter bw = new BufferedWriter(fw);
             PrintWriter out = new PrintWriter(bw))
         {
@@ -187,4 +214,23 @@ private void writeSeenOffsets(String fileName) {
             e.printStackTrace();
         }
     }
+
+    private void readSeenOffsets() {
+
+        try (FileReader fr = new FileReader(offsetsPath);
+             BufferedReader br = new BufferedReader(fr))
+        {
+            // Read in all file names from csv
+            String line;
+            while ( (line = br.readLine()) != null ) {
+                String[] columns = line.split(",");
+                seenFiles.add(columns[0]);
+            }
+
+        } catch (IOException e) {
+            // TODO
+            e.printStackTrace();
+        }
+
+    }
 }