用Python计算Spark中成对（K，V）RDD中每个键的平均值

Question

我想与Python共享这个特定的Apache Spark解决方案，因为它的文档非常糟糕。

我想通过KEY计算K/V对（存储在成对RDD中）的平均值。以下是示例数据的外观：

>>> rdd1.take(10) # Show a small sample.
[(u'2013-10-09', 7.60117302052786),
(u'2013-10-10', 9.322709163346612),
(u'2013-10-10', 28.264462809917358),
(u'2013-10-07', 9.664429530201343),
(u'2013-10-07', 12.461538461538463),
(u'2013-10-09', 20.76923076923077),
(u'2013-10-08', 11.842105263157894),
(u'2013-10-13', 32.32514177693762),
(u'2013-10-13', 26.249999999999996),
(u'2013-10-13', 10.693069306930692)]

现在，下面的代码序列是一种不太理想的方法，但它确实有效。在我找到更好的解决方案之前，我就是这么做的。这并不可怕，但是——正如你将在回答部分看到的那样——有一种更简洁、更有效的方法。

>>> import operator
>>> countsByKey = sc.broadcast(rdd1.countByKey()) # SAMPLE OUTPUT of countsByKey.value: {u'2013-09-09': 215, u'2013-09-08': 69, ... snip ...}
>>> rdd1 = rdd1.reduceByKey(operator.add) # Calculate the numerators (i.e. the SUMs).
>>> rdd1 = rdd1.map(lambda x: (x[0], x[1]/countsByKey.value[x[0]])) # Divide each SUM by it's denominator (i.e. COUNT)
>>> print(rdd1.collect())
  [(u'2013-10-09', 11.235365503035176),
   (u'2013-10-07', 23.39500642456595),
   ... snip ...
  ]

Answer 1

现在，更好的方法是使用rdd.aggregateByKey()方法。因为这个方法在Apache Spark和Python文档中的文档记录太少了--这也是我编写这个Q&A的原因--直到最近我一直在使用上面的代码序列。但同样，它的效率较低，因此除非有必要，否则请避免这样做。

下面介绍如何使用rdd.aggregateByKey()方法（推荐的）执行相同的操作。。。

按键，同时计算和（我们要计算的平均值的分子）和计数（我们要计算的平均值的分母）：

>>> aTuple = (0,0) # As of Python3, you can't pass a literal sequence to a function.
>>> rdd1 = rdd1.aggregateByKey(aTuple, lambda a,b: (a[0] + b,    a[1] + 1),
                                       lambda a,b: (a[0] + b[0], a[1] + b[1]))

上面每对a和b的含义如下所示（这样您就可以看到发生了什么）：

   First lambda expression for Within-Partition Reduction Step::
   a: is a TUPLE that holds: (runningSum, runningCount).
   b: is a SCALAR that holds the next Value

   Second lambda expression for Cross-Partition Reduction Step::
   a: is a TUPLE that holds: (runningSum, runningCount).
   b: is a TUPLE that holds: (nextPartitionsSum, nextPartitionsCount).

最后，计算每个键的平均值，并收集结果。

>>> finalResult = rdd1.mapValues(lambda v: v[0]/v[1]).collect()
>>> print(finalResult)
      [(u'2013-09-09', 11.235365503035176),
       (u'2013-09-01', 23.39500642456595),
       (u'2013-09-03', 13.53240060820617),
       (u'2013-09-05', 13.141148418977687),
   ... snip ...
  ]

我希望这个用aggregateByKey()回答的问题会有帮助。

Answer 2

只是添加了一个关于这个问题的直观和简短（但不好）的解决方案的注释。书Sam's Teach Yourself Apache Spark in 24 Hours已经在最后一章很好地解释了这个问题。

使用groupByKey可以像这样轻松地解决问题：

rdd = sc.parallelize([
        (u'2013-10-09', 10),
        (u'2013-10-09', 10),
        (u'2013-10-09', 13),
        (u'2013-10-10', 40),
        (u'2013-10-10', 45),
        (u'2013-10-10', 50)
    ])

rdd \
.groupByKey() \
.mapValues(lambda x: sum(x) / len(x)) \
.collect()

输出：

[('2013-10-10', 45.0), ('2013-10-09', 11.0)]

这是直观和吸引人的，但不要使用它！groupByKey不在映射器上进行任何组合，并将所有单独的键值对带到还原器。

尽量避免groupByKey。使用类似于@pat的reduceByKey解决方案

Answer 3

在我看来，一个更具可读性的等价物是带有两个lambdas的aggregateByKey：

rdd1 = rdd1 \
    .mapValues(lambda v: (v, 1)) \
    .reduceByKey(lambda a,b: (a[0]+b[0], a[1]+b[1]))

这样，整个平均值计算将是：

avg_by_key = rdd1 \
    .mapValues(lambda v: (v, 1)) \
    .reduceByKey(lambda a,b: (a[0]+b[0], a[1]+b[1])) \
    .mapValues(lambda v: v[0]/v[1]) \
    .collectAsMap()