Merge pull request #46 from RADAR-base/asService

Run the HDFS restructuring as a service

Author: blootsvoets

.travis.yml

@@ -1,4 +1,6 @@
 language: java
+# Force use of dist where oraclejdk8 is supported
+dist: trusty
 jdk:
   - oraclejdk8
 sudo: false

README.md

@@ -60,6 +60,8 @@ By default, files records are not deduplicated after writing. To enable this beh
 
 To set the output user ID and group ID, specify the `-p local-uid=123` and `-p local-gid=12` properties.
 
+To run the output generator as a service that will regularly poll the HDFS directory, add the `--service` flag and optionally the `--interval` flag to adjust the polling interval.
+
 ## Extending the connector
 
 To implement alternative storage paths, storage drivers or storage formats, put your custom JAR in

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

@@ -31,6 +31,9 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import org.radarcns.hdfs.accounting.Accountant;
 import org.radarcns.hdfs.config.HdfsSettings;
 import org.radarcns.hdfs.config.RestructureSettings;
@@ -58,10 +61,15 @@ public class Application implements FileStoreFactory {
     private final RecordPathFactory pathFactory;
     private final List<String> inputPaths;
     private final RestructureSettings settings;
+    private final int pollInterval;
+    private final boolean isService;
+    private RadarHdfsRestructure hdfsReader;
 
     private Application(Builder builder) {
         this.storageDriver = builder.storageDriver;
         this.settings = builder.settings;
+        this.isService = builder.asService;
+        this.pollInterval = builder.pollInterval;
 
         converterFactory = builder.formatFactory.get(settings.getFormat());
         compression = builder.compressionFactory.get(settings.getCompression());
@@ -131,6 +139,8 @@ public static void main(String [] args) {
                     .storageDriver(commandLineArgs.storageDriver)
                     .properties(commandLineArgs.properties)
                     .inputPaths(commandLineArgs.inputPaths)
+                    .asService(commandLineArgs.asService)
+                    .pollInterval(commandLineArgs.pollInterval)
                     .build();
         } catch (IllegalArgumentException ex) {
             logger.error("HDFS High availability name node configuration is incomplete."
@@ -193,10 +203,41 @@ public void start() {
         System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism",
                 String.valueOf(settings.getNumThreads() - 1));
 
-        Instant timeStart = Instant.now();
-        RadarHdfsRestructure hdfsReader = new RadarHdfsRestructure(this);
         try {
             Files.createDirectories(settings.getTempDir());
+        } catch (IOException ex) {
+            logger.error("Failed to create temporary directory");
+            return;
+        }
+
+        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+        executorService.execute(() -> hdfsReader = new RadarHdfsRestructure(this));
+
+        if (isService) {
+            logger.info("Press Ctrl+C to exit...");
+            executorService.scheduleAtFixedRate(this::runRestructure,
+                    pollInterval / 4, pollInterval, TimeUnit.SECONDS);
+        } else {
+            executorService.execute(this::runRestructure);
+        }
+
+        try {
+            Thread.sleep(Long.MAX_VALUE);
+        } catch (InterruptedException e) {
+            logger.info("Interrupted, shutting down...");
+            executorService.shutdownNow();
+            try {
+                executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+                Thread.currentThread().interrupt();
+            } catch (InterruptedException ex) {
+                logger.info("Interrupted again...");
+            }
+        }
+    }
+
+    private void runRestructure() {
+        Instant timeStart = Instant.now();
+        try {
             for (String input : inputPaths) {
                 logger.info("In:  {}", input);
                 logger.info("Out: {}", pathFactory.getRoot());
@@ -220,6 +261,8 @@ public static class Builder {
         private FormatFactory formatFactory;
         private Map<String, String> properties = new HashMap<>();
         private List<String> inputPaths;
+        private boolean asService;
+        private int pollInterval;
 
         public Builder(RestructureSettings settings) {
             this.settings = settings;
@@ -256,6 +299,16 @@ public Builder properties(Map<String, String> props) {
             return this;
         }
 
+        public Builder asService(boolean asService) {
+            this.asService = asService;
+            return this;
+        }
+
+        public Builder pollInterval(int pollInterval) {
+            this.pollInterval = pollInterval;
+            return this;
+        }
+
         public Application build() throws IOException {
             pathFactory = nonNullOrDefault(pathFactory, ObservationKeyPathFactory::new);
             pathFactory.init(properties);

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

@@ -96,6 +96,12 @@ public class CommandLineArgs {
     @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<>();
 
+    @Parameter(names = {"-S", "--service"}, description = "Run the output generation as a service")
+    public boolean asService = false;
+
+    @Parameter(names = {"-i", "--interval"}, description = "Polling interval when running as a service (seconds)")
+    public int pollInterval = 3600;
+
     public static <T> T nonNullOrDefault(T value, Supplier<T> defaultValue) {
         return value != null ? value : defaultValue.get();
     }