Spark:将RDD[List[String,List[Person]]]中的List[Person]通过spark api保存为hdfs文件时一直出现not serializable task，没办法找到"spark自定义Kryo序列化输入输出API"

　在Spark中内置支持两种系列化格式：（1）、Java serialization；（2）、Kryo serialization。在默认情况下，Spark使用的是Java的ObjectOutputStream系列化框架，它支持所有继承java.io.Serializable的类系列化，虽然Java系列化非常灵活，但是它的性能不佳。然而我们可以使用Kryo 库来系列化，它相比Java serialization系列化高效，速度很快（通常比Java快10x），但是它不支持所有的系列化对象，而且要求用户注册类。

　　在Spark中，使用Kryo系列化比使用Java系列化更明智。在shuffling和caching大量数据的情况下，使用 Kryo系列化就变得非常重要。

　　虽然Kryo支持对RDD的cache和shuffle，但是在Spark中不是内置就显示提供使用Kryo将数据系列化到磁盘中的输入输出API，RDD中的saveAsObjectFile和SparkContext中的objectFile方法仅仅支持使用Java系列化。所以如果我们可以使用Kryo系列化将会变得很棒！

实现代码：

import java.sql.Timestamp

import java.text.SimpleDateFormat

import java.util.Calendar

import org.apache.spark.api.java.JavaPairRDD

import org.apache.spark.api.java.function.PairFunction

import org.apache.spark.sql.functions._

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.sql.hive.HiveContext

import java.io._

import com.esotericsoftware.kryo.io.Input

import org.apache.hadoop.conf._

import org.apache.hadoop.fs._

import org.apache.hadoop.fs.Path._

import org.apache.hadoop.conf.Configuration

import org.apache.hadoop.fs.FileSystem

import org.apache.hadoop.fs.Path

import org.apache.hadoop.fs.permission.FsAction

import org.apache.hadoop.fs.permission.FsPermission

import org.apache.hadoop.fs.FSDataOutputStream

import org.apache.hadoop.io.{BytesWritable, NullWritable}

import org.apache.spark.rdd.RDD

import org.apache.spark.serializer.KryoSerializer

import scala.reflect.ClassTag

// user defined class that need to serialized

class Person(val name: String)

/**

  * Created by Administrator on 11/10/2017.

  */

object TestSaveClasToHdfs{

  def saveAsObjectFile[T: ClassTag](rdd: RDD[T], path: String) {

    val kryoSerializer = new KryoSerializer(rdd.context.getConf)

    rdd.mapPartitions(iter => iter.grouped()

      .map(_.toArray))

      .map(splitArray => {

        //initializes kyro and calls your registrator class

        val kryo = kryoSerializer.newKryo()

        //convert data to bytes

        val bao = new ByteArrayOutputStream()

        val output = kryoSerializer.newKryoOutput()

        output.setOutputStream(bao)

        kryo.writeClassAndObject(output, splitArray)

        output.close()

        // We are ignoring key field of sequence file

        val byteWritable = new BytesWritable(bao.toByteArray)

        (NullWritable.get(), byteWritable)

      }).saveAsSequenceFile(path)

  }

  def objectFile[T](sc: SparkContext, path: String, minPartitions: Int = )(implicit ct: ClassTag[T]) = {

    val kryoSerializer = new KryoSerializer(sc.getConf)

    sc.sequenceFile(path, classOf[NullWritable], classOf[BytesWritable],

      minPartitions)

      .flatMap(x => {

        val kryo = kryoSerializer.newKryo()

        val input = new Input()

        input.setBuffer(x._2.getBytes)

        val data = kryo.readClassAndObject(input)

        val dataObject = data.asInstanceOf[Array[T]]

        dataObject

      })

  }

  def main(args: Array[String]) {

    if (args.length < ) {

      println("Please provide output path")

      return

    }

    val conf = new SparkConf().setMaster("local").setAppName("kryoexample")

    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)

    val outputPath = args()

    //create some dummy data

    val personList =  to  map (value => new Person(value + ""))

    val personRDD = sc.makeRDD(personList)

    saveAsObjectFile(personRDD, outputPath)

    val rdd = objectFile[Person](sc, outputPath)

    println(rdd.map(person => person.name).collect().toList)

  }

}

在spark-shell中执行时，一直出现错误，但是当我把它编译为jar包使用spark-submit命令提交时，错误就没有了。