более эффективный метод в Spark, чем filter.count?

Question

у меня есть задание, в котором у меня есть rdd в Spark с записью, которая выглядит следующим образом:

[(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 раз, но поскольку это всего лишь счетчик фильтров, и мы ищем токен в наборе, который, как я ожидал, будет достаточно эффективным.

может кто-нибудь предложить другой метод что было бы более производительно?

Answer 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)

раз быстрее.