在 Spark 中为每个执行器创建单例对象

Question

我在这里描述了一个非常相似的问题： How to perform one operation on each executor once in spark 我在第一个答案中遵循了第一种方法，但仍然遇到了序列化问题。

我想要做的是，我有像 (sourceVertex, targetVertex) 元组这样的查询，并将这些查询发送给执行程序，执行程序将返回给我一条最短路径。为此，我正在使用 jgrapht。

当我这样实现时

class ShortestPath(graph: SimpleDirectedWeightedGraph[Node, DefaultWeightedEdge], 
bc: Broadcast[SimpleDirectedWeightedGraph[Node, DefaultWeightedEdge]]) extends Serializable {

  def calculateShortestPath(vertexRDD: RDD[Node]) = {

    val result = vertexRDD.map(vertex => {
      val dijkstraShortestPath: DijkstraShortestPath[Node, DefaultWeightedEdge] 
                = new DijkstraShortestPath[Node, DefaultWeightedEdge](bc.value)
      val distanceIn = dijkstraShortestPath.getPath(vertex, Node(4, 1, true)).getWeight()
      distanceIn
    })
    result.collect().foreach(println(_))
  }

}

object ShortestPath {
  def apply(graph: SimpleDirectedWeightedGraph[Node, DefaultWeightedEdge], 
bc: Broadcast[SimpleDirectedWeightedGraph[Node, DefaultWeightedEdge]]): ShortestPath = new ShortestPath(graph, bc)
}

一切正常 但问题是我认为我正在为每个任务创建dijkstraShortestPath 对象，对吗？

我的目标是为每个执行器创建这个对象，并将它用于该执行器上的每个任务。

我给出的链接说用惰性 val 创建一个对象，在这里实例化你的想法，然后使用它 RDD 映射函数。我这样实现该解决方案：

object Dij {
  lazy val dijsktra = {
    val graph = GraphCreator.createGraph()
    val dijkstraShortestPath: DijkstraShortestPath[Node, DefaultWeightedEdge] = new DijkstraShortestPath[Node, DefaultWeightedEdge](graph)
    dijkstraShortestPath
  }
}

在 ShortestPath 类中使用

    val result = vertexRDD.map(vertex => {
      val dijkstraShortestPath = Dij.dijsktra
      val distanceIn = dijkstraShortestPath.getPath(vertex, Node(4, 1, true)).getWeight()
      dijkstraShortestPath
    })

    result.collect().foreach(println(_))

但后来我收到序列化错误谢谢说 - object not serializable (class: org.jgrapht.alg.shortestpath.DijkstraShortestPath, value: org.jgrapht.alg.shortestpath.DijkstraShortestPath@2cb8e13b) 没错，当我查看实现时，没有可序列化。

另一个问题是，如果它不是可序列化的，那么我的第一个实现是如何工作的？

Answer 1

我认为您的第二个 sn-p 中有一个意外错误。

给映射的第一个函数返回一个权重（大概是Double？），第二个返回一个不可序列化的DijkstraShortestPath。这就解释了为什么两个 sn-ps 的行为不同。