如何使用 pyspark 为非 pairwiseRDD 正确 groupByKey

Posted 2023-04-15

【中文标题】如何使用 pyspark 为非 pairwiseRDD 正确 groupByKey

【英文标题】：How to correctly groupByKey for non pairwiseRDDs using pyspark

【发布时间】：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 我认为这已转变为另一个问题，我是否应该编辑回问题并选择您的帖子作为答案，然后再发布另一个问题？

【参考方案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 v​​irens”是甲虫科的一种 天牛科。')])]

【讨论】：

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