在Spark中使用FileUtil.copy将文件从HDFS上传到S3，导致DiskErrorException：目录不可写错误？

l0oc07j2  于 2022-12-09  发布在  HDFS
关注(0)|答案(1)|浏览(196)

我正在尝试将一个 parquet 文件写入HDFS，然后将其复制到S3。
我用齐柏林飞艇写的代码，运行得很好。没有任何问题，它把文件添加到了S3文件路径中。

  1. var outputFolder = "buckent_name/path"
  2.   println("\n ---- TASK 1 ----- \n writing with path " + outputFolder)
  3.   wholeParquetFile
  4.     .withColumn("date_col", to_date(col("timestamp"), "YYYYMMdd"))
  5.     .withColumn("year", year(col("date_col")))
  6.     .withColumn("month", month(col("date_col")))
  7.     .withColumn("day", dayofmonth(col("date_col")))
  8.     .drop("date_col")
  9.     .repartition(1)
  10.     .write.mode(SaveMode.Overwrite)
  11.     .partitionBy("year", "month", "day")
  12.     .parquet(outputFolder)
  14.   val sc = spark.sparkContext
  15.   val fs = FileSystem.get(sc.hadoopConfiguration)
  16.   val allTheFilesThatBennCreated: Array[FileStatus] = fs.globStatus(new Path(outputFolder + "/year=*/month=*/day=*/*"))
  18.   println("------- allTheFilesThatBennCreated   -------" + allTheFilesThatBennCreated.mkString("Array(", ", ", ")"))
  20.   // right now the file path will be outputFile + "/year=2021/month=5/day=17/part-....c000.snappy.parquet
  21.   // converting it to                outputFile + "/2021/5/17/part-....c000.snappy.parquet"
  22.   allTheFilesThatBennCreated.foreach(path => {
  24.     val newPathString = generateOutputFilePathString(path.getPath.toString)
  25.     val outputFilePath = new Path(newPathString)
  26.     val destinationFileSystem = FileSystem.get(outputFilePath.toUri, sc.hadoopConfiguration)
  27.     val sourceFileSystem = FileSystem.get(path.getPath.toUri, sc.hadoopConfiguration)
  29.     println("-------- source filesystem ------------------" + sourceFileSystem)
  31.     println("-------- path.getPath --------------" + path.getPath)
  33.     println("-------- destinationFileSystem ------------- " + destinationFileSystem)
  35.     println("-------- S3 path for Output File ------------" + outputFilePath)
  37.     // uploading to s3 from hdfs
  38.     FileUtil.copy(sourceFileSystem, path.getPath, destinationFileSystem, outputFilePath,true, sc.hadoopConfiguration)
  39.   })

