dnet-dedup/dnet-dedup-test/src/main/java/eu/dnetlib/graph/GraphProcessor.scala

package  eu.dnetlib.graph
import java.lang

import eu.dnetlib.ConnectedComponent
import eu.dnetlib.pace.model.MapDocument
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD

import scala.collection.JavaConversions
;


object GraphProcessor {

  def findCCs(vertexes: RDD[(VertexId,MapDocument)], edges:RDD[Edge[String]], maxIterations: Int): RDD[ConnectedComponent] = {
    val graph: Graph[MapDocument, String] = Graph(vertexes, edges)
    val cc = graph.connectedComponents(maxIterations).vertices

    val joinResult = vertexes.leftOuterJoin(cc).map {
      case (id, (openaireId, cc)) => {
        if (cc.isEmpty){
          (id, openaireId)
        }
        else {
          (cc.get, openaireId)
        }
      }
    }

    val connectedComponents = joinResult.groupByKey().map[ConnectedComponent](cc => asConnectedComponent(cc))

    (connectedComponents)

  }

  def asConnectedComponent(group: (VertexId, Iterable[MapDocument])) : ConnectedComponent = {
    val docs = group._2.toSet[MapDocument]
    val connectedComponent = new ConnectedComponent("empty", JavaConversions.setAsJavaSet[MapDocument](docs));
    connectedComponent.initializeID();
    connectedComponent
  }

}