Q:

spark scala type mismatch with zipwithIndex in groupbykey

I am trying to test groupByKey to find nth highest score of a subject
my data looks like this 
scala> a
res176: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[263] at map at <console>:51

scala> a.take(10).foreach{println}
(data science,DN,US,28,98,SMITH,data science)
(maths,DN,US,28,92,SMITH,maths)
(chemistry,DN,US,28,94,SMITH,chemistry)
(physics,DN,US,28,88,SMITH,physics)
(data science,DN,UK,25,93,JOHN,data science)
(maths,DN,UK,25,91,JOHN,maths)
(chemistry,DN,UK,25,95,JOHN,chemistry)
(physics,DN,UK,25,90,JOHN,physics)
(data science,DN,CA,29,67,MARK,data science)
(maths,DN,CA,29,68,MARK,maths)

scala> 

so for the first row "data science" as string is key and "DN,US,28,98,SMITH,data science" is value as a string
now I want to find 2nd highest using group by 
scala> a.groupByKey().flatMap(rec=>{ val max = rec._2.toList.map(x=>x.split(',')(3).toFloat).distinct.sortBy(x=>(-x)).zipWithIndex.filter(x=>x._2==2).toMap.keys
     | rec._2.toList.filter{x=>x.split(',')(3).toFloat==max}
     | }).take(15).foreach{println}

scala> 

I am getting nothing here 
if i run this hard-coded i get value 
scala> a.groupByKey().flatMap(rec=>{ val max = "98"
     | rec._2.toList.sortBy(x=>(-x.split(',')(3).toFloat)).takeWhile(rec=> max.contains(rec.split(',')(3)))}).take(15).foreach{println}
DN,IND,26,98,XMAN,maths
DPS,US,28,98,XOMAN,chemistry
DN,US,28,98,SMITH,data science

also this gives me value 
scala> a.groupByKey().flatMap(rec=>{ rec._2.toList.map(x=>x.split(',')(3).toFloat).distinct.sortBy(x=>(-x)).zipWithIndex.filter(x=>x._2==2).map(_._1)}).take(15).foreach{println}
94.0
92.0
95.0
93.0

some more complex code gives me output
scala> a.groupByKey().flatMap(rec=>{ val max = rec._2.toList.map(x=>x.split(',')(3).toFloat).distinct.sortBy(x=>(-x)).take(1)
     | rec._2.toList.sortBy(x=>(-x.split(',')(3).toFloat)).takeWhile(rec=> max.contains(rec.split(',')(3).toFloat))}).take(15).foreach{println}
DN,IND,26,98,XMAN,maths
DPS,UK,25,96,SOMK,physics
DPS,US,28,98,XOMAN,chemistry
DN,US,28,98,SMITH,data science

looks like there is some data type mismatch when i am using zipwithindex.
Can some one help me here

A:

There is a type mismatch due to .toMap.keys. In the result, val max is of type Iterable[Float], because method keys returns Iterable[A]. 
On of the solution would be an addition of head at the end of max calculation:
  val max = rec._2.toList
    .map(x => x.split(',')(3).toFloat)
    .distinct
    .sortBy(x => (-x))
    .zipWithIndex
    .filter(x => x._2 == 2)
    .toMap
    .keys
    .head

Basically, head will return a value of type Float. Then this code should at least compare equal types x.split(',')(3).toFloat == max.
Although, calling head is not safe method. It may throw an exception, if in your case a filter function can return empty list. Then such exception will be thrown:
java.util.NoSuchElementException: next on empty iterator
Once it work for concrete data sample, you can think to refactor this code to work with Set maybe. Instead of head do .keys.toSet and compare like you did for other examples using max.contains(rec.split(',')(3))

