使用 Spark newAPIHadoopFile(FileInputFormat) 遇到 NotSerializableException

Question

我正在使用 spark 2.1.0 和 hadoop 2.7.3。

我正在尝试使用 newAPIHadoopFile，非常简单的代码，只在一个类中使用 main 方法：

val spark = SparkSession.builder().appName("test").master("local[*]").getOrCreate()
    val sparkContext = spark.sparkContext
    val sparkConf = sparkContext.getConf
    val file = "src/main/resources/chat.csv"
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    sparkContext.getConf.registerKryoClasses(Array(
      Class.forName("org.apache.hadoop.io.LongWritable"),
      Class.forName("org.apache.hadoop.io.Text")
    ));
    sparkConf.set("spark.kryo.classesToRegister", "org.apache.hadoop.io.LongWritable, org.apache.hadoop.io.Text")
    val rdd = sparkContext.newAPIHadoopFile(file, classOf[KeyValueTextInputFormat], classOf[Text], classOf[Text])
    rdd.collect().foreach(println)

我在 StackOverflow 中查看了很多帖子，但仍然出现错误：

java.io.NotSerializableException: org.apache.hadoop.io.Text

Serialization stack:

    - object not serializable (class: org.apache.hadoop.io.Text, value:   How about Italian?"})

    - field (class: scala.Tuple2, name: _1, type: class java.lang.Object)

    - object (class scala.Tuple2, (  How about Italian?"},))

    - element of array (index: 0)

    - array (class [Lscala.Tuple2;, size 3)

    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)

编辑：chat.csv 的内容：

{from:"Gert", to:"Melissa", message:"Want to have dinner?"}
{from:"Melissa", to:"Gert", message:"Ok\
How about Italian?"}

Answer 1

我在我的机器上运行你用 spark 1.6.0 提供的相同代码，它工作正常（我在 HDFS 中使用制表符分隔文件）。您使用的是什么版本的 spark/hadoop？

在某些情况下（可能是较低版本）文本不实现可序列化。所以在收集它之前映射到 String 并重试，

如下所示，newAPIhadoopFile 的结果将是一个键值元组，

rdd.map(x=>(x._1.toString,x._2.toString)).collect().foreach(println)

试试这个。

参考：click here