Spark 在 java.io.IOException 的大洗牌作业上失败：文件系统已关闭答案

【问题标题】：Spark fails on big shuffle jobs with java.io.IOException: Filesystem closedSpark 在 java.io.IOException 的大洗牌作业上失败：文件系统已关闭
【发布时间】：2014-07-25 04:33:59
【问题描述】：

我经常发现 spark 在大型作业中失败，并出现一个相当无用且无意义的异常。工作日志看起来很正常，没有错误，但它们的状态为“KILLED”。这对于大型 shuffle 非常常见，因此像 .distinct 这样的操作。

问题是，我如何诊断出了什么问题，理想情况下，我该如何解决？

鉴于很多这些操作都是单曲面的，我一直在解决这个问题，方法是将数据分成 10 个块，在每个块上运行应用程序，然后在所有结果输出上运行应用程序。换句话说 - meta-map-reduce。

14/06/04 12:56:09 ERROR client.AppClient$ClientActor: Master removed our application: FAILED; stopping client
14/06/04 12:56:09 WARN cluster.SparkDeploySchedulerBackend: Disconnected from Spark cluster! Waiting for reconnection...
14/06/04 12:56:09 WARN scheduler.TaskSetManager: Loss was due to java.io.IOException
java.io.IOException: Filesystem closed
    at org.apache.hadoop.hdfs.DFSClient.checkOpen(DFSClient.java:703)
    at org.apache.hadoop.hdfs.DFSInputStream.readWithStrategy(DFSInputStream.java:779)
    at org.apache.hadoop.hdfs.DFSInputStream.read(DFSInputStream.java:840)
    at java.io.DataInputStream.read(DataInputStream.java:149)
    at org.apache.hadoop.io.compress.DecompressorStream.getCompressedData(DecompressorStream.java:159)
    at org.apache.hadoop.io.compress.DecompressorStream.decompress(DecompressorStream.java:143)
    at org.apache.hadoop.io.compress.DecompressorStream.read(DecompressorStream.java:85)
    at java.io.InputStream.read(InputStream.java:101)
    at org.apache.hadoop.util.LineReader.fillBuffer(LineReader.java:180)
    at org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:216)
    at org.apache.hadoop.util.LineReader.readLine(LineReader.java:174)
    at org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:209)
    at org.apache.hadoop.mapred.LineRecordReader.next(LineRecordReader.java:47)
    at org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:164)
    at org.apache.spark.rdd.HadoopRDD$$anon$1.getNext(HadoopRDD.scala:149)
    at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71)
    at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:27)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
    at scala.collection.Iterator$class.foreach(Iterator.scala:727)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
    at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:48)
    at scala.collection.mutable.ListBuffer.$plus$plus$eq(ListBuffer.scala:176)
    at scala.collection.mutable.ListBuffer.$plus$plus$eq(ListBuffer.scala:45)
    at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:273)
    at scala.collection.AbstractIterator.to(Iterator.scala:1157)
    at scala.collection.TraversableOnce$class.toList(TraversableOnce.scala:257)
    at scala.collection.AbstractIterator.toList(Iterator.scala:1157)
    at $line5.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$2.apply(<console>:13)
    at $line5.$read$$iwC$$iwC$$iwC$$iwC$$anonfun$2.apply(<console>:13)
    at org.apache.spark.rdd.RDD$$anonfun$1.apply(RDD.scala:450)
    at org.apache.spark.rdd.RDD$$anonfun$1.apply(RDD.scala:450)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:34)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:232)
    at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:232)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:34)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:241)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:232)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:161)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:102)
    at org.apache.spark.scheduler.Task.run(Task.scala:53)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:213)
    at org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:42)
    at org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:41)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1548)
    at org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:178)
    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:744)

【问题讨论】：

里面有几行是关于压缩的——at org.apache.hadoop.io.compress.DecompressorStream.getCompressedData(DecompressorStream.java:159)你的数据是如何存储在HDFS中的？
我认为这个特殊的工作是从 gz 读取的......但我们也使用 bzip2 和plain。

标签： scala hadoop hdfs apache-spark

【解决方案1】：

截至 2014 年 9 月 1 日，这是 Spark 中的“公开改进”。请参阅https://issues.apache.org/jira/browse/SPARK-3052。正如 syrza 在给定链接中指出的那样，当执行程序失败时，关闭挂钩可能以错误的顺序完成，从而导致此消息。我了解您将需要进行更多调查才能找出问题的主要原因（即您的执行人失败的原因）。如果是大洗牌，则可能是内存不足错误，导致执行程序失败，然后导致 Hadoop 文件系统在其关闭挂钩中关闭。因此，该执行程序的运行任务中的 RecordReaders 抛出“java.io.IOException：文件系统关闭”异常。我想它会在后续版本中修复，然后你会得到更多有用的错误消息:)

【讨论】：

内存不足错误似乎是我的问题。我提高了我的 spark.executor.memory 并解决了这个问题。

【解决方案2】：

调用DFSClient.close() 或DFSClient.abort()，关闭客户端。然后下一个文件操作会导致上述异常。

我会试着弄清楚什么叫close()/abort()。您可以在调试器中使用断点，或修改 Hadoop 源代码以在这些方法中引发异常，这样您将获得堆栈跟踪。

【讨论】：

【解决方案3】：

如果 Spark 作业在集群上运行，则可以解决“文件系统关闭”的异常。您可以将 spark.executor.cores 、 spark.driver.cores 和 spark.akka.threads 等属性设置为资源可用性的最大值。当我的数据集非常大，包含大约 2000 万条记录的 JSON 数据时，我遇到了同样的问题。我用上述属性修复了它，它运行起来就像一个魅力。就我而言，我将这些属性分别设置为 25,25 和 20。希望对你有帮助！！

参考链接：

http://spark.apache.org/docs/latest/configuration.html

【讨论】：