Spark：任务不可序列化（广播/RDD/SparkContext）

Question

Spark 中有很多关于Task is not serializable 的问题。但是，这个案例似乎很特殊。

我已经创建了一个类：

class Neighbours(e: RDD[E], m: KMeansModel) extends Serializable {
  val allEs: RDD[(String, E)] = e.map(e => (e.w, e))
    .persist()
  val sc = allEs.sparkContext
  val centroids = sc.broadcast(m.clusterCenters)
  [...]

该类定义了以下方法：

private def centroidDistances(v: Vector): Array[Double] = {
  centroids.value.map(c => (centroids.value.indexOf(c), Vectors.sqdist(v, c)))
    .sortBy(_._1)
    .map(_._2)
}

但是，当调用该类时，会抛出 Task is not serializable 异常。

很奇怪，Neighbours 类的标题中的微小变化就足以解决问题。我没有创建 val sc: SparkContext 用于广播，而是内联创建 Spark 上下文的代码：

class Neighbours(e: RDD[E], m: KMeansModel) extends Serializable {
  val allEs: RDD[(String, E)] = e.map(e => (e.w, e))
  .setName("embeddings")
  .persist()
  val centroids = allEmbeddings.sparkContext(m.clusterCenters)
  [...]

我的问题是：第二个变体有何不同？第一个出了什么问题？直觉上，这应该只是语法糖，这是 Spark 的 bug 吗？

我在 Hadoop/Yarn 集群上使用 Spark 1.4.1。

Answer 1

当你定义时

class Neighbours(e: RDD[E], m: KMeansModel) extends Serializable {
  ...
  val sc = allEmbeddings.sparkContext
  val centroids = sc.broadcast(m.clusterCenters)
  ...
}

您已将sc 设置为类变量，这意味着可以从Neighbours 的实例访问它，例如neighbours.sc。这意味着sc 需要可序列化，事实并非如此。

当你内联代码时，只有centroids的最终值需要可序列化。 centroids 是 Broadcast 类型，它是可序列化的。