python - 从另一个RDD的地图中的RDD查找答案

【问题标题】：python - Lookup from a RDD inside a map of another RDDpython - 从另一个RDD的地图中的RDD查找
【发布时间】：2016-11-02 19:03:11
【问题描述】：

我有 2 个 RDD：

RDD1 =（单词，分数）#word：字符串 |分数：int
RDD2 = (id, text) #id: int | text：单词列表

所以对于 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(lambda x: rdd2.values.count() * x) 无效，因为值转换和计数操作不能在 rdd1.map 转换内执行。更多信息，请参阅 SPARK-5063。

我该如何解决？我不能在一个内部使用两个 RDD 吗？如何将 RDD1 转换为字典以便在 O(1) 中查找单词？

【问题讨论】：

那么问题是什么？错误信息很清楚。
@LostInOverflow 我该怎么做才能解决它

标签： python apache-spark

【解决方案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)

【讨论】：