のpython - 私は2 RDD持っている他のRDD

Question

のマップ内のRDDから検索：

• RDD1 =（ワード、スコア）#word：文字列|スコア：int
• RDD2 =（id、text）#id：int |テキスト：その言葉

のリストRDD2内の各「ID」のために、私はそれが

def predecir(texto): 
    contador = 0  
    prediccion = 0 
    for palabra in texto: 
     puntaje = listaRDD.lookup(palabra) 
     if puntaje: 
       puntaje = puntaje[0] 
       prediccion += puntaje 
       contador += 1 
    return (float(prediccion)/ contador) 

listaTestRDD = listaTestRDD.map(lambda x: (x[0], predecir(x[1]))) 
print listaTestRDD.take(1) 

を獲得している場合は、テキスト内の各単語のスコアの平均値を計算したいと私は、このエラーメッセージが表示されます

例外：アクションまたはトランスフォーメーションからRDDまたは を参照してRDDをブロードキャストしようとしているようです。 RDD変換 とアクションは、他の 変換の内部ではなく、ドライバによってのみ呼び出すことができます。たとえば、rdd1.map変換内で値変換とカウントアクション を実行できないため、rdd1.map（lambda x：rdd2.values.count（）* x）は無効です。 の詳細については、SPARK-5063を参照してください。

どうすれば解決できますか？私はもう一方の中に2つのRDDを使用することはできませんか？ O（1）で単語を見つけるためにRDD1を辞書に変換するにはどうすればよいですか？

Answer 1

試してみてください。

RDD2.flatMapValues(lambda x: x) \ 
    .map(lambda x: (x[1], x[0])) \ 
    .leftOuterJoin(RDD1) \ 
    .values() \ 
    .map(lambda x: (x[0], (x[1], 1) if x[1] is not None else (0, 0))) \ 
    .reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1])) \ 
    .mapValues(lambda x: x[0]/float(x[1]) if x[1] else 0.0)