PySpark 中 mapPartitions 返回的空数组

Question

你好，有人可以向我解释为什么mapPartitions 对这两个函数的反应不同吗？ （我看过这个this thread，但我认为我的问题不在于我创建它时我的可迭代对象是TraversableOnce。

L=range(10)
J=range(5,15)
K=range(8,18)

data=J+K+L

def function_1(iter_listoflist):
    final_iterator=[]
    for sublist in iter_listoflist:
        final_iterator.append([x for x in sublist if x%9!=0])
    return iter(final_iterator)  

def function_2(iter_listoflist):
    final_iterator=[]
    listoflist=list(iter_listoflist)
    for i in range(len(listoflist)):
        for j in range(i+1,len(listoflist)):
            sublist=listoflist[i]+listoflist[j]
            final_iterator.append([x for x in sublist if x%9!=0])
            pass
        pass
    return iter(final_iterator)



sc.parallelize(data,3).glom().mapPartitions(function_1).collect()

返回它应该同时

sc.parallelize(data,3).glom().mapPartitions(function_2).collect()

返回一个空数组，我通过在末尾返回一个列表来检查代码，它会执行我想要的操作。

感谢您的帮助

菲利普·C

Answer 1

其实很简单。 listoflist 的长度始终等于 1。要理解为什么会这样，您必须考虑调用 glom 时发生了什么。引用 the docs 它返回：

通过合并每个分区中的所有元素创建的 RDD 进入列表。

表示当你调用时：

listoflist=list(iter_listoflist)

您将获得一个包含该分区中所有元素的单个元素列表的列表。忽略所有细节：

(sc.parallelize(data, 3)
    .glom()
    .mapPartitionsWithIndex(lambda i, iter: [(i, list(iter))])
    .collect())

## [(0, [[5, 6, 7, 8, 9, 10, 11, 12, 13, 14]]),
##     (1, [[8, 9, 10, 11, 12, 13, 14, 15, 16, 17]]),
##     (2, [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]])]

意思是

• range(len(listoflist)) 包含一个等于 0 的元素。
• range(i+1,len(listoflist)) 被替换为空 range(1, 1)

因此没有什么可做的，你得到一个空的迭代器。

附带说明，所有这些pass 语句以及iter 调用都已完全过时。