使用Spark in Cluster模式将文件写入本地系统

Question

我知道这是一种使用Spark的奇怪方式,但我正在尝试使用Spark将数据帧保存到本地文件系统(而不是hdfs),即使我在cluster mode.我知道我可以使用,client mode但我确实想要运行cluster mode并且不关心应用程序将作为驱动程序运行的节点(总共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-cluster 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保存文件,但最终得到了一个Permission denied.

我也尝试过:

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

但没有运气.

我假设这必须做的东西clusters,drivers and executors和user谁在试图写入本地文件系统,但我几乎被困在由自己解决这个问题.

我正在使用:

• Spark: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应用程序的用户.我检查whoami过Process("whoami")并且用户是yarn.问题可能是我试图输出到/home/foo/work/rhbase/r/input/input.csv虽然/home/foo/work/rhbase是由拥有yarn:yarn,但由... /home/foo拥有foo:foo.我没有详细检查,但这可能是导致这个permission问题的原因.

当我点击pwd我的Spark应用程序时Process("pwd"),它会输出/yarn/path/to/somewhere.所以我决定输出我的文件,/yarn/input.csv尽管在那里它很成功cluster mode.

我可能会得出结论,这只是一个简单的权限问题.任何进一步的解决方案都会受到欢迎,但就目前而言,这就是我解决这个问题的方式.