Merge pull request #38 from RADAR-base/release-0.5.4

Release 0.5.4

Author: blootsvoets

build.gradle

@@ -7,7 +7,7 @@ plugins {
 }
 
 group 'org.radarcns'
-version '0.5.3'
+version '0.5.4'
 mainClassName = 'org.radarcns.hdfs.Application'
 
 sourceCompatibility = '1.8'

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

@@ -114,6 +114,7 @@ public static void main(String [] args) {
                     .doDeduplicate(commandLineArgs.deduplicate)
                     .tempDir(commandLineArgs.tmpDir)
                     .numThreads(commandLineArgs.numThreads)
+                    .maxFilesPerTopic(commandLineArgs.maxFilesPerTopic)
                     .build();
 
             HdfsSettings hdfsSettings = new HdfsSettings.Builder(commandLineArgs.hdfsName)

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

@@ -26,6 +26,7 @@
 import java.time.Instant;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
@@ -63,6 +64,7 @@ public class RadarHdfsRestructure {
     private final Configuration conf;
     private final FileStoreFactory fileStoreFactory;
     private final RecordPathFactory pathFactory;
+    private final long maxFilesPerTopic;
 
     private LongAdder processedFileCount;
     private LongAdder processedRecordsCount;
@@ -71,6 +73,11 @@ public RadarHdfsRestructure(FileStoreFactory factory) {
         conf = factory.getHdfsSettings().getConfiguration();
         conf.set("fs.defaultFS", "hdfs://" + factory.getHdfsSettings().getHdfsName());
         this.numThreads = factory.getSettings().getNumThreads();
+        long maxFiles = factory.getSettings().getMaxFilesPerTopic();
+        if (maxFiles < 1) {
+            maxFiles = Long.MAX_VALUE;
+        }
+        this.maxFilesPerTopic = maxFiles;
         this.fileStoreFactory = factory;
         this.pathFactory = factory.getPathFactory();
     }
@@ -93,7 +100,7 @@ public void start(String directoryName) throws IOException {
 
             Instant timeStart = Instant.now();
             // Get filenames to process
-            TopicFileList topicPaths = getTopicPaths(fs, path, accountant.getOffsets());
+            List<TopicFileList> topicPaths = getTopicPaths(fs, path, accountant.getOffsets());
             logger.info("Time retrieving file list: {}",
                     formatTime(Duration.between(timeStart, Instant.now())));
 
@@ -104,12 +111,16 @@ public void start(String directoryName) throws IOException {
         }
     }
 
-    private TopicFileList getTopicPaths(FileSystem fs, Path path, OffsetRangeSet seenFiles) {
-        return new TopicFileList(walk(fs, path)
+    private List<TopicFileList> getTopicPaths(FileSystem fs, Path path, OffsetRangeSet seenFiles) {
+        Map<String, List<TopicFile>> topics = walk(fs, path)
                 .filter(f -> f.getName().endsWith(".avro"))
                 .map(f -> new TopicFile(f.getParent().getParent().getName(), f))
                 .filter(f -> !seenFiles.contains(f.range))
-                .collect(Collectors.toList()));
+                .collect(Collectors.groupingBy(TopicFile::getTopic));
+
+        return topics.values().stream()
+                .map(v -> new TopicFileList(v.stream().limit(maxFilesPerTopic)))
+                .collect(Collectors.toList());
     }
 
     private Stream<Path> walk(FileSystem fs, Path path) {
@@ -133,9 +144,16 @@ private Stream<Path> walk(FileSystem fs, Path path) {
                 });
     }
 
-    private void processPaths(TopicFileList topicPaths, Accountant accountant) throws InterruptedException {
+    private void processPaths(List<TopicFileList> topicPaths, Accountant accountant) throws InterruptedException {
+        int numFiles = topicPaths.stream()
+                .mapToInt(TopicFileList::numberOfFiles)
+                .sum();
+        long numOffsets = topicPaths.stream()
+                .mapToLong(TopicFileList::numberOfOffsets)
+                .sum();
+
         logger.info("Converting {} files with {} records",
-                topicPaths.files.size(), NumberFormat.getNumberInstance().format(topicPaths.size));
+                numFiles, NumberFormat.getNumberInstance().format(numOffsets));
 
         processedFileCount = new LongAdder();
         processedRecordsCount = new LongAdder();
@@ -144,14 +162,13 @@ private void processPaths(TopicFileList topicPaths, Accountant accountant) throw
 
         ExecutorService executor = Executors.newWorkStealingPool(pathFactory.isTopicPartitioned() ? this.numThreads : 1);
 
-        ProgressBar progressBar = new ProgressBar(topicPaths.size, 50, 500, TimeUnit.MILLISECONDS);
+        ProgressBar progressBar = new ProgressBar(numOffsets, 50, 500, TimeUnit.MILLISECONDS);
 
         // Actually process the files
-        topicPaths.files.stream()
-                .collect(Collectors.groupingBy(TopicFile::getTopic)).values().stream()
-                .map(TopicFileList::new)
+
+        topicPaths.stream()
                 // ensure that largest values go first on the executor queue
-                .sorted(Comparator.comparingLong(TopicFileList::getSize).reversed())
+                .sorted(Comparator.comparingLong(TopicFileList::numberOfOffsets).reversed())
                 .forEach(paths -> {
                     String size = NumberFormat.getNumberInstance().format(paths.size);
                     String topic = paths.files.get(0).topic;
@@ -186,7 +203,7 @@ private void processPaths(TopicFileList topicPaths, Accountant accountant) throw
 
         executor.shutdown();
         executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
-        progressBar.update(topicPaths.size);
+        progressBar.update(numOffsets);
     }
 
     private void processFile(TopicFile file, FileCacheStore cache,
@@ -260,14 +277,18 @@ private static class TopicFileList {
         private final List<TopicFile> files;
         private final long size;
 
-        public TopicFileList(List<TopicFile> files) {
-            this.files = files;
-            this.size = files.stream()
+        public TopicFileList(Stream<TopicFile> files) {
+            this.files = files.collect(Collectors.toList());
+            this.size = this.files.stream()
                     .mapToInt(TopicFile::size)
                     .sum();
         }
 
-        public long getSize() {
+        public int numberOfFiles() {
+            return this.files.size();
+        }
+
+        public long numberOfOffsets() {
             return size;
         }
     }

src/main/java/org/radarcns/hdfs/config/RestructureSettings.java

@@ -33,6 +33,7 @@ public class RestructureSettings {
     private final Path tempDir;
     private final Path outputPath;
     private final int numThreads;
+    private final int maxFilesPerTopic;
 
     private RestructureSettings(Builder builder) {
         this.compression = builder.compression;
@@ -42,6 +43,7 @@ private RestructureSettings(Builder builder) {
         this.tempDir = builder.tempDir;
         this.outputPath = builder.outputPath;
         this.numThreads = builder.numThreads;
+        this.maxFilesPerTopic = builder.maxFilesPerTopic;
     }
 
     public String getCompression() {
@@ -72,6 +74,10 @@ public int getNumThreads() {
         return this.numThreads;
     }
 
+    public int getMaxFilesPerTopic() {
+        return maxFilesPerTopic;
+    }
+
     public static class Builder {
         private int numThreads = 1;
         private String compression;
@@ -80,6 +86,7 @@ public static class Builder {
         private int cacheSize = CACHE_SIZE_DEFAULT;
         private Path tempDir;
         private final Path outputPath;
+        public int maxFilesPerTopic;
 
         public Builder(String outputPath) {
             this.outputPath = Paths.get(outputPath.replaceAll("/+$", ""));
@@ -124,6 +131,11 @@ public Builder numThreads(int num) {
             return this;
         }
 
+        public Builder maxFilesPerTopic(int num) {
+            this.maxFilesPerTopic = num;
+            return this;
+        }
+
         public RestructureSettings build() {
             compression = nonNullOrDefault(compression, () -> "identity");
             format = nonNullOrDefault(format, () -> "csv");

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

@@ -90,6 +90,9 @@ public class CommandLineArgs {
     @Parameter(names = {"-s", "--cache-size"}, description = "Number of files to keep in cache in a single thread.", validateWith = PositiveInteger.class)
     public int cacheSize = CACHE_SIZE_DEFAULT;
 
+    @Parameter(names = {"--max-files-per-topic"}, description = "Maximum number of records to process, per topic. Set below 1 to disable this option.")
+    public int maxFilesPerTopic = 0;
+
     public static <T> T nonNullOrDefault(T value, Supplier<T> defaultValue) {
         return value != null ? value : defaultValue.get();
     }