SparkException：このRDDにSparkContextがありません

Question

文字列サンプラーを作成するには、rddの文字列を辞書として使用し、クラスRandomDataGeneratorをorg.apache.spark.mllib.randomパッケージから作成しています。

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を生成しようとすると、辞書がSparkContextの欠けていると言ってSparkExceptionが発生しました。クラスタノードにコピーするときに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をシリアライズしてデシリアライズしようとしたときに発生します.RDDはコンテキストを失い、とにかく無意味です。

ただ、この操作を行います。

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