scala - 创建一个新的 scala 类，它依赖于没有序列化问题的 GraphFrames

Question

我正在尝试编写一个基于 Spark 的 GraphFrames 的 Scala 类。 GraphFrame 类定义为 here . GraphFrames 类扩展可序列化。我想编写一个类来扩展 GraphFrames 并计算一些额外的图形属性。为了简化这个例子，我创建了一个不包含任何函数的类。它所做的只是扩展 GraphFrames:

import org.apache.spark.sql.DataFrame
import org.graphframes._

class NewGraphFrame(@transient private val _vertices: DataFrame,
                    @transient private val _edges: DataFrame) extends GraphFrame {

}

val vertices = Seq(
  (1, "John"),
  (2, "Jane"),
  (3, "Karen")
).toDF("id", "name")

val edges = Seq(
  (1, 3),
  (2, 3),
  (2, 1)
).toDF("src", "dst")

val g = new NewGraphFrame(vertices, edges)

当我在 REPL 中运行这段代码时，出现以下错误:

java.lang.Exception: You cannot use GraphFrame objects within a Spark closure
  at org.graphframes.GraphFrame.vertices(GraphFrame.scala:125)
  at org.graphframes.GraphFrame.toString(GraphFrame.scala:55)
  at scala.runtime.ScalaRunTime$.scala$runtime$ScalaRunTime$$inner$1(ScalaRunTime.scala:332)
  at scala.runtime.ScalaRunTime$.stringOf(ScalaRunTime.scala:337)
  at scala.runtime.ScalaRunTime$.replStringOf(ScalaRunTime.scala:345)
  at .$print$lzycompute(<console>:10)
  at .$print(<console>:6)
  at $print(<console>)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:786)
  at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1047)
  at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:638)
  at scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:637)
  at scala.reflect.internal.util.ScalaClassLoader$class.asContext(ScalaClassLoader.scala:31)
  at scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:19)
  at scala.tools.nsc.interpreter.IMain$WrappedRequest.loadAndRunReq(IMain.scala:637)
  at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:569)
  at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:565)
  at scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:807)
  at scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:681)
  at scala.tools.nsc.interpreter.ILoop.processLine(ILoop.scala:395)
  at scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:415)
  at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply$mcZ$sp(ILoop.scala:923)
  at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:909)
  at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:909)
  at scala.reflect.internal.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:97)
  at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:909)
  at org.apache.spark.repl.Main$.doMain(Main.scala:76)
  at org.apache.spark.repl.Main$.main(Main.scala:56)
  at org.apache.spark.repl.Main.main(Main.scala)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
  at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:894)
  at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:198)
  at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:228)
  at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:137)
  at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我知道这意味着我要连载两次。但是，我显然对此不感兴趣。我只是想扩展这个类，以便我可以在我的类中使用图形功能。如何在 spark repl 不抛出此错误的情况下扩展此类？

Answer 1

有趣的是，从未发生过序列化，因此这并不是真正的序列化问题。

GraphFrame 的定义归结为:

class GraphFrame private(private val _vertices: DataFrame, private val _edges) extends Logging with Serializable

第一个 private 意味着主构造函数只能从 GraphFrame 或其伴随对象中访问。由于 _vertices 和 _edges 是私有(private)的，它们不能从子类访问并且不能被覆盖(您的 _vertices 和 _edges字段与 GraphFrame 中该名称的字段不同:您不能使用 override 作为前缀将证明这一点。

GraphFrame 确实定义了一个公共(public)零参数构造函数:

def this() = this(null, null)

GraphFrame 的任何子类因此对于 GraphFrame._vertices 或 GraphFrame._edges 将具有 null，这是如果您尝试序列化一个 GraphFrame，还会有什么。访问器 vertices 和 edges 检查 null 并抛出。由于这些访问器是从 toString 方法调用的，这就是为什么在您创建 NewGraphFrame 时 REPL 崩溃的原因。 GraphFrame 似乎假设如果它们是 null 则意味着尝试了序列化。

这可以在 REPL 中展示:

new NewGraphFrame(vertices, edges).vertices

这将导致 null。

如果您使用检索 NewNewGraphFrame._vertices 的方法覆盖 vertices(请注意没有 _)方法(并且对关于 edges)，你至少能够 toString 新类:

override def vertices: DataFrame = _vertices
override def edges: DataFrame = _edges

就您感兴趣的 GraphFrame 中的方法而言，使用 vertices/edges 而不是 _vertices/_edges 您将能够利用 GraphFrame 的现有方法。