如何使生成器在 spark mapPartitions() 中工作？

Question

我正在尝试在 spark 中使用 mapPartiton 来处理大型文本语料库： 假设我们有一些看起来像这样的半处理数据：

    text_1 = [['A', 'B', 'C', 'D', 'E'],
    ['F', 'E', 'G', 'A', 'B'],
    ['D', 'E', 'H', 'A', 'B'],
    ['A', 'B', 'C', 'F', 'E'],
    ['A', 'B', 'C', 'J', 'E'],
    ['E', 'H', 'A', 'B', 'C'],
    ['E', 'G', 'A', 'B', 'C'],
    ['C', 'F', 'E', 'G', 'A'],
    ['C', 'D', 'E', 'H', 'A'],
    ['C', 'J', 'E', 'H', 'A'],
    ['H', 'A', 'B', 'C', 'F'],
    ['H', 'A', 'B', 'C', 'J'],
    ['B', 'C', 'F', 'E', 'G'],
    ['B', 'C', 'D', 'E', 'H'],
    ['B', 'C', 'F', 'E', 'K'],
    ['B', 'C', 'J', 'E', 'H'],
    ['G', 'A', 'B', 'C', 'F'],
    ['J', 'E', 'H', 'A', 'B']]

每个字母都是一个单词。我也有词汇：

    V = ['D','F','G','C','J','K']
    text_1RDD = sc.parallelize(text_1)

我想在 spark 中运行以下命令：

    filtered_lists = text_1RDD.mapPartitions(partitions)

    filtered_lists.collect()

我有这个功能：

    def partitions(list_of_lists,vc):

            for w in vc:

                iterator = []
                for sub_list in list_of_lists:

                    if w in sub_list:
                        iterator.append(sub_list)

        yield (w,len(iterator))

如果我这样运行：

    c = partitions(text_1,V)
    for item in c:
        print(item)

它返回正确的计数

    ('D', 4)
    ('F', 7)
    ('G', 5)
    ('C', 15)
    ('J', 5)
    ('K', 1)

但是，我不知道如何在 spark 中运行它：

    filtered_lists = text_1RDD.mapPartitions(partitions)

    filtered_lists.collect()

它只有一个参数，在 Spark 中运行时会产生很多错误...

但即使我在分区函数中编码词汇：

    def partitionsV(list_of_lists):
            vc = ['D','F','G','C','J','K']
            for w in vc:

                iterator = []
                for sub_list in list_of_lists:

                    if w in sub_list:
                        iterator.append(sub_list)

        yield (w,len(iterator))

..我明白了：

    filtered_lists = text_1RDD.mapPartitions(partitionsV)

    filtered_lists.collect()

输出：

     [('D', 2),
     ('F', 0),
     ('G', 0),
     ('C', 0),
     ('J', 0),
     ('K', 0),
     ('D', 0),
     ('F', 0),
     ('G', 0),
     ('C', 0),
     ('J', 0),
     ('K', 0),
     ('D', 1),
     ('F', 0),
     ('G', 0),
     ('C', 0),
     ('J', 0),
     ('K', 0),
     ('D', 1),
     ('F', 0),
     ('G', 0),
     ('C', 0),
     ('J', 0),
     ('K', 0)]

显然，生成器没有按预期工作。我完全被困住了。 我对火花很陌生。如果有人能向我解释这里发生了什么，我将不胜感激......

Answer 1

这是另一个字数问题，mapPartitions 不是这项工作的工具：

from operator import add

v = set(['D','F','G','C','J','K'])

result = text_1RDD.flatMap(v.intersection).map(lambda x: (x, 1)).reduceByKey(add)

结果是

for x in result.sortByKey().collect(): 
    print(x) 

('C', 15)
('D', 4)
('F', 7)
('G', 5)
('J', 5)
('K', 1)