Added ZIP compression

Author: blootsvoets

src/main/java/org/radarcns/hdfs/data/Compression.java

@@ -21,7 +21,7 @@
 import java.io.OutputStream;
 
 public interface Compression extends Format {
-    OutputStream compress(OutputStream out) throws IOException;
+    OutputStream compress(String fileName, OutputStream out) throws IOException;
     InputStream decompress(InputStream in) throws IOException;
 
     String getExtension();

src/main/java/org/radarcns/hdfs/data/CompressionFactory.java

@@ -23,6 +23,7 @@ public class CompressionFactory implements FormatProvider<Compression> {
     public final List<Compression> getAll() {
         return Arrays.asList(
                 new GzipCompression(),
-                new IdentityCompression());
+                new IdentityCompression(),
+                new ZipCompression());
     }
 }

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

@@ -52,6 +52,7 @@ public class FileCache implements Closeable, Flushable, Comparable<FileCache> {
     private final boolean deduplicate;
     private final Accountant.Ledger ledger;
     private final Accountant accountant;
+    private final String fileName;
     private long lastUse;
     private final AtomicBoolean hasError;
 
@@ -70,12 +71,12 @@ public FileCache(FileStoreFactory factory, Path path, GenericRecord record,
         this.ledger = new Accountant.Ledger();
         this.compression = factory.getCompression();
         boolean fileIsNew = !storageDriver.exists(path) || storageDriver.size(path) == 0;
-        this.tmpPath = Files.createTempFile(tmpDir, path.getFileName().toString(),
-                ".tmp" + compression.getExtension());
+        this.fileName = path.getFileName().toString();
+        this.tmpPath = Files.createTempFile(tmpDir, fileName,".tmp" + compression.getExtension());
         this.converterFactory = factory.getRecordConverter();
         this.accountant = accountant;
 
-        OutputStream outStream = compression.compress(
+        OutputStream outStream = compression.compress(fileName,
                 new BufferedOutputStream(Files.newOutputStream(tmpPath)));
 
         InputStream inputStream;
@@ -90,7 +91,7 @@ public FileCache(FileStoreFactory factory, Path path, GenericRecord record,
                 outStream.close();
                 // clear output file
                 outStream = compression.compress(
-                        new BufferedOutputStream(Files.newOutputStream(tmpPath)));
+                        fileName, new BufferedOutputStream(Files.newOutputStream(tmpPath)));
             }
             Timer.getInstance().add("write.copyOriginal", timeCopy);
         }
@@ -141,7 +142,7 @@ public void close() throws IOException {
         if (!hasError.get()) {
             if (deduplicate) {
                 long timeDedup = System.nanoTime();
-                converterFactory.sortUnique(tmpPath, tmpPath, compression);
+                converterFactory.sortUnique(fileName, tmpPath, tmpPath, compression);
                 Timer.getInstance().add("close.deduplicate", timeDedup);
             }
 

src/main/java/org/radarcns/hdfs/data/GzipCompression.java

@@ -26,7 +26,7 @@
 
 public class GzipCompression implements Compression {
     @Override
-    public OutputStream compress(OutputStream out) throws IOException {
+    public OutputStream compress(String fileName, OutputStream out) throws IOException {
         return new GZIPOutputStream(out);
     }
 

src/main/java/org/radarcns/hdfs/data/IdentityCompression.java

@@ -23,7 +23,7 @@
 
 public class IdentityCompression implements Compression {
     @Override
-    public OutputStream compress(OutputStream out) {
+    public OutputStream compress(String fileName, OutputStream out) {
         return out;
     }
 

src/main/java/org/radarcns/hdfs/data/RecordConverterFactory.java

@@ -48,7 +48,8 @@ default boolean hasHeader() {
         return false;
     }
 
-    default void sortUnique(Path source, Path target, Compression compression)
+    default void sortUnique(String fileName, Path source, Path target,
+            Compression compression)
             throws IOException {
         // read all lines into memory; assume a 100-byte line length
         LinkedHashSet<String> sortedLines = new LinkedHashSet<>((int)(Files.size(source) / 100));
@@ -63,7 +64,7 @@ default void sortUnique(Path source, Path target, Compression compression)
         }
         try (OutputStream fileOut = Files.newOutputStream(target);
              OutputStream bufOut = new BufferedOutputStream(fileOut);
-             OutputStream zipOut = compression.compress(bufOut);
+             OutputStream zipOut = compression.compress(fileName, bufOut);
              Writer writer = new OutputStreamWriter(zipOut)) {
             writeFile(writer, header, sortedLines);
         }

src/main/java/org/radarcns/hdfs/data/ZipCompression.java

@@ -0,0 +1,37 @@
+package org.radarcns.hdfs.data;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+import java.util.zip.ZipOutputStream;
+
+public class ZipCompression implements Compression {
+    @Override
+    public OutputStream compress(String name, OutputStream out) throws IOException {
+        ZipOutputStream zipOut = new ZipOutputStream(out);
+        zipOut.setLevel(7);
+        zipOut.putNextEntry(new ZipEntry(name));
+        return zipOut;
+    }
+
+    @Override
+    public InputStream decompress(InputStream in) throws IOException {
+        ZipInputStream zipIn = new ZipInputStream(in);
+        zipIn.getNextEntry();
+        return zipIn;
+    }
+
+    @Override
+    public Collection<String> getFormats() {
+        return Collections.singleton("zip");
+    }
+
+    @Override
+    public String getExtension() {
+        return ".zip";
+    }
+}