Some code simplification

Author: blootsvoets

src/main/java/org/radarcns/Frequency.java

@@ -16,6 +16,9 @@
 
 package org.radarcns;
 
+import java.nio.file.Files;
+import java.util.Objects;
+import javax.annotation.Nonnull;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.commons.collections.MapIterator;
@@ -30,38 +33,46 @@
 public class Frequency {
     private static final Logger logger = LoggerFactory.getLogger(Frequency.class);
 
-    private MultiKeyMap bins = new MultiKeyMap();
-    private File binFilePath;
+    private final MultiKeyMap bins;
+    private final File file;
 
-    public void setBinFilePath(File binFilePath) {
-        this.binFilePath = binFilePath;
+    public Frequency(@Nonnull File file, @Nonnull MultiKeyMap initialData) {
+        Objects.requireNonNull(file);
+        Objects.requireNonNull(initialData);
+        this.file = file;
+        this.bins = initialData;
     }
 
-    public MultiKeyMap getBins() {
-        return bins;
+    public static Frequency read(File file) {
+        MultiKeyMap map = new MultiKeyMap();
+        try {
+            // Read in all lines as multikeymap (key, key, key, value)
+            Files.readAllLines(file.toPath()).forEach(line -> {
+                String[] columns = line.split(",");
+                try {
+                    map.put(columns[0], columns[1], columns[2], Integer.valueOf(columns[3]));
+                } catch (ArrayIndexOutOfBoundsException ex) {
+                    logger.warn("Unable to row of the bins file. Skipping.");
+                }
+            });
+        } catch (IOException e) {
+            logger.warn("Could not read the file with bins. Creating new file when writing.");
+        }
+        return new Frequency(file, map);
     }
 
-    public void addToBin(String topicName, String id, String timestamp, int countToAdd) {
+    public void add(String topicName, String id, GenericRecord valueField, Field timeField) {
+        String timestamp = RestructureAvroRecords.createHourTimestamp(valueField, timeField);
+
         Integer count = (Integer) bins.get(topicName, id, timestamp);
         if (count == null) {
-            bins.put(topicName, id, timestamp, countToAdd);
+            bins.put(topicName, id, timestamp, 1);
         } else {
-            bins.put(topicName, id, timestamp, count + countToAdd);
+            bins.put(topicName, id, timestamp, count + 1);
         }
     }
 
-    public void addToBin(String topicName, String id, GenericRecord valueField, Field timeField, int countToAdd) {
-        // Hour resolution
-        String hourlyTimestamp = RestructureAvroRecords.createHourTimestamp(valueField, timeField);
-
-        addToBin(topicName, id, hourlyTimestamp, countToAdd);
-    }
-
-    public void addToBin(String topicName, String id, GenericRecord valueField, Field timeField) {
-        addToBin(topicName, id, valueField, timeField, 1);
-    }
-
-    public void printBins() {
+    public void print() {
         MapIterator mapIterator = bins.mapIterator();
 
         while (mapIterator.hasNext()) {
@@ -71,50 +82,24 @@ public void printBins() {
         }
     }
 
-    public void writeBins() {
-        // Read bins from file and add to current bins
-        // Creates new bins if not existing yet
-        addBinsFromFile();
-
+    public void write() {
         // Write all bins to csv
         MapIterator mapIterator = bins.mapIterator();
-        try(FileWriter fw = new FileWriter(binFilePath, false);
-            BufferedWriter bw = new BufferedWriter(fw);
-            PrintWriter out = new PrintWriter(bw))
-        {
+        try (FileWriter fw = new FileWriter(file, false);
+                BufferedWriter bw = new BufferedWriter(fw)) {
             String header = String.join(",","topic","device","timestamp","count");
-            out.println(header);
+            bw.write(header);
+            bw.write('\n');
 
             while (mapIterator.hasNext()) {
                 MultiKey key = (MultiKey) mapIterator.next();
                 Integer value = (Integer) mapIterator.getValue();
                 String data = String.join(",", key.getKey(0).toString(), key.getKey(1).toString(), key.getKey(2).toString(), value.toString());
-                out.println(data);
-            }
-        } catch (IOException e) {
-            // TODO: exception handling
-            e.printStackTrace();
-        }
-
-        // Reset the map
-        bins = new MultiKeyMap();
-    }
-
-    private void addBinsFromFile() {
-        try (FileReader fr = new FileReader(binFilePath);
-             BufferedReader br = new BufferedReader(fr))
-        {
-            // Read in all lines as multikeymap (key, key, key, value)
-            String line;
-            br.readLine(); // Skip header
-            while ( (line = br.readLine()) != null ) {
-                String[] columns = line.split(",");
-                this.addToBin(columns[0], columns[1], columns[2], Integer.valueOf(columns[3]));
+                bw.write(data);
+                bw.write('\n');
             }
         } catch (IOException e) {
-            logger.warn("Could not read the file with bins. Creating new file when writing.");
-        } catch (ArrayIndexOutOfBoundsException e) {
-            logger.warn("Unable to parse the contents of the bins file. Skipping reading.");
+            logger.error("Failed to write bins", e);
         }
     }
 }

src/main/java/org/radarcns/RestructureAvroRecords.java

@@ -24,7 +24,6 @@
 import org.apache.avro.file.DataFileReader;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.io.DatumReader;
 import org.apache.avro.mapred.FsInput;
 import org.apache.commons.io.FilenameUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -52,16 +51,14 @@ public class RestructureAvroRecords {
     private File outputPath;
     private File offsetsPath;
     private OffsetRangeSet seenFiles;
-    private final Frequency bins = new Frequency();
+    private Frequency bins;
 
     private final Configuration conf = new Configuration();
-    private DatumReader<GenericRecord> datumReader;
 
     private int processedFileCount;
     private int processedRecordsCount;
 
     public static void main(String [] args) throws Exception {
-//        Thread.sleep(120_000);
         if (args.length != 3) {
             System.out.println("Usage: hadoop jar restructurehdfs-all-0.1.0.jar <webhdfs_url> <hdfs_topic> <output_folder>");
             System.exit(1);
@@ -106,7 +103,7 @@ public void setOutputPath(String path) {
         // Remove trailing backslash
         outputPath = new File(path.replaceAll("/$",""));
         offsetsPath = new File(outputPath, OFFSETS_FILE_NAME);
-        bins.setBinFilePath(new File(outputPath, BINS_FILE_NAME));
+        bins = Frequency.read(new File(outputPath, BINS_FILE_NAME));
     }
 
     public int getProcessedFileCount() {
@@ -154,7 +151,6 @@ private void processTopic(Path topicPath, RecordConverterFactory converterFactor
         RemoteIterator<LocatedFileStatus> files = fs.listFiles(topicPath, true);
 
         String topicName = topicPath.getName();
-        datumReader = new GenericDatumReader<>();
 
         try (FileCache cache = new FileCache(converterFactory, 100)) {
             while (files.hasNext()) {
@@ -187,7 +183,8 @@ private void processFile(Path filePath, String topicName, FileCache cache,
 
         // Read and parse avro file
         FsInput input = new FsInput(filePath, conf);
-        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(input, datumReader);
+        DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(input,
+                new GenericDatumReader<>());
 
         GenericRecord record = null;
         while (dataFileReader.hasNext()) {
@@ -200,7 +197,7 @@ record = dataFileReader.next(record);
         // Write which file has been processed and update bins
         try {
             offsets.write(range);
-            bins.writeBins();
+            bins.write();
         } catch (IOException ex) {
             logger.warn("Failed to update status. Continuing processing.", ex);
         }
@@ -230,7 +227,7 @@ private void writeRecord(GenericRecord record, String topicName, FileCache cache
         cache.writeRecord(outputFile, record);
 
         // Count data (binned and total)
-        bins.addToBin(topicName, keyField.get("sourceId").toString(), valueField, timeField);
+        bins.add(topicName, keyField.get("sourceId").toString(), valueField, timeField);
         processedRecordsCount++;
     }