Enabled GZIP compression for output files

Author: blootsvoets

README.md

@@ -25,3 +25,8 @@ By default, this will output the data in CSV format. If JSON format is preferred
 ```
 java -Dorg.radarcns.format=json -jar restructurehdfs-all-0.1.1.jar <webhdfs_url> <hdfs_topic_path> <output_folder>
 ```
+
+Another option is to output the data in compressed form. All files will get the `gz` suffix, and can be decompressed with a GZIP decoder. Note that for a very small number of records, this may actually increase the file size.
+```
+java -Dorg.radarcns.compress=gzip -jar restructurehdfs-all-0.1.1.jar <webhdfs_url> <hdfs_topic_path> <output_folder>
+```

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

@@ -33,7 +33,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
 import org.radarcns.util.CsvAvroConverter;
-import org.radarcns.util.FileCache;
+import org.radarcns.util.FileCacheStore;
 import org.radarcns.util.JsonAvroConverter;
 import org.radarcns.util.RecordConverterFactory;
 import org.slf4j.Logger;
@@ -62,6 +62,7 @@ public class RestructureAvroRecords {
 
     private int processedFileCount;
     private int processedRecordsCount;
+    private static final boolean USE_GZIP = "gzip".equalsIgnoreCase(System.getProperty("org.radarcns.compression"));
 
     public static void main(String [] args) throws Exception {
         if (args.length != 3) {
@@ -91,13 +92,21 @@ public RestructureAvroRecords(String inputPath, String outputPath) {
         this.setInputWebHdfsURL(inputPath);
         this.setOutputPath(outputPath);
 
+        String extension;
         if (System.getProperty("org.radarcns.format", "csv").equalsIgnoreCase("json")) {
+            logger.info("Writing output files in JSON format");
             converterFactory = JsonAvroConverter.getFactory();
-            outputFileExtension = "json";
+            extension = "json";
         } else {
+            logger.info("Writing output files in CSV format");
             converterFactory = CsvAvroConverter.getFactory();
-            outputFileExtension = "csv";
+            extension = "csv";
         }
+        if (USE_GZIP) {
+            logger.info("Compressing output files in GZIP format");
+            extension += ".gz";
+        }
+        outputFileExtension = extension;
     }
 
     public void setInputWebHdfsURL(String fileSystemURL) {
@@ -157,7 +166,7 @@ private void processTopic(Path topicPath, RecordConverterFactory converterFactor
 
         String topicName = topicPath.getName();
 
-        try (FileCache cache = new FileCache(converterFactory, 100)) {
+        try (FileCacheStore cache = new FileCacheStore(converterFactory, 100, USE_GZIP)) {
             while (files.hasNext()) {
                 LocatedFileStatus locatedFileStatus = files.next();
 
@@ -168,7 +177,7 @@ private void processTopic(Path topicPath, RecordConverterFactory converterFactor
         }
     }
 
-    private void processFile(Path filePath, String topicName, FileCache cache,
+    private void processFile(Path filePath, String topicName, FileCacheStore cache,
             OffsetRangeFile offsets) throws IOException {
         String fileName = filePath.getName();
 
@@ -209,7 +218,7 @@ record = dataFileReader.next(record);
         processedFileCount++;
     }
 
-    private void writeRecord(GenericRecord record, String topicName, FileCache cache)
+    private void writeRecord(GenericRecord record, String topicName, FileCacheStore cache)
             throws IOException {
         GenericRecord keyField = (GenericRecord) record.get("key");
         GenericRecord valueField = (GenericRecord) record.get("value");

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

@@ -16,144 +16,87 @@
 
 package org.radarcns.util;
 
-import java.io.BufferedWriter;
+import java.io.BufferedOutputStream;
 import java.io.Closeable;
 import java.io.File;
-import java.io.FileWriter;
+import java.io.FileOutputStream;
 import java.io.Flushable;
 import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
 import java.io.Writer;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
+import java.util.zip.GZIPOutputStream;
 import javax.annotation.Nonnull;
 import org.apache.avro.generic.GenericRecord;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-/**
- * Caches open file handles. If more than the limit is cached, the half of the files that were used
- * the longest ago cache are evicted from cache.
- */
-public class FileCache implements Flushable, Closeable {
-    private static final Logger logger = LoggerFactory.getLogger(FileCache.class);
-
-    private RecordConverterFactory converterFactory;
-    private final int maxFiles;
-    private final Map<File, SingleFileCache> caches;
-
-    public FileCache(RecordConverterFactory converterFactory, int maxFiles) {
-        this.converterFactory = converterFactory;
-        this.maxFiles = maxFiles;
-        this.caches = new HashMap<>(maxFiles * 4 / 3 + 1);
-    }
+/** Keeps file handles of a file. */
+public class FileCache implements Closeable, Flushable, Comparable<FileCache> {
+    private final OutputStream[] streams;
+    private final Writer writer;
+    private final RecordConverter recordConverter;
+    private final File file;
+    private long lastUse;
 
     /**
-     * Append a record to given file. If the file handle and writer are already open in this cache,
-     * those will be used. Otherwise, the file will be opened and the file handle cached.
-     *
-     * @param file file to append data to
-     * @param record data
-     * @return true if the cache was used, false if a new file was opened.
-     * @throws IOException when failing to open a file or writing to it.
+     * File cache of given file, using given converter factory.
+     * @param converterFactory converter factory to create a converter to write files with.
+     * @param file file to cache.
+     * @param record example record to create converter from, this is not written to file.
+     * @param gzip whether to gzip the records
+     * @throws IOException
      */
-    public boolean writeRecord(File file, GenericRecord record) throws IOException {
-        SingleFileCache cache = caches.get(file);
-        if (cache != null) {
-            cache.writeRecord(record);
-            return true;
-        } else {
-            ensureCapacity();
+    public FileCache(RecordConverterFactory converterFactory, File file,
+            GenericRecord record, boolean gzip) throws IOException {
+        this.file = file;
+        boolean fileIsNew = !file.exists() || file.length() == 0;
+
+        this.streams = new OutputStream[gzip ? 3 : 2];
+        this.streams[0] = new FileOutputStream(file, true);
+        this.streams[1] = new BufferedOutputStream(this.streams[0]);
+        if (gzip) {
+            this.streams[2] = new GZIPOutputStream(this.streams[1]);
+        }
 
-            File dir = file.getParentFile();
-            if (!dir.exists()){
-                if (dir.mkdirs()) {
-                    logger.debug("Created directory: {}", dir.getAbsolutePath());
-                } else {
-                    logger.warn("FAILED to create directory: {}", dir.getAbsolutePath());
-                }
-            }
+        this.writer = new OutputStreamWriter(this.streams[this.streams.length - 1]);
+        this.recordConverter = converterFactory.converterFor(writer, record, fileIsNew);
+    }
 
-            cache = new SingleFileCache(file, record);
-            caches.put(file, cache);
-            cache.writeRecord(record);
-            return false;
-        }
+    /** Write a record to the cache. */
+    public void writeRecord(GenericRecord record) throws IOException {
+        this.recordConverter.writeRecord(record);
+        lastUse = System.nanoTime();
     }
 
-    /**
-     * Ensure that a new filecache can be added. Evict files used longest ago from cache if needed.
-     */
-    private void ensureCapacity() throws IOException {
-        if (caches.size() == maxFiles) {
-            ArrayList<SingleFileCache> cacheList = new ArrayList<>(caches.values());
-            Collections.sort(cacheList);
-            for (int i = 0; i < cacheList.size() / 2; i++) {
-                SingleFileCache rmCache = cacheList.get(i);
-                caches.remove(rmCache.getFile());
-                rmCache.close();
-            }
+    @Override
+    public void close() throws IOException {
+        recordConverter.close();
+        writer.close();
+        for (int i = streams.length - 1; i >= 0; i--) {
+            streams[i].close();
         }
     }
 
     @Override
     public void flush() throws IOException {
-        for (SingleFileCache cache : caches.values()) {
-            cache.flush();
-        }
+        recordConverter.flush();
     }
 
+    /**
+     * Compares time that the filecaches were last used. If equal, it lexicographically compares
+     * the absolute path of the file.
+     * @param other FileCache to compare with.
+     */
     @Override
-    public void close() throws IOException {
-        try {
-            for (SingleFileCache cache : caches.values()) {
-                cache.close();
-            }
-        } finally {
-            caches.clear();
+    public int compareTo(@Nonnull FileCache other) {
+        int result = Long.compare(lastUse, other.lastUse);
+        if (result != 0) {
+            return result;
         }
+        return file.compareTo(other.file);
     }
 
-    private class SingleFileCache implements Closeable, Flushable, Comparable<SingleFileCache> {
-        private final BufferedWriter bufferedWriter;
-        private final Writer fileWriter;
-        private final RecordConverter recordConverter;
-        private final File file;
-        private long lastUse;
-
-        private SingleFileCache(File file, GenericRecord record) throws IOException {
-            this.file = file;
-            boolean fileIsNew = !file.exists() || file.length() == 0;
-            this.fileWriter = new FileWriter(file, true);
-            this.bufferedWriter = new BufferedWriter(fileWriter);
-            this.recordConverter = converterFactory.converterFor(bufferedWriter, record, fileIsNew);
-        }
-
-        private void writeRecord(GenericRecord record) throws IOException {
-            this.recordConverter.writeRecord(record);
-            lastUse = System.nanoTime();
-        }
-
-        @Override
-        public void close() throws IOException {
-            recordConverter.close();
-            bufferedWriter.close();
-            fileWriter.close();
-        }
-
-        @Override
-        public void flush() throws IOException {
-            recordConverter.flush();
-        }
-
-        @Override
-        public int compareTo(@Nonnull SingleFileCache other) {
-            return Long.compare(lastUse, other.lastUse);
-        }
-
-        private File getFile() {
-            return file;
-        }
+    /** File that the cache is maintaining. */
+    public File getFile() {
+        return file;
     }
 }

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

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2017 The Hyve
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.radarcns.util;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.Flushable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.avro.generic.GenericRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Caches open file handles. If more than the limit is cached, the half of the files that were used
+ * the longest ago cache are evicted from cache.
+ */
+public class FileCacheStore implements Flushable, Closeable {
+    private static final Logger logger = LoggerFactory.getLogger(FileCacheStore.class);
+    private final boolean gzip;
+
+    private RecordConverterFactory converterFactory;
+    private final int maxFiles;
+    private final Map<File, FileCache> caches;
+
+    public FileCacheStore(RecordConverterFactory converterFactory, int maxFiles, boolean gzip) {
+        this.converterFactory = converterFactory;
+        this.maxFiles = maxFiles;
+        this.caches = new HashMap<>(maxFiles * 4 / 3 + 1);
+        this.gzip = gzip;
+    }
+
+    /**
+     * Append a record to given file. If the file handle and writer are already open in this cache,
+     * those will be used. Otherwise, the file will be opened and the file handle cached.
+     *
+     * @param file file to append data to
+     * @param record data
+     * @return true if the cache was used, false if a new file was opened.
+     * @throws IOException when failing to open a file or writing to it.
+     */
+    public boolean writeRecord(File file, GenericRecord record) throws IOException {
+        FileCache cache = caches.get(file);
+        if (cache != null) {
+            cache.writeRecord(record);
+            return true;
+        } else {
+            ensureCapacity();
+
+            File dir = file.getParentFile();
+            if (!dir.exists()){
+                if (dir.mkdirs()) {
+                    logger.debug("Created directory: {}", dir.getAbsolutePath());
+                } else {
+                    logger.warn("FAILED to create directory: {}", dir.getAbsolutePath());
+                }
+            }
+
+            cache = new FileCache(converterFactory, file, record, gzip);
+            caches.put(file, cache);
+            cache.writeRecord(record);
+            return false;
+        }
+    }
+
+    /**
+     * Ensure that a new filecache can be added. Evict files used longest ago from cache if needed.
+     */
+    private void ensureCapacity() throws IOException {
+        if (caches.size() == maxFiles) {
+            ArrayList<FileCache> cacheList = new ArrayList<>(caches.values());
+            Collections.sort(cacheList);
+            for (int i = 0; i < cacheList.size() / 2; i++) {
+                FileCache rmCache = cacheList.get(i);
+                caches.remove(rmCache.getFile());
+                rmCache.close();
+            }
+        }
+    }
+
+    @Override
+    public void flush() throws IOException {
+        for (FileCache cache : caches.values()) {
+            cache.flush();
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        try {
+            for (FileCache cache : caches.values()) {
+                cache.close();
+            }
+        } finally {
+            caches.clear();
+        }
+    }
+
+}