但是当我尝试在spark-shell中运行相同的代码或通过spark-submit中的jar文件运行相同的代码时，就会出现此错误。

  1. 22/05/17 09:57:28 WARN LocalDirAllocator$AllocatorPerContext: /mnt/var/lib/hadoop/tmp/s3a is not writable
  3. org.apache.hadoop.util.DiskChecker$DiskErrorException: Directory is not writable: /mnt/var/lib/hadoop/tmp/s3a
  4.     at org.apache.hadoop.util.DiskChecker.checkAccessByFileMethods(DiskChecker.java:167)
  5.     at org.apache.hadoop.util.DiskChecker.checkDirInternal(DiskChecker.java:100)
  6.     at org.apache.hadoop.util.DiskChecker.checkDir(DiskChecker.java:77)
  7.     at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.confChanged(LocalDirAllocator.java:315)
  8.     at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:378)
  9.     at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.createTmpFileForWrite(LocalDirAllocator.java:461)
  10.     at org.apache.hadoop.fs.LocalDirAllocator.createTmpFileForWrite(LocalDirAllocator.java:200)
  11.     at org.apache.hadoop.fs.s3a.S3AFileSystem.createTmpFileForWrite(S3AFileSystem.java:501)
  12.     at org.apache.hadoop.fs.s3a.S3AOutputStream.<init>(S3AOutputStream.java:66)
  13.     at org.apache.hadoop.fs.s3a.S3AFileSystem.create(S3AFileSystem.java:690)
  14.     at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1075)
  15.     at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1056)
  16.     at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:945)
  17.     at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:393)
  18.     at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:366)
  19.     at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:316)
  20.     at com.propellyr.driver.ApplicationMain$$anonfun$2.apply(ApplicationMain.scala:86)
  21.     at com.propellyr.driver.ApplicationMain$$anonfun$2.apply(ApplicationMain.scala:70)
  22.     at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
  23.     at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
  24.     at com.propellyr.driver.ApplicationMain$.delayedEndpoint$com$propellyr$driver$ApplicationMain$1(ApplicationMain.scala:70)
  25.     at com.propellyr.driver.ApplicationMain$delayedInit$body.apply(ApplicationMain.scala:7)
  26.     at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
  27.     at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
  28.     at scala.App$$anonfun$main$1.apply(App.scala:76)
  29.     at scala.App$$anonfun$main$1.apply(App.scala:76)
  30.     at scala.collection.immutable.List.foreach(List.scala:392)
  31.     at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
  32.     at scala.App$class.main(App.scala:76)
  33.     at com.propellyr.driver.ApplicationMain$.main(ApplicationMain.scala:7)
  34.     at com.propellyr.driver.ApplicationMain.main(ApplicationMain.scala)
  35.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  36.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  37.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  38.     at java.lang.reflect.Method.invoke(Method.java:498)
  39.     at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
  40.     at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:863)
  41.     at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:161)
  42.     at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:184)
  43.     at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
  44.     at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:938)
  45.     at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:947)
  46.     at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  47. Exception in thread "main" org.apache.hadoop.util.DiskChecker$DiskErrorException: No space available in any of the local directories.
  48.     at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:400)
  49.     at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.createTmpFileForWrite(LocalDirAllocator.java:461)
  50.     at org.apache.hadoop.fs.LocalDirAllocator.createTmpFileForWrite(LocalDirAllocator.java:200)
  51.     at org.apache.hadoop.fs.s3a.S3AFileSystem.createTmpFileForWrite(S3AFileSystem.java:501)
  52.     at org.apache.hadoop.fs.s3a.S3AOutputStream.<init>(S3AOutputStream.java:66)
  53.     at org.apache.hadoop.fs.s3a.S3AFileSystem.create(S3AFileSystem.java:690)
  54.     at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1075)
  55.     at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:1056)
  56.     at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:945)
  57.     at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:393)
  58.     at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:366)
  59.     at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:316)
  60.     at com.propellyr.driver.ApplicationMain$$anonfun$2.apply(ApplicationMain.scala:86)
  61.     at com.propellyr.driver.ApplicationMain$$anonfun$2.apply(ApplicationMain.scala:70)
  62.     at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
  63.     at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
  64.     at com.propellyr.driver.ApplicationMain$.delayedEndpoint$com$propellyr$driver$ApplicationMain$1(ApplicationMain.scala:70)
  65.     at com.propellyr.driver.ApplicationMain$delayedInit$body.apply(ApplicationMain.scala:7)
  66.     at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
  67.     at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
  68.     at scala.App$$anonfun$main$1.apply(App.scala:76)
  69.     at scala.App$$anonfun$main$1.apply(App.scala:76)
  70.     at scala.collection.immutable.List.foreach(List.scala:392)
  71.     at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
  72.     at scala.App$class.main(App.scala:76)
  73.     at com.propellyr.driver.ApplicationMain$.main(ApplicationMain.scala:7)
  74.     at com.propellyr.driver.ApplicationMain.main(ApplicationMain.scala)
  75.     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  76.     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  77.     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  78.     at java.lang.reflect.Method.invoke(Method.java:498)
  79.     at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
  80.     at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:863)
  81.     at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:161)
  82.     at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:184)
  83.     at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
  84.     at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:938)
  85.     at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:947)
  86.     at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

有人知道怎么解决这个问题吗？

hdfs

1条答案

按热度按时间
bwleehnv

bwleehnv1#

您没有提到Zeplin和CLI的环境，无论它们是提交到同一群集还是您的CLI使用本地模式
不过，线索在堆栈跟踪中
任何本地目录中都没有可用空间。
进一步查看，错误出现在FileUtil.copy()处，它试图将临时输出写入由属性mapred.local.dir配置的路径，您可以检查该属性

赞(0）分享  回复(0）举报  2022-12-09
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com