Spark Scala中的深度搜索

libraryDependencies += "org.apache.spark" %% "spark-graphx" % "3.3.0"

import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD

// 创建顶点RDD
val vertices: RDD[(VertexId, String)] = RDD(
  (1L, "A"),
  (2L, "B"),
  (3L, "C"),
  (4L, "D")
)

// 创建边RDD
val edges: RDD[Edge[String]] = RDD(
  Edge(1L, 2L, "edge1"),
  Edge(1L, 3L, "edge2"),
  Edge(2L, 4L, "edge3"),
  Edge(3L, 4L, "edge4")
)

// 创建图
val graph = Graph(vertices, edges)

import org.apache.spark.graphx.lib.ShortestPaths

// 定义DFS函数
def dfs(graph: Graph[String, String], startVertex: VertexId): Unit = {
  val visited = scala.collection.mutable.Set[VertexId]()

  def visit(vertex: VertexId): Unit = {
    if (!visited.contains(vertex)) {
      println(s"Visited vertex: ${graph.vertices.lookup(vertex).head}")
      visited.add(vertex)
      graph.edges.filter(_.srcId == vertex).collect().foreach { edge =>
        visit(edge.dstId)
      }
    }
  }

  visit(startVertex)
}

// 执行DFS
dfs(graph, 1L)

val shortestPaths = ShortestPaths.run(graph, Seq(1L))
shortestPaths.vertices.collect().foreach { case (vertexId, path) =>
  println(s"Shortest paths from 1 to $vertexId: $path")
}