火花Scala编程为不可序列化的对象和功能

Question

当我运行Spark Scala程序时，有一个“Task not serializable”异常

• Spark RDD是不可串行化类型（java类）
• 调用的函数来自不可序列化的类（java类，再次）

我的代码是这样的

object Main{
    def main(args : Array(String){
        ...
        var rdd = sc.textFile(filename)
                  .map(line => new NotSerializableJClass(line)).cache() 
        //rdd is RDD[NotSerializableJClass]
        ...
        var test = new NotSerializableJPredicate()
        rdd = rdd.filter(elem => test.test(elem))
        //throws TaskNotSerializable on test Predicate class
    }
}

我注意到我可以用

rdd = rdd.filter(elem => (new NotSerializableJPredicate()).test(elem))

但对于RDD中的对象类，我仍然会遇到这个例外。我会以另一种方式，也会以另一种方式，也就是第二部分，因为我不想创建大量PredicateClass的对象。

你能帮我吗？我如何处理不可序列化的类？

Answer 1

帮助我避免任务序列化问题的一些一般规则：

如果您从代码中调用任何类的方法；Spark将需要序列化包含该方法的整个类。绕过的方法可以是以下任何一种

Answer 2

RDD必须是可序列化的，因此不能创建不可序列化类的RDD。

对于谓词，可以使用mapPartitions编写。

rdd.mapPartitions{
  part => 
    val test = new NotSerializableJPredicate()
    part.filter{elem => test.test(elem)}
   }

mapPartitons将在每个分区上运行一次，因此它允许您在executor上实例化不可序列化的类，但它只需要在每个分区上运行一次，而不是针对每个记录。