GroupedGraph.scala

import spark.sqlContext.implicits._

case class GroupedKvKRecord(groupId: Long, kvkRecords: Seq[KvKRecord])

// need to broadcast for rdd in rdd mapping
val bcVertices = spark.sparkContext.broadcast(vertices.collectAsMap)

// connectedComponents is the magic here
val groupedKvKRecords = linkedGraph
    .connectedComponents
    .vertices
    .map {
      case (id, smallestId) => (smallestId, id)
    }
    .groupByKey
    .mapValues(
        _.map(bcVertices.value(_)).toList // calling vertices directly in other rdd mapping will result in nullpointerexception 
    )
    .toDF("groupId", "kvkRecords")
    .as[GroupedKvKRecord]