Spark yarn-cluster 模式 - 读取通过 --files 传递的文件答案

【问题标题】：Spark yarn-cluster mode - read file passed with --filesSpark yarn-cluster 模式 - 读取通过 --files 传递的文件
【发布时间】：2015-11-20 15:57:23
【问题描述】：

我正在使用 yarn-cluster master 运行我的 spark 应用程序。

应用程序做什么？

外部服务根据对 RESTService 的 HTTP 请求生成 jsonFile
Spark 需要读取这个文件并在解析完 json 后做一些工作

想到的最简单的解决方案是使用 --files 来加载该文件。在 yarn-cluster 模式下读取文件意味着它必须在 hdfs 上可用（如果我是对的？）并且我的文件被复制到这样的路径：

/hadoop_user_path/.sparkStaging/spark_applicationId/myFile.json

我当然可以在哪里阅读它，但我找不到从任何配置/SparkEnv 对象获取此路径的方法。在 spark 代码中硬编码 .sparkStaging 就像一个坏主意。

为什么简单：

val jsonStringData = spark.textFile(myFileName)
sqlContext.read.json(jsonStringData)

无法读取通过 --files 传递的文件并抛出 FileNotFoundException？为什么 spark 只在 hadoop_user_folder 中查找文件？

我现在可行的解决方案：

在运行 spark 之前，我将文件复制到正确的 hdfs 文件夹，将文件名作为 Spark 参数传递，从已知路径处理文件，并在作业完成后删除 hdfs 格式的文件。

我认为将文件作为 --files 传递会让我忘记保存和删除该文件。类似于 pass-process-andforget。

那么你如何读取通过 --files 传递的文件呢？唯一的解决方案是手动创建路径，硬编码“.sparkStaging”文件夹路径？

【问题讨论】：

标签： hadoop apache-spark hdfs hadoop-yarn

【解决方案1】：

这个问题写得很含糊。但是，据我所知，您想从本地操作系统文件系统的任何位置读取文件，而不仅仅是从 HDFS。

Spark 使用 URI 来识别路径，并且在有效的 Hadoop/HDFS 环境可用的情况下，它将默认使用 HDFS。在这种情况下，要指向您的本地操作系统文件系统，例如 UNIX/LINUX，您可以使用以下内容：

file:///home/user/my_file.txt

如果你使用 RDD 来读取这个文件，你在 yarn-cluster 模式下运行，或者这个文件在一个任务中被访问，你需要手动复制和分发这个文件到你的所有节点集群，使用相同的路径。这就是首先将它放在 hfs 上变得容易的原因，或者这就是 --files 选项应该为您做的事情。

在Spark, External Datasets上查看更多信息。

对于通过--files 选项添加或通过SparkContext.addFile 添加的任何文件，您可以使用SparkFiles 辅助类获取有关其位置的信息。

【讨论】：

你说的只是部分正确。是的 - 在 yarn-cluster 上运行我可能需要手动将文件复制到 hdfs。但是使用 --files 它已经完成了->我的文件转到 ../.sparkStaging/applicationId/... 文件夹。我的问题是，如何轻松读取通过 --files 传递的文件，而无需硬编码 .sparkStaging 文件夹的路径。

【解决方案2】：

@hartar 的回答对我有用。这是完整的解决方案。

使用 --files 在 spark-submit 期间添加所需文件

spark-submit --name "my_job" --master yarn --deploy-mode cluster --files /home/xyz/file1.properties,/home/xyz/file2.properties --class test.main /home/xyz/my_test_jar.jar

在 main 方法中获取 spark session

SparkSession ss = new SparkSession.Builder().getOrCreate();

因为我只对 .properties 文件感兴趣，所以我正在过滤它，如果您知道要读取的文件名，那么它可以直接在 FileInputStream 中使用。

spark.yarn.dist.files 会将其存储为 file:/home/xyz/file1.properties,file:/home/xyz/file2.properties 因此用 (,) 和 (/) 分割字符串，这样我可以删除除文件名之外的其余内容。

String[] files = Pattern.compile("/|,").splitAsStream(ss.conf().get("spark.yarn.dist.files")).filter(s -> s.contains(".properties")).toArray(String[]::new);

//load all files to Property                
for (String f : files) {
    props.load(new FileInputStream(f));
}

【讨论】：

【解决方案3】：

我和你有同样的问题，其实你要知道，当你发送一个可执行文件和文件时，它们是在同一级别的，所以在你的可执行文件中，你只要把文件名放到Access就足够了因为你的可执行文件是基于它自己的文件夹。

您不需要使用 sparkFiles 或任何其他类。就像 readFile("myFile.json"); 之类的方法；

【讨论】：

【解决方案4】：

我遇到了一个简单的方法来做到这一点。我们在 Yarn 上以伪分布式模式使用 Spark 2.3.0。我们需要从 spark 查询 postgres 表，其配置在属性文件中定义。我使用 spark submit 的 --files 属性传递了属性文件。要在我的代码中读取文件，我只需使用 java.util.Properties.PropertiesReader 类。

我只需要确保我在加载文件时指定的路径与 --files 参数中传递的路径相同

例如如果火花提交命令看起来像： spark-submit --class --master yarn --deploy-mode client--files test/metadata.properties myjar.jar

然后我读取文件的代码将如下所示：属性 props = new Properties(); props.load(new FileInputStream(new File("test/metadata.properties")));

希望对您有所帮助。

【讨论】：