Merge pull request #548 from RADAR-base/remove-hdfs

Remove HDFS from code

Author: blootsvoets

README.md

@@ -1,7 +1,7 @@
 # Restructure Kafka connector output files
 
 Data streamed by a Kafka Connector will be converted to a RADAR-base oriented output directory, by organizing it by project, user and collection date.
-It supports data written by [RADAR S3 sink connector](https://github.com/RADAR-base/RADAR-S3-Connector) is streamed to files based on topic name only. This package transforms that output to a local directory structure as follows: `projectId/userId/topic/date_hour.csv`. The date and hour are extracted from the `time` field of each record, and is formatted in UTC time. This package is included in the [RADAR-Docker](https://github.com/RADAR-base/RADAR-Docker) repository, in the `dcompose/radar-cp-hadoop-stack/bin/hdfs-restructure` script.
+It supports data written by [RADAR S3 sink connector](https://github.com/RADAR-base/RADAR-S3-Connector) is streamed to files based on topic name only. This package transforms that output to a local directory structure as follows: `projectId/userId/topic/date_hour.csv`. The date and hour are extracted from the `time` field of each record, and is formatted in UTC time.
 
 ## Upgrade instructions
 
@@ -90,7 +90,7 @@ By default, this will output the data in CSV format. If JSON format is preferred
 radar-output-restructure --format json --output-directory <output_folder>  <input_path_1> [<input_path_2> ...]
 ```
 
-By default, files records are not deduplicated after writing. To enable this behaviour, specify the option `--deduplicate` or `-d`. This set to false by default because of an issue with Biovotion data. Please see - [issue #16](https://github.com/RADAR-base/Restructure-HDFS-topic/issues/16) before enabling it. Deduplication can also be enabled or disabled per topic using the config file. If lines should be deduplicated using a subset of fields, e.g. only `sourceId` and `time` define a unique record and only the last record with duplicate values should be kept, then specify `topics: <topicName>: deduplication: distinctFields: [key.sourceId, value.time]`.
+By default, files records are not deduplicated after writing. To enable this behaviour, specify the option `--deduplicate` or `-d`. This set to false by default because of an issue with Biovotion data. Please see - [issue #16](https://github.com/RADAR-base/radar-output-restructure/issues/16) before enabling it. Deduplication can also be enabled or disabled per topic using the config file. If lines should be deduplicated using a subset of fields, e.g. only `sourceId` and `time` define a unique record and only the last record with duplicate values should be kept, then specify `topics: <topicName>: deduplication: distinctFields: [key.sourceId, value.time]`.
 
 ### Compression
 

restructure.yml

@@ -7,7 +7,7 @@ service:
 # Source data resource
 # @since: 0.7.0
 source:
-  type: s3  # hdfs, azure or s3
+  type: s3  # azure or s3
   s3:
     endpoint: http://localhost:9000  # using AWS S3 endpoint is also possible.
     bucket: radar
@@ -31,9 +31,6 @@ source:
     # If true, try to read the metadata property "endOffset" to determine the
     # final offset of an input object.
     #endOffsetFromMetadata: false
-  # only actually needed if source type is hdfs
-  hdfs:
-    nameNodes: [hdfs-namenode-1, hdfs-namenode-2]
   index:
     # Interval to fully synchronize the index with the storage
     fullSyncInterval: 3600
@@ -80,7 +77,7 @@ compression:
   # Compression type: none, zip or gzip
   type: gzip
   # Compression Factory class
-  # factory: org.radarbase.hdfs.compression.CompressionFactory
+  # factory: org.radarbase.output.compression.CompressionFactory
   # Additional compression properties
   # properties: {}
 
@@ -96,7 +93,7 @@ format:
     # Ignore specific fields to consider records distinct. Disregarded if empty.
     # ignoreFields: []
   # Format factory class
-  # factory: org.radarbase.hdfs.format.FormatFactory
+  # factory: org.radarbase.output.format.FormatFactory
   # Additional format properties
   # properties: {}
   # Do not write certain fields to file

src/main/java/org/radarbase/output/cleaner/TimestampExtractionCheck.kt

@@ -24,7 +24,7 @@ class TimestampExtractionCheck(
         val result = resourceContext {
             val input = createResource { reader.newInput(file) }
             // processing zero-length files may trigger a stall. See:
-            // https://github.com/RADAR-base/Restructure-HDFS-topic/issues/3
+            // https://github.com/RADAR-base/radar-output-restructure/issues/3
             if (input.length() == 0L) {
                 logger.warn("File {} has zero length, skipping.", file.path)
                 return@resourceContext false

src/main/java/org/radarbase/output/config/CommandLineArgs.kt

@@ -51,20 +51,13 @@ class CommandLineArgs {
     )
     var compression: String? = null
 
-    // Default set to false because causes loss of records from Biovotion data. https://github.com/RADAR-base/Restructure-HDFS-topic/issues/16
+    // Default set to false because causes loss of records from Biovotion data. https://github.com/RADAR-base/radar-output-restructure/issues/16
     @Parameter(
         names = ["-d", "--deduplicate"],
         description = "Boolean to define if to use deduplication or not.",
     )
     var deduplicate: Boolean? = null
 
-    @Parameter(
-        names = ["-n", "--nameservice"],
-        description = "The HDFS name services to connect to. Eg - '<HOST>' for single configurations or <CLUSTER_ID> for high availability web services.",
-        validateWith = [NonEmptyValidator::class],
-    )
-    var hdfsName: String? = null
-
     @Parameter(
         names = ["-o", "--output-directory"],
         description = "The output folder where the files are to be extracted.",

src/main/java/org/radarbase/output/config/HdfsConfig.kt

@@ -20,7 +20,7 @@ data class PathConfig(
     val path: PathFormatterConfig = PathFormatterConfig(),
     /**
      * Bucket formatting rules for the target storage. If no configuration is provided, this
-     * will not format any bucket for local or HDFS storage, and it will use the target bucket (s3)
+     * will not format any bucket for local storage, and it will use the target bucket (s3)
      * or container (azure) as the default target bucket.
      */
     val bucket: BucketFormatterConfig? = null,

src/main/java/org/radarbase/output/config/PathConfig.kt

@@ -5,31 +5,28 @@ import org.radarbase.output.config.ResourceType.Companion.toResourceType
 import org.radarbase.output.config.RestructureConfig.Companion.copyOnChange
 
 data class ResourceConfig(
-    /** Resource type. One of s3, hdfs or local. */
+    /** Resource type. One of s3, azure or local. */
     val type: String,
     val s3: S3Config? = null,
-    val hdfs: HdfsConfig? = null,
     val local: LocalConfig? = null,
     val azure: AzureConfig? = null,
     val index: StorageIndexConfig? = null,
 ) {
     @get:JsonIgnore
     val sourceType: ResourceType by lazy {
-        requireNotNull(type.toResourceType()) { "Unknown resource type $type, choose s3, hdfs or local" }
+        requireNotNull(type.toResourceType()) { "Unknown resource type $type, choose s3, azure or local" }
     }
 
     fun validate() {
         when (sourceType) {
             ResourceType.S3 -> checkNotNull(s3) { "No S3 configuration provided." }
-            ResourceType.HDFS -> checkNotNull(hdfs) { "No HDFS configuration provided." }.also { it.validate() }
             ResourceType.LOCAL -> checkNotNull(local) { "No local configuration provided." }
             ResourceType.AZURE -> checkNotNull(azure) { "No Azure configuration provided." }
         }
     }
 
     fun withEnv(prefix: String): ResourceConfig = when (sourceType) {
         ResourceType.S3 -> copyOnChange(s3, { it?.withEnv(prefix) }) { copy(s3 = it) }
-        ResourceType.HDFS -> this
         ResourceType.LOCAL -> this
         ResourceType.AZURE -> copyOnChange(azure, { it?.withEnv(prefix) }) { copy(azure = it) }
     }

src/main/java/org/radarbase/output/config/ResourceConfig.kt

@@ -1,12 +1,11 @@
 package org.radarbase.output.config
 
 enum class ResourceType {
-    S3, HDFS, LOCAL, AZURE;
+    S3, LOCAL, AZURE;
 
     companion object {
         fun String.toResourceType() = when (lowercase()) {
             "s3" -> S3
-            "hdfs" -> HDFS
             "local" -> LOCAL
             "azure" -> AZURE
             else -> null

src/main/java/org/radarbase/output/config/ResourceType.kt

@@ -43,14 +43,6 @@ data class RestructureConfig(
         args.tmpDir?.let { copy(paths = paths.copy(temp = Paths.get(it))) }
         args.inputPaths?.let { inputs -> copy(paths = paths.copy(inputs = inputs.map { Paths.get(it) })) }
         args.outputDirectory?.let { copy(paths = paths.copy(output = Paths.get(it))) }
-        args.hdfsName?.let {
-            copy(
-                source = source.copy(
-                    hdfs = source.hdfs?.copy(nameNodes = listOf(it))
-                        ?: HdfsConfig(nameNodes = listOf(it)),
-                ),
-            )
-        }
         args.format?.let { copy(format = format.copy(type = it)) }
         args.deduplicate?.let {
             copy(format = format.copy(deduplication = format.deduplication.copy(enable = it)))

src/main/java/org/radarbase/output/config/RestructureConfig.kt

@@ -31,14 +31,6 @@ class SourceStorageFactory(
             val minioClient = requireNotNull(s3SourceClient) { "Missing S3 client configuration for source storage" }
             S3SourceStorage(minioClient, s3Config, tempPath)
         }
-        ResourceType.HDFS -> {
-            val storage = Class.forName("org.radarbase.output.source.HdfsSourceStorageFactory")
-            val constructor =
-                storage.getDeclaredConstructor(ResourceConfig::class.java, Path::class.java)
-            val factory = constructor.newInstance(resourceConfig, tempPath)
-            val createSourceStorage = storage.getDeclaredMethod("createSourceStorage")
-            createSourceStorage.invoke(factory) as SourceStorage
-        }
         ResourceType.AZURE -> {
             val azureClient = requireNotNull(azureSourceClient) { "Missing Azure client configuration for source storage" }
             val azureConfig = requireNotNull(resourceConfig.azure) { "Missing Azure configuration for source storage" }