Make the staging behaviour optional

Author: blootsvoets

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

@@ -63,6 +63,7 @@ public class RestructureAvroRecords {
     }
 
     private final RecordConverterFactory converterFactory;
+    private final boolean doStage;
 
     private java.nio.file.Path outputPath;
     private java.nio.file.Path offsetsPath;
@@ -97,6 +98,7 @@ public static void main(String [] args) {
                 commandLineArgs.outputDirectory)
                 .useGzip("gzip".equalsIgnoreCase(commandLineArgs.compression))
                 .doDeduplicate(commandLineArgs.deduplicate).format(commandLineArgs.format)
+                .doStage(!commandLineArgs.noStage)
                 .build();
 
         try {
@@ -119,6 +121,7 @@ private RestructureAvroRecords(RestructureAvroRecords.Builder builder) {
 
         this.useGzip = builder.useGzip;
         this.doDeduplicate = builder.doDeduplicate;
+        this.doStage = builder.doStage;
         logger.info("Deduplicate set to {}", doDeduplicate);
 
         String extension;
@@ -197,7 +200,7 @@ public void start(String directoryName) throws IOException {
 
             // Actually process the files
             for (Map.Entry<String, List<Path>> entry : topicPaths.entrySet()) {
-                try (FileCacheStore cache = new FileCacheStore(converterFactory, 100, useGzip, doDeduplicate)) {
+                try (FileCacheStore cache = new FileCacheStore(converterFactory, 100, useGzip, doDeduplicate, doStage)) {
                     for (Path filePath : entry.getValue()) {
                         // If JsonMappingException occurs, log the error and continue with other files
                         try {
@@ -372,6 +375,7 @@ public static class Builder {
         private String hdfsUri;
         private String outputPath;
         private String format;
+        private boolean doStage;
 
         public Builder(final String uri, final String outputPath) {
             this.hdfsUri = uri;
@@ -397,5 +401,9 @@ public RestructureAvroRecords build() {
             return new RestructureAvroRecords(this);
         }
 
+        public Builder doStage(boolean stage) {
+            this.doStage = stage;
+            return this;
+        }
     }
 }

src/main/java/org/radarcns/data/FileCache.java

@@ -46,16 +46,22 @@ public class FileCache implements Closeable, Flushable, Comparable<FileCache> {
      * @param path path to cache.
      * @param record example record to create converter from, this is not written to path.
      * @param gzip whether to gzip the records
-     * @throws IOException
+     * @throws IOException if the file and/or temporary files cannot be correctly read or written to.
      */
     public FileCache(RecordConverterFactory converterFactory, Path path,
             GenericRecord record, boolean gzip, Path tmpDir) throws IOException {
         this.path = path;
         boolean fileIsNew = !Files.exists(path) || Files.size(path) == 0;
-        this.tmpPath = Files.createTempFile(tmpDir, path.getFileName().toString(),
-                gzip ? ".tmp.gz" : ".tmp");
+        OutputStream outFile;
+        if (tmpDir == null) {
+            this.tmpPath = null;
+            outFile = Files.newOutputStream(path, StandardOpenOption.APPEND, StandardOpenOption.CREATE);
+        } else {
+            this.tmpPath = Files.createTempFile(tmpDir, path.getFileName().toString(),
+                    gzip ? ".tmp.gz" : ".tmp");
+            outFile = Files.newOutputStream(tmpPath, StandardOpenOption.WRITE);
+        }
 
-        OutputStream outFile = Files.newOutputStream(tmpPath, StandardOpenOption.WRITE);
         OutputStream bufOut = new BufferedOutputStream(outFile);
         if (gzip) {
             bufOut = new GZIPOutputStream(bufOut);
@@ -65,13 +71,11 @@ public FileCache(RecordConverterFactory converterFactory, Path path,
         if (fileIsNew) {
             inputStream = new ByteArrayInputStream(new byte[0]);
         } else {
-            inputStream = new BufferedInputStream(Files.newInputStream(path));
-            InputStream copyStream = Files.newInputStream(path);
-            if (gzip) {
-                copyStream = new GZIPInputStream(copyStream);
-                inputStream = new GZIPInputStream(inputStream);
+            inputStream = inputStream(new BufferedInputStream(Files.newInputStream(path)), gzip);
+
+            if (tmpPath != null) {
+                copy(path, bufOut, gzip);
             }
-            copy(copyStream, bufOut);
         }
 
         this.writer = new OutputStreamWriter(bufOut);
@@ -104,7 +108,9 @@ public boolean writeRecord(GenericRecord record) throws IOException {
     public void close() throws IOException {
         recordConverter.close();
         writer.close();
-        Files.move(tmpPath, path, REPLACE_EXISTING);
+        if (tmpPath != null) {
+            Files.move(tmpPath, path, REPLACE_EXISTING);
+        }
     }
 
     @Override
@@ -131,6 +137,19 @@ public Path getPath() {
         return path;
     }
 
+    private static void copy(Path source, OutputStream sink, boolean gzip) throws IOException {
+        try (InputStream copyStream = inputStream(Files.newInputStream(source), gzip)) {
+            copy(copyStream, sink);
+        }
+    }
+
+    private static InputStream inputStream(InputStream in, boolean gzip) throws IOException {
+        if (gzip) {
+            return new GZIPInputStream(in);
+        } else {
+            return in;
+        }
+    }
 
     /**
      * Reads all bytes from an input stream and writes them to an output stream.

src/main/java/org/radarcns/data/FileCacheStore.java

@@ -48,13 +48,13 @@ public class FileCacheStore implements Flushable, Closeable {
     private final int maxFiles;
     private final Map<Path, FileCache> caches;
 
-    public FileCacheStore(RecordConverterFactory converterFactory, int maxFiles, boolean gzip, boolean deduplicate) throws IOException {
+    public FileCacheStore(RecordConverterFactory converterFactory, int maxFiles, boolean gzip, boolean deduplicate, boolean stage) throws IOException {
         this.converterFactory = converterFactory;
         this.maxFiles = maxFiles;
         this.caches = new HashMap<>(maxFiles * 4 / 3 + 1);
         this.gzip = gzip;
         this.deduplicate = deduplicate;
-        this.tmpDir = Files.createTempDirectory("restructurehdfs");
+        this.tmpDir = stage ? Files.createTempDirectory("restructurehdfs") : null;
     }
 
     /**
@@ -129,10 +129,12 @@ public void close() throws IOException {
                     converterFactory.sortUnique(cache.getPath());
                 }
             }
-            Files.walk(tmpDir)
-                    .sorted(Comparator.reverseOrder())
-                    .forEach(tryCatch(Files::delete, (p, ex) -> logger.warn(
-                            "Failed to remove temporary file {}: {}", p, ex)));
+            if (tmpDir != null) {
+                Files.walk(tmpDir)
+                        .sorted(Comparator.reverseOrder())
+                        .forEach(tryCatch(Files::delete, (p, ex) -> logger.warn(
+                                "Failed to remove temporary file {}: {}", p, ex)));
+            }
         } finally {
             caches.clear();
         }

src/main/java/org/radarcns/util/commandline/CommandLineArgs.java

@@ -28,4 +28,7 @@ public class CommandLineArgs {
 
     @Parameter(names = { "-h", "--help"}, help = true, description = "Display the usage of the program with available options.")
     public boolean help;
+
+    @Parameter(names = { "--no-stage"}, description = "Do not stage output files into a temporary directory before moving them to the data directory. This increases performance but may leave corrupted data files.")
+    public boolean noStage = false;
 }

src/test/java/org/radarcns/util/FileCacheStoreTest.java

@@ -21,6 +21,8 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collection;
 
 import org.apache.avro.Schema;
 import org.apache.avro.SchemaBuilder;
@@ -29,11 +31,23 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.radarcns.data.CsvAvroConverter;
 import org.radarcns.data.FileCacheStore;
 import org.radarcns.data.RecordConverterFactory;
 
+@RunWith(Parameterized.class)
 public class FileCacheStoreTest {
+
+    @Parameterized.Parameters
+    public static Collection<Boolean> doStage() {
+        return Arrays.asList(Boolean.TRUE, Boolean.FALSE);
+    }
+
+    @Parameterized.Parameter
+    public Boolean doStage;
+
     @Rule
     public TemporaryFolder folder = new TemporaryFolder();
 
@@ -61,7 +75,7 @@ public void appendLine() throws IOException {
 
         GenericRecord record;
 
-        try (FileCacheStore cache = new FileCacheStore(csvFactory, 2, false, false)) {
+        try (FileCacheStore cache = new FileCacheStore(csvFactory, 2, false, false, doStage)) {
             record = new GenericRecordBuilder(simpleSchema).set("a", "something").build();
             assertEquals(cache.writeRecord(f1, record), FileCacheStore.WriteStatus.NO_CACHE_AND_WRITE);
             record = new GenericRecordBuilder(simpleSchema).set("a", "somethingElse").build();

src/test/java/org/radarcns/util/FileCacheTest.java

@@ -24,6 +24,8 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.radarcns.data.CsvAvroConverter;
 import org.radarcns.data.FileCache;
 import org.radarcns.data.RecordConverterFactory;
@@ -35,6 +37,10 @@
 import java.io.Reader;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
 import java.util.zip.GZIPInputStream;
 
 import static org.junit.Assert.assertEquals;
@@ -43,18 +49,27 @@
 /**
  * Created by joris on 03/07/2017.
  */
+@RunWith(Parameterized.class)
 public class FileCacheTest {
+    @Parameterized.Parameters
+    public static Collection<Boolean> useTmpDir() {
+        return Arrays.asList(Boolean.TRUE, Boolean.FALSE);
+    }
+
     @Rule
     public TemporaryFolder folder = new TemporaryFolder();
     private Path path;
     private RecordConverterFactory csvFactory;
     private Record exampleRecord;
     private Path tmpDir;
 
+    @Parameterized.Parameter
+    public Boolean useTmpDir;
+
     @Before
     public void setUp() throws IOException {
         this.path = folder.newFile("f").toPath();
-        this.tmpDir = folder.newFolder().toPath();
+        this.tmpDir = useTmpDir ? folder.newFolder().toPath() : null;
 
         this.csvFactory = CsvAvroConverter.getFactory();
         Schema schema = SchemaBuilder.record("simple").fields()