SparkException：此 RDD 缺少 SparkContext

Question

我正在尝试使用 rdd 的字符串作为字典和来自 org.apache.spark.mllib.random 包的类 RandomDataGenerator 创建一个字符串采样器。

import org.apache.spark.mllib.random.RandomDataGenerator
import org.apache.spark.rdd.RDD
import scala.util.Random

class StringSampler(var dic: RDD[String], var seed: Long = System.nanoTime) extends RandomDataGenerator[String] {
    require(dic != null, "Dictionary cannot be null")
    require(!dic.isEmpty, "Dictionary must contains lines (words)")
    Random.setSeed(seed)

    var fraction: Long = 1 / dic.count()

    //return a random line from dictionary
    override def nextValue(): String = dic.sample(withReplacement = true, fraction).take(1)(0)

    override def setSeed(newSeed: Long): Unit = Random.setSeed(newSeed)

    override def copy(): StringSampler = new StringSampler(dic)

    def setDictionary(newDic: RDD[String]): Unit = {
        require(newDic != null, "Dictionary cannot be null")
        require(!newDic.isEmpty, "Dictionary must contains lines (words)")
        dic = newDic
        fraction = 1 / dic.count()
    }
}

---

val dictionaryName: String
val dictionaries: Broadcast[Map[String, RDD[String]]]
val dictionary: RDD[String] = dictionaries.value(dictionaryName) // dictionary.cache()
val sampler = new StringSampler(dictionary)
RandomRDDs.randomRDD(context, sampler, size, numPartitions)

但是当我尝试生成字符串的随机 RDD 时，我遇到了一个 SparkException，说字典缺少 SparkContext。似乎 spark 在将字典 rdd 复制到集群节点时丢失了上下文，我不知道如何修复它。

我尝试在将字典传递给 StringSampler 之前缓存它，但它并没有改变任何东西...... 我正在考虑将它链接回原来的 SparkContext，但我什至不知道这是否可能。有人有想法吗？

Caused by: org.apache.spark.SparkException: This RDD lacks a SparkContext. It could happen in the following cases: 
(1) RDD transformations and actions are NOT invoked by the driver, but inside of other transformations; for example, rdd1.map(x => rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation. For more information, see SPARK-5063.
(2) When a Spark Streaming job recovers from checkpoint, this exception will be hit if a reference to an RDD not defined by the streaming job is used in DStream operations. For more information, See SPARK-13758.

Answer 1

我相信问题出在这里：

val dictionaries: Broadcast[Map[String, RDD[String]]]
val dictionary: RDD[String] = dictionaries.value(dictionaryName)

您不应该广播任何包含 RDD 的内容。 RDD 已经并行化并分布在整个集群中。该错误来自尝试序列化和反序列化 RDD，这会丢失其上下文并且无论如何都没有意义。

这样做：

val dictionaries: Map[String, RDD[String]]
val dictionary: RDD[String] = dictionaries(dictionaryName)