title: Implementing topological sort in Spark GraphX
body: <p>I am trying to implement <a href="https://en.wikipedia.org/wiki/Topological_sorting#Kahn.27s_algorithm" rel="nofollow noreferrer">topological sort</a> using <a href="http://spark.apache.org/graphx/" rel="nofollow noreferrer">Spark's GraphX</a> library.</p>
  
  <p>This is the code I've written so far:</p>
  
  <p><strong><em>MyObject.scala</em></strong></p>
  
  <pre><code>import java.util.ArrayList
  
  import scala.collection.mutable.Queue
  
  import org.apache.spark.SparkConf
  import org.apache.spark.SparkContext
  import org.apache.spark.graphx.Edge
  import org.apache.spark.graphx.EdgeDirection
  import org.apache.spark.graphx.Graph
  import org.apache.spark.graphx.Graph.graphToGraphOps
  import org.apache.spark.graphx.VertexId
  import org.apache.spark.rdd.RDD
  import org.apache.spark.rdd.RDD.rddToPairRDDFunctions
  
  object MyObject {
  
    def main(args: Array[String]): Unit = {
      val conf = new SparkConf().setAppName("Spark-App").setMaster("local[2]")
      val sc = new SparkContext(conf)
  
      val resources: RDD[Resource] = makeResources(sc)
      val relations: RDD[Relation] = makeRelations(sc)
  
      println("Building graph ...")
      var graph = buildGraph(resources, relations, sc)
      println("Graph built!!")
  
      println("Testing topo sort ...")
      val topoSortResult = topoSort(graph, sc);
      println("topoSortResult = " + topoSortResult)
      println("Testing topo sort done!")
    }
  
    def buildGraph(resources: RDD[Resource], relations: RDD[Relation], sc: SparkContext): Graph[Resource, Relation] =
      {
        val vertices: RDD[(Long, Resource)] = resources.map(resource =&gt; (resource.id, resource))
        val edges: RDD[Edge[Relation]] = relations.map(relation =&gt; Edge(relation.srcId, relation.dstId, relation))
        var graph = Graph[Resource, Relation](vertices, edges)
        graph
      }
  
    def makeResources(sc: SparkContext): RDD[Resource] =
      {
        var list: List[Resource] = List()
        list = list :+ new Resource(1L)
        list = list :+ new Resource(2L)
        list = list :+ new Resource(3L)
        list = list :+ new Resource(4L)
        list = list :+ new Resource(5L)
        sc.parallelize(list)
      }
  
    def makeRelations(sc: SparkContext): RDD[Relation] =
      {
        var list: List[Relation] = List()
        list = list :+ new Relation(1L, "depends_on", 2L)
        list = list :+ new Relation(3L, "depends_on", 2L)
        list = list :+ new Relation(4L, "depends_on", 2L)
        list = list :+ new Relation(5L, "depends_on", 2L)
        sc.parallelize(list)
  
      }
  
    def topoSort(graph: Graph[Resource, Relation], sc: SparkContext): java.util.List[(VertexId, Resource)] =
      {
        // Will contain the result
        val sortedResources: java.util.List[(VertexId, Resource)] = new ArrayList()
  
        // Contains all the vertices
        val vertices = graph.vertices
  
        // Contains all the vertices whose in-degree &gt; 0
        val inDegrees = graph.inDegrees;
        val inDegreesKeys_array = inDegrees.keys.collect();
  
        // Contains all the vertices whose in-degree == 0
        val inDegreeZeroList = vertices.filter(vertex =&gt; !inDegreesKeys_array.contains(vertex._1))
  
        // A map of vertexID vs its in-degree
        val inDegreeMapRDD = inDegreeZeroList.map(vertex =&gt; (vertex._1, 0)).union(inDegrees);
  
        // Insert all the resources whose in-degree == 0 into a queue
        val queue = new Queue[(VertexId, Resource)]
        for (vertex &lt;- inDegreeZeroList.toLocalIterator) { queue.enqueue(vertex) }
  
        // Get an RDD containing the outgoing edges of every vertex
        val neighbours = graph.collectNeighbors(EdgeDirection.Out)
  
        // Initiate the algorithm
        while (!queue.isEmpty) {
          val vertex_top = queue.dequeue()
          // Add the topmost element of the queue to the result
          sortedResources.add(vertex_top)
  
          // Get the neigbours (from outgoing edges) of this vertex
          // This will be an RDD containing just 1 element which will be an array of neighbour vertices
          val vertex_neighbours = neighbours.filter(vertex =&gt; vertex._1.equals(vertex_top._1))
  
          // For each vertex, decrease its in-degree by 1
          vertex_neighbours.foreach(arr =&gt; {
            val neighbour_array = arr._2
            neighbour_array.foreach(vertex =&gt; {
              val oldInDegree = inDegreeMapRDD.filter(vertex_iter =&gt; (vertex_iter._1 == vertex._1)).first()._2
              val newInDegree = oldInDegree - 1
              // Reflect the new in-degree in the in-degree map RDD
              inDegreeMapRDD.map(vertex_iter =&gt; {
                if (vertex_iter._1 == vertex._1) {
                  (vertex._1, newInDegree)
                }
                else{
                  vertex_iter
                }
              });
              // Add this vertex to the result if its in-degree has become zero
              if (newInDegree == 0) {
                queue.enqueue(vertex)
              }
            })
          })
        }
  
        return sortedResources
      }
  
  }
  </code></pre>
  
  <p><strong><em>Resource.scala</em></strong></p>
  
  <pre><code>class Resource(val id: Long) extends Serializable {
    override def toString(): String = {
      "id = " + id
    }
  }
  </code></pre>
  
  <p><strong><em>Relation.scala</em></strong></p>
  
  <pre><code>class Relation(val srcId: Long, val name: String, val dstId: Long) extends Serializable {
    override def toString(): String = {
      srcId + " " + name + " " + dstId
    }
  }
  </code></pre>
  
  <p>I am getting the error :</p>
  
  <pre><code>org.apache.spark.SparkException: RDD transformations and actions can only be invoked by the driver, not inside of other transformations; for example, rdd1.map(x =&gt; rdd2.values.count() * x) is invalid because the values transformation and count action cannot be performed inside of the rdd1.map transformation. For more information, see SPARK-5063.
  </code></pre>
  
  <p>for the line <code>val oldInDegree = inDegreeMapRDD.filter(vertex_iter =&gt; (vertex_iter._1 == vertex._1)).first()._2</code>.</p>
  
  <p>I guess this is because it is illegal to modify an RDD inside the for-each loop of some other RDD.</p>
  
  <p>Also, I fear that <code>queue.enqueue(vertex)</code> will not work, since <a href="https://stackoverflow.com/a/23396500/1925388">it is not possible to modify a local collection inside a <code>for-each</code> loop</a>.</p>
  
  <p><strong>How do I correctly implement this topological sort algorithm ?</strong></p>
  
  <p>The full stack trace of the exception is uploaded <a href="https://gist.github.com/Anmol-Singh-Jaggi/95e084d33824c8af97888304a3460d34" rel="nofollow noreferrer">here</a> <em>(Had to upload it externally to prevent exceeding the body size limit of StackOverflow)</em>. </p>