Spark 中比 filter.count 更有效的方法？答案

【问题标题】：more efficient method in Spark than filter.count?Spark 中比 filter.count 更有效的方法？
【发布时间】：2020-06-25 07:14:48
【问题描述】：

我有一个任务，我在 Spark 中有一个 rdd，其记录如下所示：

[(id, group), {'token1', 'token2'...}]

例如 '''tokenizedTweetsByUser.take(5)''' 提供：

[(('470520068', 3),  {'#berniesanders',   '#goldmansachs',   '$',   '.',   '/',   '4',   'a',   'adorned',   'bc',   'capitalist',   'class',   "doesn't",   'he',   "i'm",   'pig',   'ride',   'rigged',   'system',   'voting',   'w',   'war'}), (('2176120173', 6),  {'!',   '#america',   '#trump',   '#votetrump',   '&',   '.',   ':',   ';',   '@realdonaldtrump',   '@trumpnewmedia',   'amp',   'change',   "don't",   'get',   'htt',   'if',   "it's",   'nothing',   'out',   'rt',   'simple',   'that',   'will',   'you',   '…'}), (('145087572', 3),  {'!',   '#colorado',   '#denver',   '%',   ',',   '-',   '.',   '1',   '11am',   '1pm',   ':',   '@allonmedicare',   '@berniesanders',   '@libertea2012',   '@rockportbasset',   'america',   'and',   'capitol',   'co',   'endorse',   'for',   'herself',   'hillary',   'http',   'icymi',   'in',   'is',   'leading',   'liar',   'mst',   'only',   'out',   'positive',   'progressive',   'proof',   'rt',   's',   'state',   'that',   'the',   'to',   'today',   'voices',   'wake-up',   'weasel',   '’',   '…'}), (('23047147', 6),  {'@madworldnews',   '[',   ']',   'after',   'bernie',   'deal',   'fans',   'had',   'liberal',   'pour',   'supporter',   'tears',   'to',   'trump',   'via',   'vid',   'with'}), (('526506000', 4),  {'.',   ':',   '@justinamash',   '@tedcruz',   'calls',   'candidate',   'cartel',   'correctly',   'he',   'i',   'is',   'on',   'only',   'remaining',   'rt',   'take',   'the',   'to',   'trust',   'washington',   'what',   '…'})]

令牌来自推文和前 100 个令牌的列表，我需要计算每个组中每个令牌的数量。有8组。

我的实现很简单：

    tokenizedTweetsByUser.cache()
    groupCounts = []
    for i in range(8):
        groupCounts.append([])
        for token in tokensList:
          #the following statement take too long!
          item_count = tokenizedTweetsByUser.filter(lambda x: (x[0][1] == i) and (token in x[1])).count()
        if item_count > 0:
            groupCounts[i].append((token, item_count))

但这需要很长时间。我知道 filter.count 将运行 800 次，但因为它只是一个过滤器计数，我们正在寻找一组我希望性能相当好的令牌。

有人可以建议另一种性能更高的方法吗？

【问题讨论】：

你能提供一个真实数据的小样本吗
谢谢。在问题中添加了示例数据。
对于输出可能是一个想法，这就是我们在 SO 上的工作方式。谢谢
我跑了它但无法跟随，它似乎只取最后一个令牌，例如tokensList = ['装饰'，'资本主义'，'骗子'，'@berniesanders'，'#votetrump'，'#goldmansachs']
问题需要更多关注。

标签： python-3.x apache-spark filter count

【解决方案1】：

reduceByKey 似乎比在许多 filter.count 实例上循环更有效。

在这种情况下，将组和列表项组合成一个字符串，该字符串将成为键，以便每个列表项都是单独的行。然后执行reduceByKey：

tokenizedTweetsByUser.flatMapValues(lambda x: x).map(lambda x: (str(x[0][1])+" "+str(x[1]), 1)).reduceByKey(lambda x, y: x + y)

速度快几倍。

【讨论】：