org.apache.spark.SparkException：任务不可序列化 - 传递 RDD

Question

我有以下三个课程，我正在学习

任务未序列化

错误。完整的堆栈跟踪见下文。

第一类是一个序列化的人：

public class Person implements Serializable 
  {
      private String name;
      private int age;

      public String getName() 
      {
        return name;
      }

      public void setAge(int age) 
      {
        this.age = age;
      }
    }

这个类从文本文件中读取并映射到 person 类：

public class sqlTestv2 implements Serializable

{

    private int appID;
    private int dataID;
    private JavaSparkContext sc;



    public JavaRDD<Person> getDataRDD()
    {

    JavaRDD<String> test = sc.textFile("hdfs://localhost:8020/user/cloudera/people.txt");

         JavaRDD<Person> people = test.map(
                    new Function<String, Person>() {
                    public Person call(String line) throws Exception {
                      String[] parts = line.split(",");

                      Person person = new Person();
                      person.setName(parts[0]);
                      person.setAge(Integer.parseInt(parts[1].trim()));

                      return person;
                    }
                  });


        return people;



    }

}

这会检索RDD并对其执行操作：

public class sqlTestv1 implements Serializable

{

    public static void main(String[] arg) throws Exception 
    {
          SparkConf conf = new SparkConf().setMaster("local").setAppName("wordCount");
          JavaSparkContext sc = new JavaSparkContext(conf);
          SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);
          sqlTestv2 v2=new sqlTestv2(1,1,sc);
          JavaRDD<Person> test=v2.getDataRDD();


          DataFrame schemaPeople = sqlContext.createDataFrame(test, Person.class);
          schemaPeople.registerTempTable("people");

          DataFrame df = sqlContext.sql("SELECT age FROM people");

          df.show();





    }

}

完全错误：

线程“主”org.apache.spark.SparkException 中的异常：任务不是 可序列化 在 org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:315) 在 org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:305) 在 org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:132) 在 org.apache.spark.SparkContext.clean(SparkContext.scala:1893) 在 org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:294) 在 org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:293) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:147) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:108) 在 org.apache.spark.rdd.RDD.withScope(RDD.scala:286) 在 org.apache.spark.rdd.RDD.map(RDD.scala:293) 在 org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:90) 在 org.apache.spark.api.java.AbstractJavaRDDLike.map(JavaRDDLike.scala:47) 在 com.oreilly.learningsparkexamples.mini.java.sqlTestv2.getDataRDD(sqlTestv2.java:54) 在 com.oreilly.learningsparkexamples.mini.java.sqlTestv1.main(sqlTestv1.java:41) 引起：java.io.NotSerializableException：org.apache.spark.api.java.JavaSparkContext 序列化栈： - 对象不可序列化（类：org.apache.spark.api.java.JavaSparkContext，值： org.apache.spark.api.java.JavaSparkContext@3c3b144b) - 字段（类：com.oreilly.learningsparkexamples.mini.java.sqlTestv2，名称：sc，类型： 类 org.apache.spark.api.java.JavaSparkContext) - 对象（com.oreilly.learningsparkexamples.mini.java.sqlTestv2 类， com.oreilly.learningsparkexamples.mini.java.sqlTestv2@3752fdda) - 字段（类：com.oreilly.learningsparkexamples.mini.java.sqlTestv2$1，名称：this$0， 类型：com.oreilly.learningsparkexamples.mini.java.sqlTestv2 类） - 对象（com.oreilly.learningsparkexamples.mini.java.sqlTestv2$1 类， com.oreilly.learningsparkexamples.mini.java.sqlTestv2$1@70c171ec) - 字段（类：org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1， 名称：fun$1，类型：接口 org.apache.spark.api.java.function.Function) - 对象（类 org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1， ) 在 org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40) 在 org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47) 在 org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:81)

Answer 1

堆栈告诉你答案。这是您传递给sqlTestv2 的JavaSparkContext。您应该将 sc 传递给方法，而不是类

Answer 2

可以给sc加上“transient”修饰符，使其不被序列化。