运行PySp中的自定义Java类

Question

我试图在PySpark中运行一个自定义HDFS阅读器类。这个类是用Java编写的，我需要从PySpark访问它，可以从shell访问，也可以使用spark submit访问。

在PySpark中，我从SparkContext（sc._gateway）检索JavaGateway。

假设我有课：

package org.foo.module

public class Foo {

    public int fooMethod() {
        return 1;
    }

}

我试图将它打包到一个jar中，并将它与--jar选项一起传递给pyspark，然后运行：

from py4j.java_gateway import java_import

jvm = sc._gateway.jvm
java_import(jvm, "org.foo.module.*")

foo = jvm.org.foo.module.Foo()

但我得到了错误：

Py4JError: Trying to call a package.

有人能帮忙吗？谢谢。

Answer 1

您应该使用--packages将包导入到spark-submit操作中，而不是--jars。

Answer 2

在PySpark中尝试以下操作

from py4j.java_gateway import java_import
java_import(sc._gateway.jvm,"org.foo.module.Foo")

func = sc._gateway.jvm.Foo()
func.fooMethod()

确保您已经将Java代码编译成一个可运行的jar，并像这样提交spark作业

spark-submit --driver-class-path "name_of_your_jar_file.jar" --jars "name_of_your_jar_file.jar" name_of_your_python_file.py

Answer 3

您描述的问题通常表示org.foo.module不在驱动程序类路径上。一种可能的解决方案是使用spark.driver.extraClassPath添加jar文件。例如，它可以在conf/spark-defaults.conf中设置，也可以作为命令行参数提供。

旁注：

• 如果您使用的类是自定义输入格式，那么就不需要使用Py4j网关。您可以简单地使用SparkContext.hadoop*/SparkContext.newAPIHadoop*方法。

• 使用java_import(jvm, "org.foo.module.*")看起来是个坏主意。一般来说，您应该避免在JVM上进行不必要的导入。这不是公开的原因，你真的不想搞砸。尤其是当访问方式使此导入完全过时时。所以放下java_import，坚持jvm.org.foo.module.Foo()。