Merge pull request #42 from RADAR-base/release-0.5.5

Release 0.5.5

Author: blootsvoets

README.md

@@ -8,7 +8,7 @@ Data streamed to HDFS using the [RADAR HDFS sink connector](https://github.com/R
 
 This package is available as docker image [`radarbase/radar-hdfs-restructure`](https://hub.docker.com/r/radarbase/radar-hdfs-restructure). The entrypoint of the image is the current application. So in all of the commands listed in usage, replace `radar-hdfs-restructure` with for example:
 ```shell
-docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-hdfs-restructure:0.5.3 -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-hdfs-restructure:0.5.5 -n hdfs-namenode -o /output /myTopic
 ```
 if your docker cluster is running in the `hadoop` network and your output directory should be `./output`.
 
@@ -23,7 +23,7 @@ This package requires at least Java JDK 8. Build the distribution with
 and install the package into `/usr/local` with for example
 ```shell
 sudo mkdir -p /usr/local
-sudo tar -xzf build/distributions/radar-hdfs-restructure-0.5.3.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-hdfs-restructure-0.5.5.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-hdfs-restructure` command should be available.

build.gradle

@@ -2,12 +2,12 @@ plugins {
     id 'java'
     id 'java-library'
     id 'application'
-    id 'com.jfrog.bintray' version '1.8.1'
+    id 'com.jfrog.bintray' version '1.8.4'
     id 'maven-publish'
 }
 
 group 'org.radarcns'
-version '0.5.4'
+version '0.5.5'
 mainClassName = 'org.radarcns.hdfs.Application'
 
 sourceCompatibility = '1.8'
@@ -25,6 +25,7 @@ ext {
     hadoopVersion = '3.0.3'
     jCommanderVersion = '1.72'
     almworksVersion = '1.1.1'
+    junitVersion = '5.4.0-M1'
 }
 
 repositories {
@@ -43,7 +44,9 @@ dependencies {
 
     runtimeOnly group: 'org.apache.hadoop', name: 'hadoop-hdfs-client', version: hadoopVersion
 
-    testImplementation group: 'junit', name: 'junit', version: '4.12'
+    testCompile group: 'org.junit.jupiter', name: 'junit-jupiter-api', version: junitVersion
+    testCompile group: 'org.junit.jupiter', name: 'junit-jupiter-params', version: junitVersion
+    testRuntime group: 'org.junit.jupiter', name: 'junit-jupiter-engine', version: junitVersion
 }
 
 ext.sharedManifest = manifest {
@@ -80,9 +83,17 @@ startScripts {
     }
 }
 
+test {
+    useJUnitPlatform()
+    testLogging {
+        events "passed", "skipped", "failed"
+    }
+}
+
+
 tasks.withType(Tar){
     compression = Compression.GZIP
-    extension = 'tar.gz'
+    archiveExtension.set('tar.gz')
 }
 
 task downloadDependencies {
@@ -93,18 +104,18 @@ task downloadDependencies {
 
 task copyDependencies(type: Copy) {
     from configurations.runtimeClasspath.files
-    into "${buildDir}/third-party/"
+    into "$buildDir/third-party/"
 }
 
 // custom tasks for creating source/javadoc jars
 task sourcesJar(type: Jar, dependsOn: classes) {
-    classifier = 'sources'
+    archiveClassifier.set('sources')
     from sourceSets.main.allSource
     manifest.from sharedManifest
 }
 
 task javadocJar(type: Jar, dependsOn: javadoc) {
-    classifier = 'javadoc'
+    archiveClassifier.set('javadoc')
     from javadoc.destinationDir
     manifest.from sharedManifest
 }
@@ -179,5 +190,5 @@ bintray {
 }
 
 wrapper {
-    gradleVersion '4.10.2'
+    gradleVersion '5.2.1'
 }

gradle/wrapper/gradle-wrapper.jar

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-5.2.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradle/wrapper/gradle-wrapper.properties

@@ -28,7 +28,7 @@ APP_NAME="Gradle"
 APP_BASE_NAME=`basename "$0"`
 
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-DEFAULT_JVM_OPTS=""
+DEFAULT_JVM_OPTS='"-Xmx64m"'
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"

gradlew

@@ -14,7 +14,7 @@ set APP_BASE_NAME=%~n0
 set APP_HOME=%DIRNAME%
 
 @rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-set DEFAULT_JVM_OPTS=
+set DEFAULT_JVM_OPTS="-Xmx64m"
 
 @rem Find java.exe
 if defined JAVA_HOME goto findJavaFromJavaHome

gradlew.bat

@@ -1,2 +1 @@
 rootProject.name = 'radar-hdfs-restructure'
-enableFeaturePreview('STABLE_PUBLISHING')

settings.gradle

@@ -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/Application.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018 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.hdfs;
+
+import static java.time.ZoneOffset.UTC;
+
+import java.time.format.DateTimeFormatter;
+
+public class MonthlyObservationKeyPathFactory extends ObservationKeyPathFactory {
+    private static final DateTimeFormatter PER_MONTH_HOURLY_TIME_BIN_FORMAT = DateTimeFormatter.ofPattern("yyyyMM/yyyyMMdd_HH'00'")
+            .withZone(UTC);
+
+
+    public DateTimeFormatter getTimeBinFormat() {
+        return PER_MONTH_HOURLY_TIME_BIN_FORMAT;
+    }
+}

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

@@ -58,13 +58,14 @@ public class RadarHdfsRestructure {
     private static final Logger logger = LoggerFactory.getLogger(RadarHdfsRestructure.class);
 
     /** Number of offsets to process in a single task. */
-    private static final int BATCH_SIZE = 500_000;
+    private static final long BATCH_SIZE = 500_000;
 
     private final int numThreads;
     private final Configuration conf;
     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));
 
@@ -174,8 +177,8 @@ private void processPaths(List<TopicFileList> topicPaths, Accountant accountant)
                     String topic = paths.files.get(0).topic;
                     logger.info("Processing {} records for topic {}", size, topic);
                     executor.execute(() -> {
-                        int batchSize = (int)(BATCH_SIZE * ThreadLocalRandom.current().nextDouble(0.75, 1.25));
-                        int currentSize = 0;
+                        long batchSize = Math.round(BATCH_SIZE * ThreadLocalRandom.current().nextDouble(0.75, 1.25));
+                        long currentSize = 0;
                         try (FileCacheStore cache = fileStoreFactory.newFileCacheStore(accountant)) {
                             for (TopicFile file : paths.files) {
                                 try {
@@ -280,7 +283,7 @@ private static class TopicFileList {
         public TopicFileList(Stream<TopicFile> files) {
             this.files = files.collect(Collectors.toList());
             this.size = this.files.stream()
-                    .mapToInt(TopicFile::size)
+                    .mapToLong(TopicFile::size)
                     .sum();
         }
 
@@ -308,8 +311,8 @@ public String getTopic() {
             return topic;
         }
 
-        public int size() {
-            return 1 + (int) (range.getOffsetTo() - range.getOffsetFrom());
+        public long size() {
+            return 1 + range.getOffsetTo() - range.getOffsetFrom();
         }
     }
 }