Last-ditch optimization

Author: blootsvoets

build.gradle

@@ -22,9 +22,9 @@ ext {
 
     avroVersion = '1.8.2'
     jacksonVersion = '2.9.6'
-    hadoopVersion = '2.7.6'
+    hadoopVersion = '3.0.3'
     jCommanderVersion = '1.72'
-
+    almworksVersion = '1.1.1'
 }
 
 repositories {
@@ -36,11 +36,12 @@ dependencies {
     implementation group: 'com.fasterxml.jackson.core' , name: 'jackson-databind', version: jacksonVersion
     implementation group: 'com.fasterxml.jackson.dataformat' , name: 'jackson-dataformat-csv', version: jacksonVersion
     implementation group: 'com.beust', name: 'jcommander', version: jCommanderVersion
+    implementation group: 'com.almworks.integers', name: 'integers', version: almworksVersion
 
     implementation group: 'org.apache.avro', name: 'avro-mapred', version: avroVersion
     implementation group: 'org.apache.hadoop', name: 'hadoop-common', version: hadoopVersion
 
-    runtimeOnly group: 'org.apache.hadoop', name: 'hadoop-hdfs', version: hadoopVersion
+    runtimeOnly group: 'org.apache.hadoop', name: 'hadoop-hdfs-client', version: hadoopVersion
 
     testImplementation group: 'junit', name: 'junit', version: '4.12'
 }

src/main/java/org/radarcns/hdfs/RadarHdfsRestructure.java

@@ -29,8 +29,10 @@
 import org.radarcns.hdfs.accounting.Bin;
 import org.radarcns.hdfs.accounting.OffsetRange;
 import org.radarcns.hdfs.accounting.OffsetRangeSet;
+import org.radarcns.hdfs.accounting.TopicPartition;
 import org.radarcns.hdfs.data.FileCacheStore;
 import org.radarcns.hdfs.util.ProgressBar;
+import org.radarcns.hdfs.util.ReadOnlyFunctionalValue;
 import org.radarcns.hdfs.util.Timer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -111,12 +113,9 @@ public void start(String directoryName) throws IOException {
 
     private TopicFileList getTopicPaths(FileSystem fs, Path path, OffsetRangeSet seenFiles) {
         return new TopicFileList(walk(fs, path)
-                .filter(f -> {
-                    String name = f.getName();
-                    return name.endsWith(".avro")
-                            && !seenFiles.contains(OffsetRange.parseFilename(name));
-                })
+                .filter(f -> f.getName().endsWith(".avro"))
                 .map(f -> new TopicFile(f.getParent().getParent().getName(), f))
+                .filter(f -> !seenFiles.contains(f.range))
                 .collect(Collectors.toList()));
     }
 
@@ -147,6 +146,8 @@ private void processPaths(TopicFileList topicPaths, Accountant accountant) throw
 
         processedFileCount = new LongAdder();
         processedRecordsCount = new LongAdder();
+        OffsetRangeSet seenOffsets = accountant.getOffsets()
+                .withFactory(ReadOnlyFunctionalValue::new);
 
         ExecutorService executor = Executors.newWorkStealingPool(pathFactory.isTopicPartitioned() ? this.numThreads : 1);
 
@@ -168,7 +169,7 @@ private void processPaths(TopicFileList topicPaths, Accountant accountant) throw
                         try (FileCacheStore cache = fileStoreFactory.newFileCacheStore(accountant)) {
                             for (TopicFile file : paths.files) {
                                 try {
-                                    this.processFile(file, cache, progressBar);
+                                    this.processFile(file, cache, progressBar, seenOffsets);
                                 } catch (JsonMappingException exc) {
                                     logger.error("Cannot map values", exc);
                                 }
@@ -196,7 +197,7 @@ private void processPaths(TopicFileList topicPaths, Accountant accountant) throw
     }
 
     private void processFile(TopicFile file, FileCacheStore cache,
-            ProgressBar progressBar) throws IOException {
+            ProgressBar progressBar, OffsetRangeSet seenOffsets) throws IOException {
         logger.debug("Reading {}", file.path);
 
         // Read and parseFilename avro file
@@ -216,45 +217,49 @@ private void processFile(TopicFile file, FileCacheStore cache,
                 new GenericDatumReader<>());
 
         GenericRecord record = null;
-        int i = 0;
+        long offset = file.range.getOffsetFrom();
         while (dataFileReader.hasNext()) {
             record = dataFileReader.next(record);
             timer.add("read", timeRead);
 
             long timeAccount = System.nanoTime();
-            OffsetRange singleOffset = file.range.createSingleOffset(i++);
+            boolean alreadyContains = seenOffsets.contains(file.range.getTopicPartition(), offset);
             timer.add("accounting.create", timeAccount);
-            // Get the fields
-            this.writeRecord(record, file.topic, cache, singleOffset, 0);
-
+            if (!alreadyContains) {
+                // Get the fields
+                this.writeRecord(file.range.getTopicPartition(), record, cache, offset, 0);
+            }
             processedRecordsCount.increment();
             progressBar.update(processedRecordsCount.sum());
+
+            offset++;
             timeRead = System.nanoTime();
         }
     }
 
-    private void writeRecord(GenericRecord record, String topicName, FileCacheStore cache,
-            OffsetRange offset, int suffix) throws IOException {
-        RecordPathFactory.RecordOrganization metadata = pathFactory.getRecordOrganization(topicName, record, suffix);
+    private void writeRecord(TopicPartition topicPartition, GenericRecord record,
+            FileCacheStore cache, long offset, int suffix) throws IOException {
+        RecordPathFactory.RecordOrganization metadata = pathFactory.getRecordOrganization(
+                topicPartition.topic, record, suffix);
 
         Timer timer = Timer.getInstance();
 
         long timeAccount = System.nanoTime();
         String timeBin = pathFactory.getTimeBin(metadata.getTime());
-        Accountant.Transaction transaction = new Accountant.Transaction(offset,
-                new Bin(topicName, metadata.getCategory(), timeBin));
+        Accountant.Transaction transaction = new Accountant.Transaction(topicPartition, offset,
+                new Bin(topicPartition.topic, metadata.getCategory(), timeBin));
         timer.add("accounting.create", timeAccount);
 
         // Write data
         long timeWrite = System.nanoTime();
         FileCacheStore.WriteResponse response = cache.writeRecord(
-                topicName, metadata.getPath(), record, transaction);
+                metadata.getPath(), record, transaction);
         timer.add("write", timeWrite);
 
         if (!response.isSuccessful()) {
             // Write was unsuccessful due to different number of columns,
             // try again with new file name
-            writeRecord(record, topicName, cache, offset, ++suffix);
+            writeRecord(topicPartition, record, cache, offset, ++suffix);
         }
     }
 

src/main/java/org/radarcns/hdfs/accounting/Accountant.java

@@ -3,8 +3,8 @@
 import org.radarcns.hdfs.FileStoreFactory;
 import org.radarcns.hdfs.config.RestructureSettings;
 import org.radarcns.hdfs.data.StorageDriver;
-import org.radarcns.hdfs.util.TemporaryDirectory;
 import org.radarcns.hdfs.util.DirectFunctionalValue;
+import org.radarcns.hdfs.util.TemporaryDirectory;
 import org.radarcns.hdfs.util.Timer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -16,7 +16,6 @@
 import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.TreeSet;
 
 public class Accountant implements Flushable, Closeable {
     private static final Logger logger = LoggerFactory.getLogger(Accountant.class);
@@ -80,25 +79,14 @@ public OffsetRangeSet getOffsets() {
     @Override
     public void flush() throws IOException {
         long timeFlush = System.nanoTime();
-        IOException exception = null;
-        try {
-            binFile.flush();
-        } catch (IOException ex) {
-            logger.error("Failed to close bins", ex);
-            exception = ex;
-        }
+
+        binFile.triggerWrite();
 
         try {
             offsetFile.flush();
-        } catch (IOException ex) {
-            logger.error("Failed to close offsets", ex);
-            exception = ex;
+        } finally {
+            Timer.getInstance().add("accounting.flush", timeFlush);
         }
-
-        if (exception != null) {
-            throw exception;
-        }
-        Timer.getInstance().add("accounting.flush", timeFlush);
     }
 
     public BinFile getBins() {
@@ -110,25 +98,31 @@ public static class Ledger {
         private final Map<Bin, Long> bins;
 
         public Ledger() {
-            offsets = new OffsetRangeSet(() -> new DirectFunctionalValue<>(new TreeSet<>()));
+            offsets = new OffsetRangeSet(DirectFunctionalValue::new);
             bins = new HashMap<>();
         }
 
         public void add(Transaction transaction) {
             long timeAdd = System.nanoTime();
-            offsets.add(transaction.offset);
+            offsets.add(transaction.topicPartition, transaction.offset);
             bins.compute(transaction.bin, (b, vOld) -> vOld == null ? 1L : vOld + 1L);
             Timer.getInstance().add("accounting.add", timeAdd);
         }
     }
 
     public static class Transaction {
-        private final OffsetRange offset;
+        private final TopicPartition topicPartition;
+        private final long offset;
         private final Bin bin;
 
-        public Transaction(OffsetRange offset, Bin bin) {
+        public Transaction(TopicPartition topicPartition, long offset, Bin bin) {
+            this.topicPartition = topicPartition;
             this.offset = offset;
             this.bin = bin;
         }
+
+        public TopicPartition getTopicPartition() {
+            return topicPartition;
+        }
     }
 }

src/main/java/org/radarcns/hdfs/accounting/OffsetRange.java

@@ -16,17 +16,11 @@
 
 package org.radarcns.hdfs.accounting;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import javax.annotation.Nonnull;
-
 /** POJO class for storing offsets. */
-public class OffsetRange implements Comparable<OffsetRange> {
-    private final String topic;
-    private final int partition;
-    private long offsetFrom;
-    private long offsetTo;
+public class OffsetRange {
+    private final TopicPartition topicPartition;
+    private final long offsetFrom;
+    private final long offsetTo;
 
     public static OffsetRange parseFilename(String filename) throws NumberFormatException, IndexOutOfBoundsException {
         String[] fileNameParts = filename.split("[+.]");
@@ -39,61 +33,50 @@ public static OffsetRange parseFilename(String filename) throws NumberFormatExce
     }
 
     /** Full constructor. */
-    @JsonCreator
-    public OffsetRange(
-            @JsonProperty("topic") String topic,
-            @JsonProperty("partition") int partition,
-            @JsonProperty("offsetFrom") long offsetFrom,
-            @JsonProperty("offsetTo") long offsetTo) {
-        this.topic = topic;
-        this.partition = partition;
+    public OffsetRange(String topic, int partition, long offsetFrom, long offsetTo) {
+        this(new TopicPartition(topic, partition), offsetFrom, offsetTo);
+    }
+
+    public OffsetRange(TopicPartition topicPartition, long offsetFrom, long offsetTo) {
+        this.topicPartition = topicPartition;
         this.offsetFrom = offsetFrom;
         this.offsetTo = offsetTo;
     }
 
     public String getTopic() {
-        return topic;
+        return topicPartition.topic;
     }
 
     public int getPartition() {
-        return partition;
+        return topicPartition.partition;
     }
 
     public long getOffsetFrom() {
         return offsetFrom;
     }
 
-    public void setOffsetFrom(long offsetFrom) {
-        this.offsetFrom = offsetFrom;
-    }
-
     public long getOffsetTo() {
         return offsetTo;
     }
 
-    public void setOffsetTo(long offsetTo) {
-        this.offsetTo = offsetTo;
-    }
-
     public OffsetRange createSingleOffset(int index) {
         if (index < 0 || index > offsetTo - offsetFrom) {
             throw new IndexOutOfBoundsException("Index " + index + " does not reference offsets "
                     + offsetFrom + " to " + offsetTo);
         }
 
         long singleOffset = offsetFrom + index;
-        return new OffsetRange(topic, partition, singleOffset, singleOffset);
+        return new OffsetRange(topicPartition, singleOffset, singleOffset);
     }
 
     @Override
     public String toString() {
-        return topic + '+' + partition + '+' + offsetFrom + '+' + offsetTo;
+        return topicPartition.topic + '+' + topicPartition.partition + '+' + offsetFrom + '+' + offsetTo;
     }
 
     @Override
     public int hashCode() {
-        int result = topic.hashCode();
-        result = 31 * result + partition;
+        int result = topicPartition.hashCode();
         result = 31 * result + (int) (offsetFrom ^ (offsetFrom >>> 32));
         result = 31 * result + (int) (offsetTo ^ (offsetTo >>> 32));
         return result;
@@ -105,26 +88,13 @@ public boolean equals(Object o) {
             return false;
         }
         OffsetRange other = (OffsetRange) o;
-        return topic.equals(other.topic)
-                && partition == other.partition
+        return topicPartition.equals(other.topicPartition)
                 && offsetFrom == other.offsetFrom
                 && offsetTo == other.offsetTo;
     }
 
-    @Override
-    public int compareTo(@Nonnull OffsetRange o) {
-        int ret = Long.compare(offsetFrom, o.offsetFrom);
-        if (ret != 0) {
-            return ret;
-        }
-        ret = Long.compare(offsetTo, o.offsetTo);
-        if (ret != 0) {
-            return ret;
-        }
-        ret = topic.compareTo(o.topic);
-        if (ret != 0) {
-            return ret;
-        }
-        return Integer.compare(partition, o.partition);
+    public TopicPartition getTopicPartition() {
+        return topicPartition;
     }
+
 }