让我们创建您的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。感谢您的跟进。 –