它的py4j gatewayServer在spark的入口点是什么？

Question

我试着用py4j在pyspark中运行一个java函数。Py4j使得能够访问java虚拟机中的java对象。我创建了另一个JVM实例，并且能够成功地运行java函数。

py4j通过GatewayServer实例实现这种通信。

我想知道我们是否可以以某种方式访问spark的内部JVM来运行我的java函数？py4j网关服务器在spark中的入口点是什么？如何将我的函数添加到入口点？

Answer 1

我不确定这是否是您需要的，但我看到了两个地方：

sc._gateway.jvm

可用于java_import或直接用于

sc._jvm

因此，要访问包a.b.c中的类X，您可以执行以下操作之一：

jvm = sc._gateway.jvm
java_import(jvm,"a.b.c.X")
instance = a.b.c.X()

或者更直接地：

instance = sc._jvm.a.b.c.X()

要添加一个java函数，你需要确保它在类路径中，如果你想在workers中使用它(例如在UDF中)，那么你需要把它发送给workers。要实现这一点，您可以使用-- driver -class-path开关将spark-submit (或pyspark)添加到驱动程序中，并使用--jars将其发送给工作程序。

Answer 2

看

$SPARK_HOME/python/pyspark/java_gateway.py

您将在那里看到用于与Java/Scala后端交互的机制。

您将需要更新一个或多个Java文件，如下所示：

java_import(gateway.jvm, "org.apache.spark.SparkConf")
java_import(gateway.jvm, "org.apache.spark.api.java.*")
java_import(gateway.jvm, "org.apache.spark.api.python.*")
java_import(gateway.jvm, "org.apache.spark.mllib.api.python.*")
# TODO(davies): move into sql
java_import(gateway.jvm, "org.apache.spark.sql.*")
java_import(gateway.jvm, "org.apache.spark.sql.hive.*")
java_import(gateway.jvm, "scala.Tuple2")

这些表示Spark-Java入口点。

Pyspark使用Spark-Java入口点，而不是直接使用Scala。您需要- (a)在这些API类中使用现有的入口点，或者- (b)在这些类中添加新的入口点，并构建您自己的Spark版本