如何使用 pyspark 为非 pairwiseRDD 正确 groupByKey答案

【问题标题】：How to correctly groupByKey for non pairwiseRDDs using pyspark如何使用 pyspark 为非 pairwiseRDD 正确 groupByKey
【发布时间】：2018-05-19 14:17:18
【问题描述】：

我是 Python 新手。我也是 pysaprk 的新手。我正在尝试运行一个代码，该代码采用一个看起来像 (id , (span, mention)) 的元组来执行 .map(lambda (id, (span, text)): (id, text))。

我正在使用的代码是：

 m = text\
            .map(lambda (id, (span, text)): (id, text))\
            .mapValues(lambda v: ngrams(v, self.max_ngram))\'''error triggered here'''
            .flatMap(lambda (target, tokens): (((target, t), 1) for t in tokens))\

这是原始数据的格式(id, source, span, text)：

 {'_id': u'en.wikipedia.org/wiki/Cerambycidae',
  'source': 'en.wikipedia.org/wiki/Plinthocoelium_virens',
  'span': (61, 73),
  'text': u'"Plinthocoelium virens" is a species of beetle in the family Cerambycidae.'},
 {'_id': u'en.wikipedia.org/wiki/Dru_Drury',
  'source': 'en.wikipedia.org/wiki/Plinthocoelium_virens',
  'span': (20, 29),
  'text': u'It was described by Dru Drury in 1770.'}]

我收到此错误：

 for k, v in iterator:
TypeError: tuple indices must be integers, not str

我知道 groupByKey 在 pairwiseRDD 上工作，所以我想知道如何正确执行 groupByKey 来解决这个问题？

我们将不胜感激任何帮助或指导。

我正在使用 python 2.7 和 pyspark 2.3.0。

提前谢谢你。

【问题讨论】：

我不明白你想做什么。为什么需要 groupByKey？您发布的代码有什么问题？
@user3689574 这段代码是计算tfidf的第一部分。当 ngram 行调用它时会导致上述错误。请让我知道我是否应该提供有关代码或错误的更多信息。
“提及”是否有您在上面显示为行的字典？
@user3689574 你说的行是什么意思？它在 pyspark 上运行，以便打印我所做的 dict mentions.take(2)
@user3689574 我认为这已转变为另一个问题，我是否应该编辑回问题并选择您的帖子作为答案，然后再发布另一个问题？

标签： python python-2.7 hadoop pyspark rdd

【解决方案1】：

首先你需要将数据映射成一个有键和值的表单，然后是groupByKey。

键值形式总是一个元组（a，b），键是a，值是b。 a 和 b 本身可能是元组。

rdd = sc.parallelize([{'_id': u'en.wikipedia.org/wiki/Cerambycidae',
  'source': 'en.wikipedia.org/wiki/Plinthocoelium_virens',
  'span': (61, 73),
  'text': u'"Plinthocoelium virens" is a species of beetle in the family Cerambycidae.'},
 {'_id': u'en.wikipedia.org/wiki/Dru_Drury',
  'source': 'en.wikipedia.org/wiki/Plinthocoelium_virens',
  'span': (20, 29),
  'text': u'It was described by Dru Drury in 1770.'},
 {'_id': u'en.wikipedia.org/wiki/Dru_Drury',
  'source': 'en.wikipedia.org/wiki/Plinthocoelium_virens2',
  'span': (20, 29, 2),
  'text': u'It was described by Dru Drury in 1770.2'}])

print rdd.map(lambda x: (x["_id"], (x["span"], x["text"]))).groupByKey()\
.map(lambda x: (x[0], list(x[1]))).collect()

[(u'en.wikipedia.org/wiki/Dru_Drury', [((20, 29), u'被描述由 Dru Drury 在 1770 年。'), ((20, 29, 2), u'它是由 Dru Drury 描述的在 1770.2')]), (u'en.wikipedia.org/wiki/Cerambycidae', [((61, 73), u'“Plinthocoelium virens”是甲虫科的一种天牛科。')])]

【讨论】：

谢谢，但它在执行 ngram 的下一行中导致了一个新错误（我编辑了问题以显示代码和错误），我不知道下一行期望什么格式。非常感谢。