使用 pyspark RDD .groupByKey 提取每组的最高值元素答案

【问题标题】：Using pyspark RDD .groupByKey extract highest value element per group使用 pyspark RDD .groupByKey 提取每组的最高值元素
【发布时间】：2020-03-29 03:24:41
【问题描述】：

TOKEN_RE = re.compile(r"\b[\w']+\b")
def pos_tag_counter(line):
    toks = nltk.regexp_tokenize(line.lower(), TOKEN_RE)
    postoks = nltk.tag.pos_tag(toks)
    return postoks

pos_tag_counts = text.filter(lambda line: len(line) > 0) \
    .filter(lambda line: re.findall('^(?!URL).*', line)) \
    .flatMap(pos_tag_counter) \
    .map(lambda word: (word, 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .map(lambda x: (x[0][1], (x[1], x[0][0]))) \
    .groupByKey().map(lambda x : (x[0], list(x[1])))

我有一个文本文件，它被简化为行，而不是单词，单词被计算并用 POS（词性）标签标记。所以我现在拥有的是一系列元组（pos，（word，count））。 POS是关键。我需要为每个 POS 找到最常用的词。

[('NN', (1884, '华盛顿')),
('NN', (5, '恒星')),
('VBD', (563, '保留')),
('DT', (435969, 'the')),
('JJ', (9300, '第一')),
('NN', (1256, '一半')),
('NN', (4028, '季节')),

这是我的第一个 pyspark 项目，所以我认为我不太了解这个概念。我用组

[('VBD',
[（563，'保留'），
(56715, '说'),
(2640, '得到'),
(12370, 's'),
(55523, '是'),
(62, '捕捉'),

理想情况下，只要元组显示每个 POS 的最高计数单词，输出将是 - (POS, count, word)：

('NN', 1884, '华盛顿')
('DT', 435969, 'the')
等等

【问题讨论】：

标签： python pyspark grouping rdd top-n

【解决方案1】：

基本思路是groupByKey，然后求每个组的最大值。由于您需要最长的单词，您可以将max 方法的键定义为单词的长度。

rdd = sc.parallelize([('NN', (1884, 'washington')),
    ('NN', (5, 'stellar')),
    ('VBD', (563, 'kept')),
    ('DT', (435969, 'the')),
    ('JJ', (9300, 'first')),
    ('NN', (1256, 'half')),
    ('NN', (4028, 'season'))])

pos_count = rdd.groupByKey()
               .mapValues(lambda v: max(v, key=lambda x: len(x[1])))

print(pos_count.collect())
# [('DT', (435969, 'the')), ('VBD', (563, 'kept')), ('NN', (1884, 'washington')), ('JJ', (9300, 'first'))]

【讨论】：

我不需要最长的单词，我需要最频繁的单词，所以我需要元组中最大的数字才能在那里发挥作用，但你肯定帮了我很多。所以最后一行的解决方案是 .groupByKey().mapValues(lambda v: max(v, key=lambda x: x[0]))

【解决方案2】：

您不能将映射步骤更改为map(lambda x: (x[0][1], x[1], x[0][0]))，即：

pos_tag_counts = text.filter(lambda line: len(line) > 0) \
    .filter(lambda line: re.findall('^(?!URL).*', line)) \
    .flatMap(pos_tag_counter) \
    .map(lambda word: (word, 1)) \
    .reduceByKey(lambda x, y: x + y) \
    .map(lambda x: (x[0][1], x[1], x[0][0]))

【讨论】：

对不起，我的意思是每个 POS 中出现频率最高的词