PySpark 的 reduceByKey 没有按预期工作答案

【问题标题】：PySpark's reduceByKey not working as expectedPySpark 的 reduceByKey 没有按预期工作
【发布时间】：2016-01-08 15:57:13
【问题描述】：

我正在编写一个大型 PySpark 程序，最近在 RDD 上使用 reduceByKey 时遇到了麻烦。我已经能够用一个简单的测试程序重新创建问题。代码是：

from pyspark import SparkConf, SparkContext

APP_NAME = 'Test App'

def main(sc):
    test = [(0, [i]) for i in xrange(100)]
    test = sc.parallelize(test)
    test = test.reduceByKey(method)
    print test.collect()

def method(x, y):
    x.append(y[0])
    return x

if __name__ == '__main__':
    # Configure Spark
    conf = SparkConf().setAppName(APP_NAME)
    conf = conf.setMaster('local[*]')
    sc = SparkContext(conf=conf)

    main(sc)

根据 Spark 文档，我希望输出为 (0, [0,1,2,3,4,...,98,99])。相反，我得到以下输出：

[(0, [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 24, 36, 48, 60, 72, 84])]

有人可以帮我理解为什么会生成这个输出吗？

附带说明，当我使用时

def method(x, y):
    x = x + y
    return x

我得到了预期的输出。

【问题讨论】：

作为一个小提示，如果您决定坚持使用reduceByKey（相对于groupByKey），则有一个名为reduce(_func_) 的操作用作减少然后收集。跨度>

标签： python apache-spark pyspark rdd reduce

【解决方案1】：

首先看起来你实际上想要groupByKey而不是reduceByKey：

rdd = sc.parallelize([(0, i) for i in xrange(100)])
grouped = rdd.groupByKey()
k, vs = grouped.first()
assert len(list(vs)) == 100

有人可以帮我理解为什么会生成这个输出吗？

reduceByKey assumes f 是 associative 而你的 method 显然不是。根据操作的顺序，输出是不同的。假设您从某个键的以下数据开始：

[1], [2], [3], [4]

现在添加让我们添加一些括号：

((([1], [2]), [3]), [4])
(([1, 2], [3]), [4])
([1, 2, 3], [4])
[1, 2, 3, 4]

还有另一组括号

(([1], ([2], [3])), [4])
(([1], [2, 3]), [4])
([1, 2], [4])
[1, 2, 4]

当你改写如下：

method = lambda x, y: x + y

或者干脆

from operator import add
method = add

你得到一个关联函数，它按预期工作。

一般而言，对于 reduce* 操作，您需要既是关联函数又是 commutative 的函数。

【讨论】：