使用Spark的奇怪“任务不可序列化”

Question

在我的程序中，我有一个返回一些RDD的方法，我们称它为myMethod，它接受一个不可序列化的参数，并让RDD的类型为Long（我真正的RDD是元组类型，但只包含基元类型）。

当我尝试这样的事情时：

val x: NonSerializableThing = ...
val l: Long = ...
myMethod(x, l).map(res => res + l) // myMethod's RDD does NOT include the NonSerializableThing

我得到的任务不可序列化。

当我用res 1L替换res l（即某个常数）时，它会运行。

从序列化跟踪中，它试图序列化非序列化对象，并在那里阻塞，但我仔细检查了我的方法，这个对象从未出现在RDD中。

当我试图直接收集mymethod的输出时，即

myMethod(x, l).take(1) foreach println

我也没有问题。

该方法使用NonSerializableThing获取（本地）值序列，对其进行多个Cassandra查询（这是必需的，因为我需要构造要查询的分区键），如下所示：

def myMethod(x: NonSerializableThing, l: Long): RDD[Long] = {
  val someParam1: String = x.someProperty
  x.getSomeSeq.flatMap(y: OtherNonSerializableThing => {
    val someParam2: String = y.someOtherProperty
    y.someOtherSeq.map(someParam3: String =>
      sc.cassandraTable("fooKeyspace", "fooTable").
      select("foo").
      where("bar=? and quux=? and baz=? and l=?", someParam1, someParam2, someParam3, l).
      map(_.getLong(0))
  }.reduce((a, b) => a.union(b))
}

getSomeSeq和someOtherSeq返回纯非火花Seqs

我想要实现的是“联合”多个Cassandra查询。

这里有什么问题？

根据杰姆·塔克的要求进行编辑和补遗：

我班上有这样的东西：

implicit class MySparkExtension(sc: SparkContext) {

  def getThing(/* some parameters */): NonSerializableThing = { ... }

  def myMethod(x: NonSerializableThing, l: Long): RDD[Long] = {
    val someParam1: String = x.someProperty
    x.getSomeSeq.flatMap(y: OtherNonSerializableThing => {
      val someParam2: String = y.someOtherProperty
      y.someOtherSeq.map(someParam3: String =>
        sc.cassandraTable("fooKeyspace", "fooTable").
        select("foo").
        where("bar=? and quux=? and baz=? and l=?", someParam1, someParam2, someParam3, l).
        map(_.getLong(0))
    }.reduce((a, b) => a.union(b))
  }
}

这是在包对象中声明的。问题出现在这里：

// SparkContext is already declared as sc
import my.pkg.with.extension._

val thing = sc.getThing(/* parameters */)
val l = 42L
val rdd = sc.myMethod(thing, l)
// until now, everything is OK.
// The following still works:
rdd.take(5) foreach println
// The following causes the exception:
rdd.map(x => x >= l).take(5) foreach println
// While the following works:
rdd.map(x => x >= 42L).take(5) foreach println

我在Spark shell以及通过Spark submit提交的算法中测试了输入“live”的数据。

我现在想尝试（根据我最后的评论）如下：

implicit class MySparkExtension(sc: SparkContext) {

  def getThing(/* some parameters */): NonSerializableThing = { ... }

  def myMethod(x: NonSerializableThing, l: Long): RDD[Long] = {
    val param1 = x.someProperty
    val partitionKeys =
      x.getSomeSeq.flatMap(y => {
        val param2 = y.someOtherProperty
        y.someOtherSeq.map(param3 => (param1, param2, param3, l)
      }
    queryTheDatabase(partitionKeys)
  }

  private def queryTheDatabase(partitionKeys: Seq[(String, String, String, Long)]): RDD[Long] = {
    partitionKeys.map(k =>
      sc.cassandraTable("fooKeyspace", "fooTable").
         select("foo").
         where("bar=? and quux=? and baz=? and l=?", k._1, k._2, k._3, k._4).
         map(_.getLong(0))
    ).reduce((a, b) => a.union(b))
  }
}

我相信这可以工作，因为RDD是在方法queryTheDatabase中构建的，其中不存在nonSerializableThing。

另一个选项可能是：NonSerializableThing确实是可序列化的，但我将SparkContext作为隐式构造函数参数传入其中。我想如果我让这个暂时的，它会（无用地）被序列化，但不会引起任何问题。

Answer 1

当用1L替换l时，Spark不再尝试用中的方法/变量序列化类，因此不会引发错误。

您应该能够通过标记val x:NonSerializableThing= 作为瞬态，例如。

@transient
val x: NonSerializableThing = ...

这意味着当类被序列化时，这个变量应该被忽略。