Merge branch 'master' into dev

Author: blootsvoets

.travis.yml

@@ -1,16 +1,16 @@
 language: java
 jdk:
   - oraclejdk8
-env:
-  TERM: dumb
-before_cache:
-  - rm -f  $HOME/.gradle/caches/modules-2/modules-2.lock
-  - rm -fr $HOME/.gradle/caches/*/plugin-resolution
-  - rm -fr $HOME/.gradle/caches/*/fileHashes
+sudo: false
+
 cache:
   directories:
-    - $HOME/.gradle/caches/
-    - $HOME/.gradle/wrapper/
+    - $HOME/.gradle/caches/jars-1
+    - $HOME/.gradle/caches/jars-2
+    - $HOME/.gradle/caches/jars-3
+    - $HOME/.gradle/caches/modules-2/files-2.1/
+    - $HOME/.gradle/native
+    - $HOME/.gradle/wrapper
 
 deploy:
   provider: releases

README.md

@@ -15,18 +15,18 @@ Build jar from source with
 ```shell
 ./gradlew build
 ```
-and find the output JAR file as `build/libs/restructurehdfs-all-0.1.2-SNAPSHOT.jar`. Then run with:
+and find the output JAR file as `build/libs/restructurehdfs-all-0.2.jar`. Then run with:
 
 ```shell
-java -jar restructurehdfs-all-0.1.2-SNAPSHOT.jar <webhdfs_url> <hdfs_topic_path> <output_folder>
+java -jar restructurehdfs-all-0.2.jar <webhdfs_url> <hdfs_topic_path> <output_folder>
 ```
 
 By default, this will output the data in CSV format. If JSON format is preferred, use the following instead:
 ```
-java -Dorg.radarcns.format=json -jar restructurehdfs-all-0.1.2-SNAPSHOT.jar <webhdfs_url> <hdfs_topic_path> <output_folder>
+java -Dorg.radarcns.format=json -jar restructurehdfs-all-0.2.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>
+java -Dorg.radarcns.compress=gzip -jar restructurehdfs-all-0.2.jar <webhdfs_url> <hdfs_topic_path> <output_folder>
 ```

build.gradle

@@ -2,7 +2,7 @@ apply plugin: 'java'
 apply plugin: 'application'
 
 group 'org.radarcns.restructurehdfs'
