scala - Cluster 模式下使用 Spark 写入文件到本地系统

Question

我知道这是一种使用 Spark 的奇怪方式，但我正在尝试使用 Spark 将数据帧保存到本地文件系统(不是 hdfs)，即使我处于集群模式。我知道我可以使用客户端模式，但我确实想在集群模式下运行并且不关心哪个节点(3 个中的)应用程序将作为驱动程序运行。下面的代码是我正在尝试做的伪代码。

// create dataframe
val df = Seq(Foo("John", "Doe"), Foo("Jane", "Doe")).toDF()
// save it to the local file system using 'file://' because it defaults to hdfs://
df.coalesce(1).rdd.saveAsTextFile(s"file://path/to/file")

这就是我提交 spark 应用程序的方式。

spark-submit --class sample.HBaseSparkRSample --master yarn 集群 hbase-spark-r-sample-assembly-1.0.jar

如果我处于 local mode 但不是 yarn-cluster mode，这工作正常。

例如，java.io.IOException: Mkdirs failed to create file 发生在上面的代码中。

我已将 df.coalesce(1) 部分更改为 df.collect 并尝试使用普通 Scala 保存文件，但它以 权限被拒绝。

我也试过:

• spark-submit 与 root 用户
• chowned yarn:yarn, yarn:hadoop, spark:spark
• 给相关目录chmod 777

但没有运气。

我假设这必须与 clusters、drivers and executors 以及尝试写入本地的 user 相关文件系统，但我几乎无法自己解决这个问题。

我正在使用:

• 星火:1.6.0-cdh5.8.2
• 斯卡拉:2.10.5
• Hadoop:2.6.0-cdh5.8.2

欢迎任何支持，并提前致谢。

我试过的一些文章:

• “Spark saveAsTextFile() 导致 Mkdirs 无法为目录的一半创建”-> 尝试更改用户但没有任何改变
• “无法将 RDD 作为文本文件保存到本地文件系统”-> chmod 没有帮助我

编辑(2016/11/25)

这是我得到的异常。

java.io.IOException: Mkdirs failed to create file:/home/foo/work/rhbase/r/input/input.csv/_temporary/0/_temporary/attempt_201611242024_0000_m_000000_0 (exists=false, cwd=file:/yarn/nm/usercache/foo/appcache/application_1478068613528_0143/container_e87_1478068613528_0143_01_000001)
    at org.apache.hadoop.fs.ChecksumFileSystem.create(ChecksumFileSystem.java:449)
    at org.apache.hadoop.fs.ChecksumFileSystem.create(ChecksumFileSystem.java:435)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:920)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:813)
    at org.apache.hadoop.mapred.TextOutputFormat.getRecordWriter(TextOutputFormat.java:135)
    at org.apache.spark.SparkHadoopWriter.open(SparkHadoopWriter.scala:91)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1$$anonfun$13.apply(PairRDDFunctions.scala:1193)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1$$anonfun$13.apply(PairRDDFunctions.scala:1185)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)
16/11/24 20:24:12 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, localhost): java.io.IOException: Mkdirs failed to create file:/home/foo/work/rhbase/r/input/input.csv/_temporary/0/_temporary/attempt_201611242024_0000_m_000000_0 (exists=false, cwd=file:/yarn/nm/usercache/foo/appcache/application_1478068613528_0143/container_e87_1478068613528_0143_01_000001)
    at org.apache.hadoop.fs.ChecksumFileSystem.create(ChecksumFileSystem.java:449)
    at org.apache.hadoop.fs.ChecksumFileSystem.create(ChecksumFileSystem.java:435)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:920)
    at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:813)
    at org.apache.hadoop.mapred.TextOutputFormat.getRecordWriter(TextOutputFormat.java:135)
    at org.apache.spark.SparkHadoopWriter.open(SparkHadoopWriter.scala:91)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1$$anonfun$13.apply(PairRDDFunctions.scala:1193)
    at org.apache.spark.rdd.PairRDDFunctions$$anonfun$saveAsHadoopDataset$1$$anonfun$13.apply(PairRDDFunctions.scala:1185)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

我要回答我自己的问题，因为最终，没有一个答案似乎没有解决我的问题。尽管如此，感谢您提供所有答案，并指出我可以检查的替代方案。

我认为 @Ricardo 最接近于提及 Spark 应用程序的用户。我用 Process("whoami") 检查了 whoami，用户是 yarn。问题可能是我试图输出到 /home/foo/work/rhbase/r/input/input.csv 并且尽管 /home/foo/work/rhbase属于yarn:yarn，/home/foo 属于foo:foo。我没有详细检查，但这可能是导致此 permission 问题的原因。

当我使用 Process("pwd") 在我的 Spark 应用程序中点击 pwd 时，它输出 /yarn/path/to/somewhere .所以我决定将我的文件输出到 /yarn/input.csv，尽管在​​ 集群模式 下它还是成功的。

我大概可以断定这只是一个简单的权限问题。欢迎任何进一步的解决方案，但就目前而言，这就是我解决这个问题的方式。