使用Spark在集羣模式下將文件寫入本地系統

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。

我也試過：

• ​​與root用戶
• chown版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

歡迎任何支持，並提前致謝。

有的文章我已經試過：

• 「星火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，它仍然成功。

我大概可以得出結論，這只是一個簡單的權限問題。任何進一步的解決方案將是受歡迎的，但現在，這是我如何解決這個問題的方式。

Answer 2

請參閱火花文檔來了解在​​使用--master選項。

• --master local應該在本地運行時使用。

• --master yarn --deploy-mode cluster應該在實際在紗線羣上運行時使用。

參照this和this。

Answer 3

檢查您是否嘗試使用非Spark服務的用戶運行/寫入文件。 在這種情況下，您可以通過預設目錄ACL來解決權限問題。例如：

setfacl -d -m group:spark:rwx /path/to/ 

（修改「火花」到您的用戶組試圖寫入文件）

Answer 4

使用forEachPartition方法，然後爲每個分區獲取文件系統對象，寫一個接一個地記錄它，下面是這裏的示例代碼，我寫信給HDFS，而不是如果您運行作業，yarn-cluster mode您可以使用本地文件系統以及

Dataset<String> ds=.... 

ds.toJavaRdd.foreachPartition(new VoidFunction<Iterator<String>>() { 
    @Override 
    public void call(Iterator<String> iterator) throws Exception { 

    final FileSystem hdfsFileSystem = FileSystem.get(URI.create(finalOutPathLocation), hadoopConf); 

    final FSDataOutputStream fsDataOutPutStream = hdfsFileSystem.exists(finalOutPath) 
      ? hdfsFileSystem.append(finalOutPath) : hdfsFileSystem.create(finalOutPath); 


    long processedRecCtr = 0; 
    long failedRecsCtr = 0; 


    while (iterator.hasNext()) { 

     try { 
      fsDataOutPutStream.writeUTF(iterator.next); 
     } catch (Exception e) { 
      failedRecsCtr++; 
     } 
     if (processedRecCtr % 3000 == 0) { 
      LOGGER.info("Flushing Records"); 
      fsDataOutPutStream.flush(); 
     } 
    } 

    fsDataOutPutStream.close(); 
     } 
}); 

Answer 5

，司機會在其中任何一個被管理的機器上運行YARN，所以如果saveAsTextFile有位置l文件路徑，那麼它會將輸出存儲在驅動程序運行的任何機器上。

嘗試運行工作作爲yarn-client mode使駕駛員在客戶機