-version '0.1.2-SNAPSHOT'
+version '0.2'
 mainClassName = 'org.radarcns.RestructureAvroRecords'
 
 run {
@@ -12,10 +12,9 @@ run {
 sourceCompatibility = '1.8'
 targetCompatibility = '1.8'
 
-ext.avroVersion = '1.8.1'
-ext.jacksonVersion = '2.8.5'
+ext.avroVersion = '1.8.2'
+ext.jacksonVersion = '2.8.9'
 ext.hadoopVersion = '2.7.3'
-ext.log4jVersion = '2.8.1'
 
 repositories {
     jcenter()
@@ -57,6 +56,6 @@ artifacts {
 }
 
 task wrapper(type: Wrapper) {
-    gradleVersion = '3.4.1'
-    distributionUrl distributionUrl.replace("bin", "all")
+    gradleVersion = '3.5'
+    distributionType 'all'
 }

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,6 @@
-#Thu Apr 13 12:11:40 CEST 2017
+#Mon Jul 03 09:50:31 CEST 2017
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-3.4.1-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-3.5-all.zip

gradle/wrapper/gradle-wrapper.properties

@@ -19,14 +19,17 @@
 import java.io.File;
 import java.io.IOException;
 import java.text.SimpleDateFormat;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.TimeZone;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.file.DataFileReader;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.mapred.FsInput;
-import org.apache.commons.io.FilenameUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocatedFileStatus;
@@ -35,6 +38,7 @@
 import org.radarcns.util.CsvAvroConverter;
 import org.radarcns.util.FileCacheStore;
 import org.radarcns.util.JsonAvroConverter;
+import org.radarcns.util.ProgressBar;
 import org.radarcns.util.RecordConverterFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,18 +59,17 @@ public class RestructureAvroRecords {
 
     private File outputPath;
     private File offsetsPath;
-    private OffsetRangeSet seenFiles;
     private Frequency bins;
 
     private final Configuration conf = new Configuration();
 
-    private int processedFileCount;
-    private int processedRecordsCount;
+    private long processedFileCount;
+    private long 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) {
-            System.out.println("Usage: hadoop jar restructurehdfs-all-0.1.0.jar <webhdfs_url> <hdfs_topic> <output_folder>");
+            System.out.println("Usage: hadoop jar restructurehdfs-all-0.2.jar <webhdfs_url> <hdfs_root_directory> <output_folder>");
             System.exit(1);
         }
 
@@ -120,80 +123,89 @@ public void setOutputPath(String path) {
         bins = Frequency.read(new File(outputPath, BINS_FILE_NAME));
     }
 
-    public int getProcessedFileCount() {
+    public long getProcessedFileCount() {
         return processedFileCount;
     }
 
-    public int getProcessedRecordsCount() {
+    public long getProcessedRecordsCount() {
         return processedRecordsCount;
     }
 
     public void start(String directoryName) throws IOException {
         // Get files and directories
         Path path = new Path(directoryName);
         FileSystem fs = FileSystem.get(conf);
-        RemoteIterator<LocatedFileStatus> files = fs.listLocatedStatus(path);
+
 
         try (OffsetRangeFile offsets = new OffsetRangeFile(offsetsPath)) {
+            OffsetRangeSet seenFiles;
             try {
                 seenFiles = offsets.read();
             } catch (IOException ex) {
                 logger.error("Error reading offsets file. Processing all offsets.");
                 seenFiles = new OffsetRangeSet();
             }
-            // Process the directories topics
-            processedFileCount = 0;
+            logger.info("Retrieving file list from {}", path);
+            // Get filenames to process
+            Map<String, List<Path>> topicPaths = new HashMap<>();
+            long toProcessFileCount = 0L;
+            processedFileCount = 0L;
+            RemoteIterator<LocatedFileStatus> files = fs.listFiles(path, true);
             while (files.hasNext()) {
                 LocatedFileStatus locatedFileStatus = files.next();
-                Path filePath = locatedFileStatus.getPath();
-
-                if (filePath.toString().contains("+tmp")) {
+                if (locatedFileStatus.isDirectory()) {
                     continue;
                 }
+                Path filePath = locatedFileStatus.getPath();
 
-                if (locatedFileStatus.isDirectory()) {
-                    processTopic(filePath, converterFactory, offsets);
+                String topic = getTopic(filePath, seenFiles);
+                if (topic != null) {
+                    topicPaths.computeIfAbsent(topic, k -> new ArrayList<>()).add(filePath);
+                    toProcessFileCount++;
                 }
             }
-        }
-    }
-
-    private void processTopic(Path topicPath, RecordConverterFactory converterFactory,
-            OffsetRangeFile offsets) throws IOException {
-        // Get files in this topic directory
-        FileSystem fs = FileSystem.get(conf);
-        RemoteIterator<LocatedFileStatus> files = fs.listFiles(topicPath, true);
 
-        String topicName = topicPath.getName();
+            logger.info("Converting {} files", toProcessFileCount);
 
-        try (FileCacheStore cache = new FileCacheStore(converterFactory, 100, USE_GZIP)) {
-            while (files.hasNext()) {
-                LocatedFileStatus locatedFileStatus = files.next();
+            ProgressBar progressBar = new ProgressBar(toProcessFileCount, 10);
+            progressBar.update(0);
 
-                if (locatedFileStatus.isFile()) {
-                    this.processFile(locatedFileStatus.getPath(), topicName, cache, offsets);
+            // Actually process the files
+            for (Map.Entry<String, List<Path>> entry : topicPaths.entrySet()) {
+                try (FileCacheStore cache = new FileCacheStore(converterFactory, 100, USE_GZIP)) {
+                    for (Path filePath : entry.getValue()) {
+                        this.processFile(filePath, entry.getKey(), cache, offsets);
+                        progressBar.update(++processedFileCount);
+                    }
                 }
             }
         }
     }
 
-    private void processFile(Path filePath, String topicName, FileCacheStore cache,
-            OffsetRangeFile offsets) throws IOException {
-        String fileName = filePath.getName();
+    private static String getTopic(Path filePath, OffsetRangeSet seenFiles) {
+        if (filePath.toString().contains("+tmp")) {
+            return null;
+        }
 
+        String fileName = filePath.getName();
         // Skip if extension is not .avro
-        if (!FilenameUtils.getExtension(fileName).equals("avro")) {
-            logger.info("Skipped non-avro file: {}", fileName);
-            return;
+        if (!fileName.endsWith(".avro")) {
+            logger.info("Skipping non-avro file: {}", fileName);
+            return null;
         }
 
         OffsetRange range = OffsetRange.parse(fileName);
         // Skip already processed avro files
         if (seenFiles.contains(range)) {
-            return;
+            return null;
         }
 
-        logger.info("{}", filePath);
+        return filePath.getParent().getParent().getName();
+    }
+
+    private void processFile(Path filePath, String topicName, FileCacheStore cache,
+            OffsetRangeFile offsets) throws IOException {
+        logger.debug("Reading {}", filePath);
 
         // Read and parse avro file
         FsInput input = new FsInput(filePath, conf);
@@ -218,12 +230,12 @@ record = dataFileReader.next(record);
 
         // Write which file has been processed and update bins
         try {
+            OffsetRange range = OffsetRange.parse(filePath.getName());
             offsets.write(range);
             bins.write();
         } catch (IOException ex) {
             logger.warn("Failed to update status. Continuing processing.", ex);
         }
-        processedFileCount++;
     }
 
     private void writeRecord(GenericRecord record, String topicName, FileCacheStore cache)
@@ -273,4 +285,5 @@ public static String createHourTimestamp(GenericRecord valueField, Field timeFie
         Date date = new Date((long) (time * 1000d));
         return FILE_DATE_FORMAT.format(date);
     }
+
 }

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

@@ -0,0 +1,78 @@
+/*
+ * 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;
+
+/**
+ * Progress bar.
+ * Based on https://stackoverflow.com/a/43381186/574082.
+ */
+public class ProgressBar {
+    private final long total;
+    private final int numStripes;
+    private int previousPercentage;
+
+    public ProgressBar(long total, int numStripes) {
+        if (total < 0) {
+            throw new IllegalArgumentException("Total of progress bar must be positive");
+        }
+        if (numStripes <= 0) {
+            throw new IllegalArgumentException("Number of stripes in progress bar must be positive");
+        }
+        this.total = total;
+        this.numStripes = numStripes;
+        this.previousPercentage = -1;
+    }
+
+    public void update(long remain) {
+        if (remain > total || remain < 0) {
+            throw new IllegalArgumentException(
+                    "Update value " + remain + " out of range [0, " + total + "].");
+        }
+        int remainPercent;
+        if (total > 0) {
+            remainPercent = (int) ((100 * remain) / total);
+        } else {
+            remainPercent = 100;
+        }
+        if (remainPercent == previousPercentage) {
+            return;
+        }
+        previousPercentage = remainPercent;
+        int stripesFilled = remainPercent / numStripes;
+        char notFilled = '-';
+        char filled = '*';
+        // 2 init + numStripes + 2 end + 4 percentage
+        StringBuilder builder = new StringBuilder(numStripes + 8);
+        builder.append("\r[");
+        for (int i = 0; i < stripesFilled; i++) {
+            builder.append(filled);
+        }
+        for (int i = stripesFilled; i < numStripes; i++) {
+            builder.append(notFilled);
+        }
+        builder.append("] ").append(remainPercent).append('%');
+        if (remain < total) {
+            System.out.print(builder.toString());
+        } else {
+            System.out.println(builder.toString());
+        }
+    }
+
+    public boolean isDone() {
+        return previousPercentage == 100;
+    }
+}