java - 针对不可序列化对象和函数的 Spark Scala 编程

Question

当我使用 Spark Scala 程序运行时出现“任务不可序列化”异常

• 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

RDD 必须可序列化，因此您无法创建不可序列化类的 RDD。

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

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

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