Use the mapValues api. I made an example of doing what you wanted below. You'll have to update the listbuffer to use the types you have, but it looks like its doing what you want. val rdd1 = sc.parallelize(Array((1,2),(2,3),(1,3),(2,4)))
val gRdd = rdd1.groupByKey()
val indxRdd = gRdd.mapValues(a => {
val b = a.toArray
var indx = 2
val lb = new ListBuffer[(Int, Int)]
for(i <- 0 to b.size-1) {
lb.append((b(i), indx))
indx += 1
}
lb.toArray
}).collectAsMap()
indxRdd.collectAsMap()
res8: scala.collection.Map[Int,Array[(Int, Int)]] =
Map(
2 -> Array((3,2), (4,3)),
1 -> Array((2,2), (3,3))
)
... View more
