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

Question

我有一个 Spark (Spark 1.5.2) 应用程序，可将数据从 Kafka 流式传输到 HDFS。我的应用程序包含两个 Typesafe 配置文件来配置某些东西，比如 Kafka 主题等。

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

因此，我排除了我的项目的配置文件，并通过“驱动程序类路径”添加了它们。这在客户端模式下有效，但如果我现在将配置文件移动到 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 参数试过了，但也没有用。有人知道我该如何解决这个问题吗？

更新：

我做了一些进一步的研究，发现它可能与 HDFS 路径有关。我将 HDFS 路径更改为“hdfs:///iot-master:8020//user//spark//config 但不幸的是，这也不起作用。但也许这可以帮助你。

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

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 文件将无法工作，除非您能够在之前运行 hdfs dfs -get <....> 来检索文件。就我而言，我想从 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)
}

P.S 错误仅表示 ConfigFactory 没有找到任何配置文件，因此他找不到您要查找的属性。

Answer 2

一种选择是使用--files 标志和HDFS 位置，并确保使用spark.executor.extraClassPath 标志和-Dconfig.file 将其添加到执行程序类路径中：

Spark 使用以下 URL 方案来允许不同的 传播罐子的策略：

• file: - 绝对路径和 file:/ URI 由驱动程序的 HTTP 提供 文件服务器，每个执行程序从驱动程序 HTTP 拉文件 服务器。
• hdfs:, http:, https:, ftp: - 这些下拉文件和 JAR 来自 URI 的预期
• local: - 以 local:/ 开头的 URI 期望作为本地文件存在于每个工作节点上。这意味着 不会产生网络 IO，并且适用于大文件/JAR 推送给每个工作人员，或通过 NFS、GlusterFS 等共享。

此外，您可以在查看spark-submit 的帮助文档时看到它：

--files FILES           Comma-separated list of files to be placed in the working
                        directory of each executor.

使用 spark-submit 运行：

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