distributed-bfs.scala

def bfs(graph: RDD[(Int, List[Int])], source: Int): Unit = {
  // Boolean constants
  val VISITED: boolean = true
  val NOT_VISITED: boolean = false
  val IN_QUEUE: boolean = true
  val NOT_IN_QUEUE: boolean = false

  // Creating the traversal object
  var traversal = graph.map{
  (vertex_id, adj_list) =>  
    if(vertex_id == source) 
      (vertex_id, adj_list, VISITED, IN_QUEUE) 
    else 
      (vertex_id, adj_list, NOT_VISITED, NOT_IN_QUEUE)
  }
  var nextStep:boolean = true
  
  while(nextStep) {
    val open_queue = traversal.filter{
      (vertex_id, adj_list, visited_flag, queue_flag) => 
        queue_flag && !visited_flag // in the queue and not visited
    }
    val queue_nodes = open_queue{
      (vertex_id, adj_list, visited_flag, queue_flag) => vertex_id
    }.collect()

    if(open_queue.isEmpty)
      nextStep = false // Stop condition
    else{
      /* Identify neighbors */
      val neighbors = open_queue.flatMap{
        (vertex_id, adj_list, visited_flag, queue_flag) => adj_list
      }

      /* Update traversed information to all vertices */
      val nextTraversal = traversal.map{
        (vertex_id, adj_list, visited_flag, queue_flag) =>
          if (neighbors.contains(vertex_id))
            (vertex_id, adj_list, visited_flag, IN_QUEUE)
          else if (queue_nodes.contains(vertex_id)) =>
            (vertex_id, adj_list, VISITED, NOT_IN_QUEUE) // set as visited
          else
            (vertex_id, adj_list, visited_flag, queue_flag)
      }
      
      traversal = nextTraversal
    }
  }
}