Flink: Connected Components - type mismatch error

Question

I am trying to run connected components algorithm on a graph using the Scala API as shown in the programming guide and other examples.

val graph = Graph.fromDataSet(vertices, edges, env).getUndirected
val maxIterations = 10
val components = graph.run(new ConnectedComponents(maxIterations))

And I get the following error:

Type mismatch, expected: GraphAlgorithm[Long,String,Long,NotInferedT], actual: ConnectedComponents[Nothing,Nothing]

And even if I add

val components = graph.run(new ConnectedComponents[Long,String,Long](maxIterations))

I get:

Type mismatch, expected: GraphAlgorithm[Long,String,Long,NotInferedT], actual: ConnectedComponents[Long,String]

My imports are these:

import org.apache.flink.api.scala._
import org.apache.flink.graph.library.ConnectedComponents
import org.apache.flink.graph.{Vertex, Edge}
import org.apache.flink.graph.scala.Graph

Can someone please explain why this is happening?

Answer 1

the ConnectedComponents Gelly library algorithm takes 2 type parameters, the vertex ID type and the edge value type, so you need to call it like this for example graph.run(new ConnectedComponents[Long, NullValue](maxIterations). Also, since it is a Java implementation, make sure to import java.lang.Long. You can also look at the org.apache.flink.graph.scala.example.ConnectedComponents which uses the GSA version of the library algorithm.

Answer 2

This looks like a typical java/scala type mismatch. Please check again whether you use java.lang.Long or scala.Long in this case.

Answer 3

The problem is that the ConnectedComponents implementation expects vertices to have a java.lang.Long vertex value. Unfortunately, scala.Long and java.lang.Long are not type compatible. Thus, in order to use the algorithm, your vertices data set must be of type DataSet[K, java.lang.Long] with K being an arbitrary key type.

Answer 4

vasia and Till Rohrmann were right, but in my case the whole problem backtracks to the creation of vertices and edges and not just the usage of the connected components algorithm. I was creating vertices and edges using scala.Long and not java.lang.Long.