减少单词列表，将元组计数到聚合键

Posted 2023-04-17

【中文标题】减少单词列表，将元组计数到聚合键

【英文标题】：Reduce list of word, count tuples up to aggregate key

【发布时间】：2017-09-29 16:27:13

【问题描述】：

我正在尝试使用 Spark 字数统计示例并按其他值汇总字数（例如，在下面的情况下，人是“VI”或“MO”的人的字数和计数）

我有一个 rdd，它是一个元组列表，其值为元组列表：

from operator import add
reduced_tokens = tokenized.reduceByKey(add)
reduced_tokens.take(2)

这给了我：

[(u'VI', [(u'word1', 1), (u'word2', 1), (u'word3', 1)]),
 (u'MO',
  [(u'word4', 1),
   (u'word4', 1),
   (u'word5', 1),
   (u'word8', 1),
   (u'word10', 1),
   (u'word1', 1),
   (u'word4', 1),
   (u'word6', 1),
   (u'word9', 1),
   ...
 )]

我想要类似的东西：

[
 ('VI', 
    [(u'word1', 1), (u'word2', 1), (u'word3', 1)],
 ('MO', 
    [(u'word4', 58), (u'word8', 2), (u'word9', 23) ...)
]

类似于word count example here，我希望能够过滤掉某个人的计数低于某个阈值的单词。谢谢！

【参考方案1】：

您尝试减少的键是 (name, word) 对，而不仅仅是名称。因此，您需要执行.map 步骤来修复您的数据：

def key_by_name_word(record):
  name, (word, count) = record
  return (name, word), count

tokenized_by_name_word = tokenized.map(key_by_name_word)
counts_by_name_word = tokenized_by_name_word.reduce(add)

这应该给你

[
  (('VI', 'word1'), 1),
  (('VI', 'word2'), 1),
  (('VI', 'word3'), 1),
  (('MO', 'word4'), 58),
  ...
]

要将其转换为您提到的完全相同的格式，您可以这样做：

def key_by_name(record):
  # this is the inverse of key_by_name_word
  (name, word), count = record
  return name, (word, count)

output = counts_by_name_word.map(key_by_name).reduceByKey(add)

但实际上使用counts_by_name_word 所在的平面格式的数据可能更容易。

【讨论】：

我的数据结构略有不同，但这有助于我了解如何修复它。我的初始数据看起来像[Row(key=u'VI', item=u'word1 word2 word3'), ...]，我创建了一个函数来标记项目并返回[((name, token), 1) for token in tokens]。从那里我使用 flatMap 将函数应用于我的数据以获得您建议的结构。

【参考方案2】：

为了完整起见，以下是我解决问题各个部分的方法：

问题 1：按某个键汇总字数

import re

def restructure_data(name_and_freetext):
    name = name_and_freetext[0]
    tokens = re.sub('[&|/|\d4|\.|\,|\:|\-|\(|\)|\+|\$|\!]', ' ', name_and_freetext[1]).split()
    return [((name, token), 1) for token in tokens]

filtered_data = data.filter((data.flag==1)).select('name', 'item')
tokenized = filtered_data.rdd.flatMap(restructure_data)

问题 2：过滤掉计数低于某个阈值的单词：

from operator import add

# keep words which have counts >= 5
counts_by_state_word = tokenized.reduceByKey(add).filter(lambda x: x[1] >= 5)

# map filtered word counts into a list by key so we can sort them
restruct = counts_by_name_word.map(lambda x: (x[0][0], [(x[0][1], x[1])]))

奖励：将单词从最频繁到最不频繁排序

# sort the word counts from most frequent to least frequent words
output = restruct.reduceByKey(add).map(lambda x: (x[0], sorted(x[1], key=lambda y: y[1], reverse=True))).collect()