Spark：如何在map（python）中生成多個元素？

Question

我有由鍵和值列表組成的數據。 RDD =（鍵，[超值]）

什麼，我想要的是：

rdd.map(lambda pair: func(pair)) 

其中函數返回若干（價值NEW_VALUE）的一對。請注意，New_value也取決於Key！

所以，簡單地說，在

(key1, [Value1, Value2]) 
(key2, [Value3, Value4, Value5]) 

使用.map()後，我想收到：

(Value1, NewValue1) 
(Value2, NewValue2) 
(Value3, NewValue3) 
(Value4, NewValue4) 
(Value5, NewValue5) 

我怎樣才能使它在PySpark？

Answer 1

讓func成爲連接的鍵值。

def func(kv): 
    return '-'.join(kv) 

你想所有列表拉平至（鍵，值）對

rdd.flatMap(lambda kv: ((kv[0], v) for v in kv[1])) 

然後映射在所有對func同時保持初始值作爲結果

map(lambda kv: (kv[1], func(kv))) 

的一部分

總之，

rdd.flatMap(lambda kv: ((kv[0], v) for v in kv[1])).map(lambda kv: (kv[1], func(kv))) 

實施例

>>> rdd = sc.parallelize([('k1', ['v1', 'v2']), ('k2', ['v3','v4','v5'])]) 
>>> rdd.flatMap(lambda kv: ((kv[0], v) for v in kv[1])).map(lambda kv: (kv[1], func(kv))).collect() 
[('v1', 'k1-v1'), 
('v2', 'k1-v2'), 
('v3', 'k2-v3'), 
('v4', 'k2-v4'), 
('v5', 'k2-v5')] 

Answer 2

我找到不同的使用yield

>>> rdd = sc.parallelize([('k1', ['v1', 'v2']), ('k2', ['v3','v4','v5'])]) 
>>> def func(kv): 
...  for v in kv[1]: 
...   yield (v, kv[0] + '-' + v) 
>>>rdd.flatMap(lambda kv: func(kv)).collect() 
[('v1', 'k1-v1'), 
('v2', 'k1-v2'), 
('v3', 'k2-v3'), 
('v4', 'k2-v4'), 
('v5', 'k2-v5')]