Pyspark：通过键对rdd的值应用reduce答案

【问题标题】：Pyspark: Applying reduce by key to the values of an rddPyspark：通过键对rdd的值应用reduce
【发布时间】：2019-11-02 17:26:23
【问题描述】：

经过一些转换，我最终得到了一个具有以下格式的 rdd：

[(0, [('a', 1), ('b', 1), ('b', 1), ('b', 1)])

(1, [('c', 1), ('d', 1), ('h', 1), ('h', 1)])]

我不知道如何在这个 rdd 的值部分上本质上“reduceByKey()”。

这就是我想要实现的目标：

[(0, [('a', 1), ('b', 3)])

(1, [('c', 1), ('d', 1), ('h', 2)])]

我最初使用 .values() 然后将 reduceByKey 应用于结果，但最终我丢失了原始密钥（在本例中为 0 或 1）。

【问题讨论】：

标签： apache-spark pyspark rdd reduce

【解决方案1】：

您丢失了原始密钥，因为.values() 只会连续获得key-value 的值。您应该对行中的元组求和。

from collections import defaultdict

def sum_row(row):
    result = defaultdict(int)
    for key, val in row[1]:
        result[key] += val
    return (row[0],list(result.items()))

data_rdd = data_rdd.map(sum_row)
print(data_rdd.collect())

# [(0, [('a', 1), ('b', 3)]), (1, [('h', 2), ('c', 1), ('d', 1)])]

【讨论】：

【解决方案2】：

虽然values 提供了RDD，但reduceByKey 对RDD 上的所有值起作用，而不是按行。

您也可以使用groupby（需要订购）来实现相同的效果：

from itertools import groupby

distdata.map(lambda x: (x[0], [(a, sum(c[1]  for c in b)) for a,b in groupby(sorted(x[1]), key=lambda p: p[0]) ])).collect()

【讨论】：