具有元组值的Spark Python中的reduceByKey

Question

我的RDD为(key, (val1,val2))。为此rdd，我想应用reduceByKey函数，我的要求是val2针对单个键找到的最小值，并提取val1结果的最小值val2。例如：(1,(a,4)),(2,(b,3)),(1,(c,2)),(2,(d,1)) 在此，我希望结果集为(1,(c,2)),(2,(d,1))我在python代码下面有想法，但是在这里我得到了第一个val1，而不是val1与minimum对应的代码val2。rdd2 = rdd1.map(lambda x:(x[0],(x[1],x[3])))rdd3 = rdd2.reduceByKey(lambda x,y:(x[0],min(x[1],y[1])))请帮助我修改代码，以便获得所需的结果。

Answer 1

你需要的一切都if else在reduceByKey功能

rdd3 = rdd2.reduceByKey(lambda x, y: x if(x[1] < y[1]) else y)

这应该给你你想要的输出为

#(1, ('c', 2))

#(2, ('d', 1))

我希望答案是有帮助的

Answer 2

在这里，我将通过使用scala获得所需的结果，因为我现在已经学到了更多scala，所以我现在可以回答我自己的问题，如下所示：

val list = List((1,('a',4)),(2,('b',3)),(1,('c',2)),(2,('d',1)))

val rdd = sc.parallelize(list)

rdd.reduceByKey((rec1,rec2) => if(rec1._2 > rec2._2) rec2 else rec1).foreach(println)

输出：

（2，（d，1））

（1，（c，2））

我在这里提到了此代码，因为其他人可以参考并发现它很有用。谢谢...

Answer 3

min与key参数一起使用：

from functools import partial

from operator import itemgetter

rdd.reduceByKey(partial(min, key=itemgetter(1)))