如何将位于HDFS上的类型安全配置文件添加到spark-submit（集群模式）？

Question

现在我想在一个集群中用spark-submit（集群模式）运行我的应用程序。我的项目的所有依赖项的jar文件都存储在HDFS上。只要我的配置文件包含在jar文件中，一切都可以正常工作。但这对于测试目的来说是不实际的，因为我总是要重新构建JAR。

因此，我排除了项目的配置文件，并通过“driver-class-path”添加了它们。这在客户端模式下工作，但如果我现在将配置文件移动到HDFS并在集群模式下运行应用程序，它将无法找到设置。下面你可以找到我的spark-submit命令：

/usr/local/spark/bin/spark-submit \
    --total-executor-cores 10 \
    --executor-memory 15g \
    --verbose \
    --deploy-mode cluster\
    --class com.hdp.speedlayer.SpeedLayerApp \
    --driver-class-path hdfs://iot-master:8020/user/spark/config \
    --master spark://spark-master:6066 \
    hdfs://iot-master:8020/user/spark/speed-layer-CONFIG.jar

我已经用--file参数试过了，但也不起作用。有没有人知道我该怎么解决这个问题？

下面你还可以看到我在集群模式下运行驱动程序时得到的错误：

Exception in thread "main" java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:497)
    at org.apache.spark.deploy.worker.DriverWrapper$.main(DriverWrapper.scala:58)
    at org.apache.spark.deploy.worker.DriverWrapper.main(DriverWrapper.scala)
Caused by: java.lang.ExceptionInInitializerError
    at com.speedlayer.SpeedLayerApp.main(SpeedLayerApp.scala)
    ... 6 more
Caused by: com.typesafe.config.ConfigException$Missing: No configuration setting found for key 'application'
    at com.typesafe.config.impl.SimpleConfig.findKey(SimpleConfig.java:124)
    at com.typesafe.config.impl.SimpleConfig.find(SimpleConfig.java:145)
    at com.typesafe.config.impl.SimpleConfig.find(SimpleConfig.java:159)
    at com.typesafe.config.impl.SimpleConfig.find(SimpleConfig.java:164)
...

Answer 1

为了达到同样的结果，我发现了以下几点：

1. --files:仅与运行spark-submit命令的计算机上的本地文件关联，并转换为conf.addfile()。因此，除非您能够在检索文件之前运行hdfs dfs-get<...>,否则hdfs文件将无法工作。在我的例子中，我想从oozie运行它，所以我不知道它将在哪台机器上运行，而且我不想在我的工作流中添加复制文件操作。
2. @yuval_itzchakov引用的引用是指--jars，因为它将转换为conf.addjar()

因此，据我所知，从HDFS加载配置文件是不可能的。

private val HDFS_IMPL_KEY = "fs.hdfs.impl"
def loadConf(pathToConf: String): Config = {
   val path = new Path(pathToConf)
   val confFile = File.createTempFile(path.getName, "tmp")
   confFile.deleteOnExit()
   getFileSystemByUri(path.toUri).copyToLocalFile(path, new Path(confFile.getAbsolutePath))

   ConfigFactory.load(ConfigFactory.parseFile(confFile))
}

def getFileSystemByUri(uri: URI) : FileSystem  = {
   val hdfsConf = new Configuration()
   hdfsConf.set(HDFS_IMPL_KEY, classOf[org.apache.hadoop.hdfs.DistributedFileSystem].getName)
FileSystem.get(uri, hdfsConf)
}