PySpark实现CCF-iterate算法时RDD.collect()执行失败求助

问题场景 #

尝试实现带二次排序的CCF-iterate算法，代码在Jupyter环境中执行到rdd.collect()时抛出TypeError: 'list' object is not an iterator错误。

原代码如下：

rdd = sc.parallelize([(1, 2), (2, 3), (2, 4), (4, 5), (6, 7), (7, 8)])

num_iterations = 5

def map_func(pair):
    key, value = pair
    yield (key, value)
    yield (value, key)

def reduce_func(key, values):
    min_value = next(values)
    if min_value < key:
        yield (key, min_value)
        for value in values:
            yield (value, min_value)

for i in range(num_iterations):
    mapped_rdd = rdd.flatMap(map_func)

    updated_rdd = mapped_rdd.groupByKey().flatMap(lambda x: reduce_func(x[0], sorted(x[1])))

    dedup_rdd = updated_rdd.distinct()

    mapped_rdd_2 = dedup_rdd.flatMap(lambda x: [(x[0], x[1]), (x[1], x[0])])

    updated_rdd_2 = mapped_rdd_2.groupByKey().flatMap(lambda x: reduce_func(x[0], sorted(x[1])))

    rdd = updated_rdd_2.distinct()

print(rdd.collect())

错误原因 #

报错核心是：reduce_func中使用next(values)尝试获取迭代器的下一个元素，但传入的values是列表（sorted(x[1])返回列表），而列表不支持next()方法，只有迭代器对象才具备该方法。

修复方案 #

有两种可行的修复方式：

方式1：在reduce_func内部将输入转换为迭代器 #

修改reduce_func，先把传入的可迭代对象转为迭代器：

def reduce_func(key, values):
    # 将可迭代对象转换为迭代器
    values_iter = iter(values)
    min_value = next(values_iter)
    if min_value < key:
        yield (key, min_value)
        for value in values_iter:
            yield (value, min_value)

方式2：调用reduce_func时传入迭代器 #

在调用reduce_func的地方，将sorted(x[1])转为迭代器：

updated_rdd = mapped_rdd.groupByKey().flatMap(lambda x: reduce_func(x[0], iter(sorted(x[1]))))
# 同时修改另一处调用
updated_rdd_2 = mapped_rdd_2.groupByKey().flatMap(lambda x: reduce_func(x[0], iter(sorted(x[1]))))

修复后完整代码 #

这里采用方式1的修复，完整代码如下：

rdd = sc.parallelize([(1, 2), (2, 3), (2, 4), (4, 5), (6, 7), (7, 8)])

num_iterations = 5

def map_func(pair):
    key, value = pair
    yield (key, value)
    yield (value, key)

def reduce_func(key, values):
    values_iter = iter(values)
    min_value = next(values_iter)
    if min_value < key:
        yield (key, min_value)
        for value in values_iter:
            yield (value, min_value)

for i in range(num_iterations):
    mapped_rdd = rdd.flatMap(map_func)

    updated_rdd = mapped_rdd.groupByKey().flatMap(lambda x: reduce_func(x[0], sorted(x[1])))

    dedup_rdd = updated_rdd.distinct()

    mapped_rdd_2 = dedup_rdd.flatMap(lambda x: [(x[0], x[1]), (x[1], x[0])])

    updated_rdd_2 = mapped_rdd_2.groupByKey().flatMap(lambda x: reduce_func(x[0], sorted(x[1])))

    rdd = updated_rdd_2.distinct()

print(rdd.collect())

内容的提问来源于stack exchange，提问作者Mina Hrb