Apache Spark EOF 异常答案

【问题标题】：Apache Spark EOF exceptionApache Spark EOF 异常
【发布时间】：2014-06-18 19:48:35
【问题描述】：

运行读取文本文件并收集结果的简单作业时，我收到 EOFException。这在我的开发机器上运行良好，但在独立模式下执行时失败（单机，master+worker）。我的设置是预先构建的 Apache Spark 0.9.1 Hadoop 2。

我正在使用 sbt-assembly 插件部署我的代码并生成一个可执行的 jar 文件。

相关堆栈跟踪：

14/05/27 08:22:03 WARN scheduler.TaskSetManager: Loss was due to java.io.EOFException
java.io.EOFException
    at java.io.ObjectInputStream$BlockDataInputStream.readFully(ObjectInputStream.java:2742)
    at java.io.ObjectInputStream.readFully(ObjectInputStream.java:1014)
    at org.apache.hadoop.io.WritableUtils.readCompressedByteArray(WritableUtils.java:39)
    at org.apache.hadoop.io.WritableUtils.readCompressedString(WritableUtils.java:87)
    at org.apache.hadoop.io.WritableUtils.readCompressedStringArray(WritableUtils.java:185)
    at org.apache.hadoop.conf.Configuration.readFields(Configuration.java:2378)
    at org.apache.hadoop.io.ObjectWritable.readObject(ObjectWritable.java:285)
    at org.apache.hadoop.io.ObjectWritable.readFields(ObjectWritable.java:77)
    at org.apache.spark.SerializableWritable.readObject(SerializableWritable.scala:39)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:40)
    at org.apache.spark.broadcast.HttpBroadcast$.read(HttpBroadcast.scala:165)
    at org.apache.spark.broadcast.HttpBroadcast.readObject(HttpBroadcast.scala:56)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
    at sun.reflect.GeneratedMethodAccessor10.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
    at sun.reflect.GeneratedMethodAccessor10.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
    at sun.reflect.GeneratedMethodAccessor10.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
    at sun.reflect.GeneratedMethodAccessor9.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at scala.collection.immutable.$colon$colon.readObject(List.scala:362)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:622)
    at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1001)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1892)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:1989)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1914)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1797)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:40)
    at org.apache.spark.scheduler.ResultTask$.deserializeInfo(ResultTask.scala:63)
    at org.apache.spark.scheduler.ResultTask.readExternal(ResultTask.scala:139)
    at java.io.ObjectInputStream.readExternalData(ObjectInputStream.java:1836)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1795)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1349)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:369)
    at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:40)
    at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:62)
    at org.apache.spark.executor.Executor$TaskRunner$$anonfun$run$1.apply$mcV$sp(Executor.scala:193)
    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:416)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1491)
    at org.apache.spark.deploy.SparkHadoopUtil.runAsUser(SparkHadoopUtil.scala:41)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:176)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1146)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:701)

[编辑]

请注意我更改了序列化程序，现在我正在使用 Kryo（只是尝试过看看是否是问题所在）。

我的火花上下文：

//Load Spark config file
lazy val conf = ConfigFactory.load

//Set Spark config object
val sparkConf = new SparkConf()
      .setMaster(conf.getString("spark.prod.master"))  //Something like spark://host:port
      .setAppName(conf.getString("app.name"))
      .set("spark.executor.memory", conf.getString("spark.prod.config.executorMemory"))
      .set("spark.cores.max", conf.getString("spark.prod.config.coresMax"))
      .set("spark.serializer", conf.getString("spark.prod.config.serializer"))
      .set("spark.kryo.registrator", conf.getString("spark.prod.config.kryoRegistrator"))
      .set("spark.kryoserializer.buffer.mb", conf.getString("spark.prod.config.kryoSerializerBufferSize"))
      .set("spark.logConf", conf.getString("spark.prod.config.logConf"))

有什么提示吗？

【问题讨论】：

您是如何添加jar 文件的？你用SparkContext.addJar了吗？
@visakh 我将 master config 属性设置为 master 的 spark url (spark://whatever)，然后从 master 启动可执行 jar 文件。一切正常启动，但在提交一些任务后，我开始收到这些错误。
@visakh jar 文件包含一个 main 类，其中包含一个 main 方法，该方法执行所有与 spark 相关的代码。
如果不查看您用于读取文件的代码，我不确定是否可以回答这个问题。
@DanielDarabos 没什么特别的，我只是使用 sc.textFile 方法读取本地文本文件（我没有使用 hdfs）并玩一些地图/过滤器/等转换。

标签： scala hadoop apache-spark

【解决方案1】：

经过几天的努力，我终于想出了一个解决方案。我不得不添加相应的hadoop-client 依赖来避免这个奇怪的异常。

在那之后，其他一些已经reported errors 出现了。连接被拒绝问题的解决方法是：

更改sbin/start-master.sh 和/或sbin/start-slaves.sh 并将$SPARK_MASTER_IP 设置为hostname -f 而不是hostname。似乎 Akka 仅适用于完全限定名称，而不适用于主机名或 IP 地址。
还将conf/spark-env.sh 中的$SPARK_MASTER_IP 设置为hostname -f，以便集群工作人员可以访问主服务器。
确保conf/slaves 也使用完全限定的域名，而不是主机名/IP 地址。

在这些更改之后，一切正常。

希望对其他人有所帮助！

【讨论】：

您是如何找到解决方案的。 hadoop-client?