Merge pull request #40 from RADAR-base/exclude_option

Add option to exclude topics

Author: yatharthranjan

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

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

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

@@ -65,6 +65,7 @@ public class RadarHdfsRestructure {
     private final FileStoreFactory fileStoreFactory;
     private final RecordPathFactory pathFactory;
     private final long maxFilesPerTopic;
+    private List<String> excludeTopics;
 
     private LongAdder processedFileCount;
     private LongAdder processedRecordsCount;
@@ -78,6 +79,7 @@ public RadarHdfsRestructure(FileStoreFactory factory) {
             maxFiles = Long.MAX_VALUE;
         }
         this.maxFilesPerTopic = maxFiles;
+        this.excludeTopics = factory.getSettings().getExcludeTopics();
         this.fileStoreFactory = factory;
         this.pathFactory = factory.getPathFactory();
     }
@@ -115,6 +117,7 @@ private List<TopicFileList> getTopicPaths(FileSystem fs, Path path, OffsetRangeS
         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 -> !excludeTopics.contains(f.topic))
                 .filter(f -> !seenFiles.contains(f.range))
                 .collect(Collectors.groupingBy(TopicFile::getTopic));
 

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

@@ -24,6 +24,8 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
 
 public class RestructureSettings {
     private final String compression;
@@ -34,6 +36,7 @@ public class RestructureSettings {
     private final Path outputPath;
     private final int numThreads;
     private final int maxFilesPerTopic;
+    private final List<String> excludeTopics;
 
     private RestructureSettings(Builder builder) {
         this.compression = builder.compression;
@@ -44,6 +47,7 @@ private RestructureSettings(Builder builder) {
         this.outputPath = builder.outputPath;
         this.numThreads = builder.numThreads;
         this.maxFilesPerTopic = builder.maxFilesPerTopic;
+        this.excludeTopics = builder.excludeTopics;
     }
 
     public String getCompression() {
@@ -78,6 +82,10 @@ public int getMaxFilesPerTopic() {
         return maxFilesPerTopic;
     }
 
+    public List<String> getExcludeTopics() {
+        return excludeTopics;
+    }
+
     public static class Builder {
         private int numThreads = 1;
         private String compression;
@@ -87,6 +95,7 @@ public static class Builder {
         private Path tempDir;
         private final Path outputPath;
         public int maxFilesPerTopic;
+        private List<String> excludeTopics;
 
         public Builder(String outputPath) {
             this.outputPath = Paths.get(outputPath.replaceAll("/+$", ""));
@@ -136,6 +145,11 @@ public Builder maxFilesPerTopic(int num) {
             return this;
         }
 
+        public Builder excludeTopics(List<String> topics) {
+            this.excludeTopics = topics;
+            return this;
+        }
+
         public RestructureSettings build() {
             compression = nonNullOrDefault(compression, () -> "identity");
             format = nonNullOrDefault(format, () -> "csv");
@@ -146,6 +160,8 @@ public RestructureSettings build() {
                     throw new UncheckedIOException("Cannot create temporary directory", ex);
                 }
             });
+            excludeTopics = nonNullOrDefault(excludeTopics, ArrayList::new);
+
             return new RestructureSettings(this);
         }
     }

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

@@ -93,6 +93,9 @@ public class CommandLineArgs {
     @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;
 
+    @Parameter(names = {"--exclude-topic"}, description = "Topic to exclude when processing the records. Can be supplied more than once to exclude multiple topics.")
+    public List<String> excludeTopics = new ArrayList<>();
+
     public static <T> T nonNullOrDefault(T value, Supplier<T> defaultValue) {
         return value != null ? value : defaultValue.get();
     }