错误的FS加载json与来自s3的火花

Question

我正在尝试使用spark和magellan library加载geojson文件我的加载代码是：

val polygons = spark.read.format("magellan").option("type", "geojson").load(inJson)

其中inJson是我在s3上的json的路径：s3n：//bucket-name/geojsons/file.json

堆栈跟踪错误：

阶段0.0中的0.3（TID 3，ip-172-31-19-102.eu-west-1.compute.internal，executor 1）：java.lang.IllegalArgumentException：Wrong FS：s3n：// bucket-name / geojsons /file.json，expected：hdfs：//ip-172-31-27-182.eu-west-1.compute.internal：8020 at org.apache.hadoop.fs.FileSystem.checkPath（FileSystem.java:653 ）org.apache.hadoop.hdfs.DistributedFileSystem.getPathName（DistributedFileSystem.java:194）atg.apache.hadoop.hdfs.DistributedFileSystem.access $ 000（DistributedFileSystem.java:106）org.apache.hadoop.hdfs.DistributedFileSystem $ 3.doCall（DistributedFileSystem.java:304）org.apache.hadoop.hdfs.DistributedFileSystem $ 3.doCall（DistributedFileSystem.java:299）at org.apache.hadoop.fs.FileSystemLinkResolver.resolve（FileSystemLinkResolver.java:81）at org.apache.hadoop.hdfs.DistributedFileSystem.open（DistributedFileSystem.java:312）atg.apache.hadoop.fs.FileSystem.open（FileSystem.java:773）at magellan.mapreduce.WholeFileReader.nextKeyValue（WholeFileReader.scala： 45）在org.apache.spark.rdd.NewHad oopRDD $$ anon $ 1.hasNext（NewHadoopRDD.scala：199）at org.apache.spark.InterruptibleIterator.hasNext（InterruptibleIterator.scala：39）at scala.collection.Iterator $$ anon $ 12.hasNext（Iterator.scala：439） at scala.collection.Iterator $$ anon $ 11.hasNext（Iterator.scala：408）at scala.collection.Iterator $$ anon $ 11.hasNext（Iterator.scala：408）at scala.collection.Iterator $ class.foreach（Iterator .scala：893）scala.collection.AbstractIterator.foreach（Iterator.scala：1336）at scala.collection.TraversableOnce $ class.foldLeft（TraversableOnce.scala：157）at scala.collection.AbstractIterator.foldLeft（Iterator.scala： 1336）scala.collection.TraversableOnce $ class.fold（TraversableOnce.scala：212）at scala.collection.AbstractIterator.fold（Iterator.scala：1336）at org.apache.spark.rdd.RDD $$ anonfun $ fold $ 1 $$ anonfun $ 20.apply（RDD.scala：1086）org.apache.spark.rdd.RDD $$ anonfun $ fold $ 1 $$ anonfun $ 20.apply（RDD.scala：1086）at org.apache.spark.SparkContext来自org.apache.spark.SparkConte的$$ anonfun $ 33.apply（SparkContext.scala：1980） xt $$ anonfun $ 33.apply（SparkContext.scala：1980）org.apache.spark.scheduler.ResultTask.runTask（ResultTask.scala：87）at org.apache.spark.scheduler.Task.run（Task.scala： 99）atg.apache.spark.executor.Executor $ TaskRunner.run（Executor.scala：282）at java.util.concurrent.ThreadPoolExecutor.runWorker（ThreadPoolExecutor.java:1142）at java.util.concurrent.ThreadPoolExecutor $ Worker .run（ThreadPoolExecutor.java:617）在java.lang.Thread.run（Thread.java:745）

只有当我在多台计算机上运行它时才会出现问题，因此它在具有master和核心组中的1个实例的EMR集群上正常工作，但是在核心组中有10个实例则失败

Answer 1

这在Magellan WholeFileReader中是个问题。它正在获取默认的FileSystem。

它用this pull request解决了

解决方案是这样的：

-      val fs = FileSystem.get(conf)
+      val fs = path.getFileSystem(conf)

Answer 2

如果您在EMR上运行，则可以使用“s3：// bucket / path”而不是“s3n：// ....”