lowerByKey в искровом питоне со значениями кортежа

Question

У меня есть СДР (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, соответствующий минимуму 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, (д, 1))

(1, (с, 2))

Я упоминал этот код здесь, потому что другие могут ссылаться и находить его полезным. Спасибо ...

Answer 3

Использовать min с аргументом key:

from functools import partial
from operator import itemgetter

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