support reading config file from hdfs (#106)

* support reading config file from hdfs * fix test
vesoft-inc · Nov 9, 2022 · 5879b25 · 5879b25
1 parent b86efc5
commit 5879b25
Show file tree

Hide file tree

Showing 11 changed files with 26 additions and 15 deletions.
diff --git a/exchange-common/src/main/scala/com/vesoft/exchange/common/config/Configs.scala b/exchange-common/src/main/scala/com/vesoft/exchange/common/config/Configs.scala
@@ -5,13 +5,15 @@
 
 package com.vesoft.exchange.common.config
 
-import java.io.File
+import java.io.{File, InputStreamReader}
 import java.nio.file.Files
 
 import com.google.common.net.HostAndPort
 import com.typesafe.config.{Config, ConfigFactory}
 import com.vesoft.exchange.Argument
 import com.vesoft.exchange.common.KeyPolicy
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FSDataInputStream, FileSystem, Path}
 import org.apache.log4j.Logger
 
 import scala.collection.mutable
@@ -255,12 +257,21 @@ object Configs {
     * @param configPath
     * @return
     */
-  def parse(configPath: File): Configs = {
-    if (!Files.exists(configPath.toPath)) {
-      throw new IllegalArgumentException(s"${configPath} not exist")
+  def parse(configPath: String): Configs = {
+    var config: Config = null
+    if (configPath.startsWith("hdfs://")) {
+      val hadoopConfig: Configuration = new Configuration()
+      val fs: FileSystem              = org.apache.hadoop.fs.FileSystem.get(hadoopConfig)
+      val file: FSDataInputStream     = fs.open(new Path(configPath))
+      val reader                      = new InputStreamReader(file)
+      config = ConfigFactory.parseReader(reader)
+    } else {
+      if (!Files.exists(new File(configPath).toPath)) {
+        throw new IllegalArgumentException(s"${configPath} not exist")
+      }
+      config = ConfigFactory.parseFile(new File(configPath))
     }
 
-    val config        = ConfigFactory.parseFile(configPath)
     val nebulaConfig  = config.getConfig("nebula")
     val addresses     = nebulaConfig.getStringList("address.graph").asScala.toList
     val metaAddresses = nebulaConfig.getStringList("address.meta").asScala.toList

diff --git a/exchange-common/src/test/scala/com/vesoft/exchange/common/config/ConfigsSuite.scala b/exchange-common/src/test/scala/com/vesoft/exchange/common/config/ConfigsSuite.scala
@@ -42,7 +42,7 @@ class ConfigsSuite {
     assert(c.hive)
     assert(c.directly)
 
-    val configs             = Configs.parse(new File(c.config))
+    val configs             = Configs.parse(c.config)
     val dataBaseConfigEntry = configs.databaseConfig
     val userConfig          = configs.userConfig
     val connectionConfig    = configs.connectionConfig

diff --git a/nebula-exchange_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala b/nebula-exchange_spark_2.2/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala
@@ -70,7 +70,7 @@ object Exchange {
         sys.exit(-1)
     }
 
-    val configs = Configs.parse(new File(c.config))
+    val configs = Configs.parse(c.config)
     LOG.info(s"Config ${configs}")
 
     val session = SparkSession

diff --git a/...ge_spark_2.2/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala b/...ge_spark_2.2/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala
@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
 
 class EdgeProcessorSuite {
   val config: Configs =
-    Configs.parse(new File("../exchange-common/src/test/resources/process_application.conf"))
+    Configs.parse("../exchange-common/src/test/resources/process_application.conf")
 
   var data: DataFrame             = null
   var edgeConfig: EdgeConfigEntry = config.edgesConfig.head

diff --git a/...park_2.2/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala b/...park_2.2/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala
@@ -36,7 +36,7 @@ class VerticesProcessorSuite {
   private[this] lazy val LOG = Logger.getLogger(this.getClass)
 
   val config: Configs =
-    Configs.parse(new File("../exchange-common/src/test/resources/process_application.conf"))
+    Configs.parse("../exchange-common/src/test/resources/process_application.conf")
 
   var data: DataFrame           = null
   var tagConfig: TagConfigEntry = config.tagsConfig.head

diff --git a/nebula-exchange_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala b/nebula-exchange_spark_2.4/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala
@@ -70,7 +70,7 @@ object Exchange {
         sys.exit(-1)
     }
 
-    val configs = Configs.parse(new File(c.config))
+    val configs = Configs.parse(c.config)
     LOG.info(s"Config ${configs}")
 
     val session = SparkSession

diff --git a/...ge_spark_2.4/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala b/...ge_spark_2.4/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala
@@ -34,7 +34,7 @@ import scala.collection.JavaConverters._
 
 class EdgeProcessorSuite {
   val config: Configs =
-    Configs.parse(new File("../exchange-common/src/test/resources/process_application.conf"))
+    Configs.parse("../exchange-common/src/test/resources/process_application.conf")
 
   var data: DataFrame             = null
   var edgeConfig: EdgeConfigEntry = config.edgesConfig.head

diff --git a/...park_2.4/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala b/...park_2.4/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala
@@ -34,7 +34,7 @@ import scala.collection.JavaConverters._
 
 class VerticesProcessorSuite {
   val config: Configs =
-    Configs.parse(new File("../exchange-common/src/test/resources/process_application.conf"))
+    Configs.parse("../exchange-common/src/test/resources/process_application.conf")
 
   var data: DataFrame           = null
   var tagConfig: TagConfigEntry = config.tagsConfig.head

diff --git a/nebula-exchange_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala b/nebula-exchange_spark_3.0/src/main/scala/com/vesoft/nebula/exchange/Exchange.scala
@@ -70,7 +70,7 @@ object Exchange {
         sys.exit(-1)
     }
 
-    val configs = Configs.parse(new File(c.config))
+    val configs = Configs.parse(c.config)
     LOG.info(s"Config ${configs}")
 
     val session = SparkSession

diff --git a/...ge_spark_3.0/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala b/...ge_spark_3.0/src/test/scala/com/vesoft/nebula/exchange/processor/EdgeProcessorSuite.scala
@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
 
 class EdgeProcessorSuite {
   val config: Configs =
-    Configs.parse(new File("../exchange-common/src/test/resources/process_application.conf"))
+    Configs.parse("../exchange-common/src/test/resources/process_application.conf")
 
   var data: DataFrame             = null
   var edgeConfig: EdgeConfigEntry = config.edgesConfig.head

diff --git a/...park_3.0/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala b/...park_3.0/src/test/scala/com/vesoft/nebula/exchange/processor/VerticesProcessorSuite.scala
@@ -33,7 +33,7 @@ import scala.collection.JavaConverters._
 
 class VerticesProcessorSuite {
   val config: Configs =
-    Configs.parse(new File("../exchange-common/src/test/resources/process_application.conf"))
+    Configs.parse("../exchange-common/src/test/resources/process_application.conf")
 
   var data: DataFrame           = null
   var tagConfig: TagConfigEntry = config.tagsConfig.head