如何在火花中分组和加起来？ [复制]答案

【问题标题】：How to group and add up in spark? [duplicate]如何在火花中分组和加起来？ [复制]
【发布时间】：2016-08-20 02:58:56
【问题描述】：

我有一个这样的 RDD：

{"key1" : "fruit" , "key2" : "US" , "key3" : "1" }

{"key1" : "fruit" , "key2" : "US" , "key3" : "2" }

{"key1" : "vegetable" , "key2" : "US" , "key3" : "1" }

{"key1" : "fruit" , "key2" : "Japan" , "key3" : "3" }

{"key1" : "vegetable" , "key2" : "Japan" , "key3" : "3" }

我的目标是先按 key1 分组，然后按 key2 分组最后添加 key3。

我期待最终的结果，

key1          key2      key3
"fruit"     , "US"    , 3
"vegetable" , "US"    , 1
"fruit"     , "Japan" , 3
"vegetable" , "Japan" , 3

我的代码开头如下，

rdd_arm = rdd_arm.map(lambda x: x[1])

rdd_arm 包含上面的 key : value 格式。

我不确定下一步该去哪里。有人能帮帮我吗？

【问题讨论】：

标签： python apache-spark pyspark distributed-computing rdd

【解决方案1】：

我自己解决了。

我必须创建一个包含多个密钥的密钥，然后加起来。

rdd_arm.map( lambda x : x[0] + ", " + x[1] , x[2] ).reduceByKey( lambda a,b : a + b )

以下问题很有用。

How to group by multiple keys in spark?

【讨论】：

请允许我说这对我不起作用，我遇到了未定义名称的错误，在得到它们之后，我无法使其工作。结果我发布了一个新答案，希望你喜欢它！我赞成这个问题，因为它让我练习！谢谢！

【解决方案2】：

让我们创建你的 RDD：

In [1]: rdd_arm = sc.parallelize([{"key1" : "fruit" , "key2" : "US" , "key3" : "1" }, {"key1" : "fruit" , "key2" : "US" , "key3" : "2" }, {"key1" : "vegetable" , "key2" : "US" ,  "key3" : "1" }, {"key1" : "fruit" , "key2" : "Japan" , "key3" : "3" }, {"key1" : "vegetable" , "key2" : "Japan" , "key3" : "3" }])
In [2]: rdd_arm.collect()
Out[2]: 
[{'key1': 'fruit', 'key2': 'US', 'key3': '1'},
 {'key1': 'fruit', 'key2': 'US', 'key3': '2'},
 {'key1': 'vegetable', 'key2': 'US', 'key3': '1'},
 {'key1': 'fruit', 'key2': 'Japan', 'key3': '3'},
 {'key1': 'vegetable', 'key2': 'Japan', 'key3': '3'}]

首先，您必须创建一个新密钥，即key1 和key2 对。它的值是key3，所以你想做这样的事情：

In [3]: new_rdd = rdd_arm.map(lambda x: (x['key1'] + ", " + x['key2'], x['key3']))

In [4]: new_rdd.collect()
Out[4]: 
[('fruit, US', '1'),
 ('fruit, US', '2'),
 ('vegetable, US', '1'),
 ('fruit, Japan', '3'),
 ('vegetable, Japan', '3')]

然后，我们要添加重复键的值，只需调用reduceByKey()，如下所示：

In [5]: new_rdd = new_rdd.reduceByKey(lambda a, b: int(a) + int(b))

In [6]: new_rdd.collect()
Out[6]: 
[('fruit, US', 3),
 ('fruit, Japan', '3'),
 ('vegetable, US', '1'),
 ('vegetable, Japan', '3')]

我们完成了！

当然，这可以是单行的，像这样：

new_rdd = rdd_arm.map(lambda x: (x['key1'] + ", " + x['key2'], x['key3'])).reduceByKey(lambda a, b: int(a) + int(b))

【讨论】：

你好 gsamaras 。感谢您的跟进。