scala - 将 Scalaz 与 Spark 结合使用时出现不可序列化异常

Question

我做了一个简单的例子来尝试整合scalaz library 使用 Apache Spark 1.5 编写代码。

这是一个简单的 Spark 程序来说明我的问题:

package test

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

import ca.crim.deti.re.spark.sparkConf
import scalaz._
import scalaz.Scalaz._

object TestSpark {

  def main(args: Array[String]) = {
    val conf = new SparkConf().setAppName("Test").setMaster("local")
    val SC = new SparkContext(conf)
    val c = SC.parallelize(List(1, 2, 3, 4, 5))

    println(func1(c).count) // WORKS
    println(func2(c).count) // DOES NOT WORK.. NotSerializableException
  }

  // WORKS!
  def func1(rdd: RDD[Int]) = {
    rdd.filter { i => f(i, i) }
  }
  // DOES NOT WORK!
  def func2[I: Equal](rdd: RDD[I]) = {
    rdd.filter { i => f(i, i) }
  }

  def f[I: Equal](i1: I, i2: I) = {
    i1 === i2
  }
}

我想通过在函数定义中使用 Equal 来使 func2 工作。

当使用 func2 在本地模式下在 Spark 环境中执行时，出现以下异常:

Answer 1

由于您的函数具有 Equal[I] 约束，因此 Spark 正在关闭它并尝试在进行分发时对其进行序列化。由于 scalaz.Equal 类型类不是Serializable ( https://github.com/scalaz/scalaz/blob/v7.2.0/core/src/main/scala/scalaz/Equal.scala#L10 ) Spark 在运行时失败。

您可以通过使用 Twitter 的 chill 库中的 MeatLocker 来绕过它不是 Serialziable 的事实:https://github.com/twitter/chill#the-meatlocker

或者，cats和 algebra库(有一个类似于您上面使用的 Equal 类型类)有可序列化的类型类，您应该能够像上面那样毫无问题地使用它们。