Scala: lowerByKey для нахождения максимума в кортеже

Question

У меня простой rdd выглядит как

scala> gp_by_cls.collect
res82: Array[(Int, (Int, Int))] = Array((3,(1,26)), (1,(0,37)), (3,(0,77)), (1,(2,914)), (2,(1,13)), (1,(1,25)), (2,(2,893)), (3,(2,1661)), (2,(0,42)))

Я хочу получить максимум ._2 для каждой клавиши (1661, 914, 893), поэтому я сделал

scala> gp_by_cls.reduceByKey{case((a,b),(c,d)) => b if (b > d) else d}

но получите следующую ошибку:

<console>:1: error: ';' expected but 'if' found.
gp_by_cls.reduceByKey{case((a,b),(c,d)) => b if (b > d) else d}
                                             ^

Я что-то здесь упускаю?

Answer 1

ReducebyKey принимает в качестве параметра функцию, которая возвращает тот же тип значений в вашем PairRDD, в этом случае ваш RDD равен RDD[(Int, (Int, Int)], поэтому тип возвращаемого значения должен быть (Int, Int), ваша функция должна быть:

gp_by_cls.reduceByKey{
    case ((a, b), (c, d)) => if(b > d) (a, b) else (c, d)
}

Answer 2

1. Ваш оператор if имеет проблему. Это должно быть if (b > d) b else d
2. Вы не можете использовать функцию reduceByKey после выполнения collect над СДР. Пока это все еще RDD / DataFrame / Dataset, вы можете выполнить reduceByKey