Spark Scala动态创建可序列化对象

Question

我需要为 Scala Spark 过滤器使用测试器，测试器实现 java 的 Predicate 接口并通过参数接收特定的类名。 我正在做这样的事情

val tester = Class.forName(qualifiedName).newInstance().asInstanceOf[Predicate[T]]
var filtered = rdd.filter(elem => tester.test(elem))

问题是在运行时我有一个 Spark“TaskNotSerializable Exception”，因为我的特定 Predicate 类不是 Serializable。

如果我这样做了

val tester = Class.forName(qualifiedName).newInstance()
             .asInstanceOf[Predicate[T] with Serializable]
var filtered = rdd.filter(elem => tester.test(elem))

我得到同样的错误。 如果我在 rdd.filter 中创建测试器调用它可以工作：

var filtered = rdd.filter { elem => 
    val tester = Class.forName(qualifiedName).newInstance()
             .asInstanceOf[Predicate[T] with Serializable]
    tester.test(elem)
}

但我会创建一个对象（也许是广播）进行测试。我该如何解决？

Answer 1

您只需要要求类实现Serializable。请注意，asInstanceOf[Predicate[T] with Serializable] 转换是一个谎言：它实际上并没有检查值是Serializable，这就是为什么第二种情况在转换期间不会立即产生错误，而最后一种情况“成功”。

但我会创建一个对象（也许要广播）用于测试。

你不能。广播与否，反序列化将在工作节点上创建 new 对象。但是您只能在每个分区上创建一个实例：

var filtered = rdd.mapPartitions { iter => 
    val tester = Class.forName(qualifiedName).newInstance()
             .asInstanceOf[Predicate[T]]
    iter.filter(tester.test)
}

它实际上比序列化tester、发送它和反序列化它的性能更好，因为它的工作量非常少。