Register UDF from external Java jar class in pyspa

2019-08-29 00:08发布

问题:

This question already has an answer here:

  • Spark: How to map Python with Scala or Java User Defined Functions? 1 answer
  • Calling Java/Scala function from a task 1 answer

I have Java Jar which contains functions, as example

package com.test.oneid;
public class my_class {

    public static void main(String args[]) {

    }

    public static int add(int x) throws IOException {       
        try {
            return (x+2);
        } catch(Exception e) {
            throw new IOException("Caught exception processing input row ", e);
        }
    }
}

in my spark session i included this jar with --jars option.

from pyspark.sql.types import *
from pyspark.sql.functions import *
from py4j.java_gateway import java_import
java_import(sc._gateway.jvm,"com.test.oneid.my_class")
my_func = sc._gateway.jvm.my_class()

def add_udf(s):
    x=my_func.add(s)
    return x

add_udf(10)

till here it works, but when i try to register this as UDF to use in Spark SQL or Dataframe, getting following error,

>>> spark.udf.register('my_udf', add_udf, IntegerType())
Traceback (most recent call last):
  File "/usr/hdp/current/spark2-client/python/pyspark/cloudpickle.py", line 147, in dump
    return Pickler.dump(self, obj)
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 224, in dump
    self.save(obj)
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 554, in save_tuple
    save(element)
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/hdp/current/spark2-client/python/pyspark/cloudpickle.py", line 248, in save_function
    self.save_function_tuple(obj)
  File "/usr/hdp/current/spark2-client/python/pyspark/cloudpickle.py", line 296, in save_function_tuple
    save(f_globals)
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 655, in save_dict
    self._batch_setitems(obj.iteritems())
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 692, in _batch_setitems
    save(v)
  File "/data1/anaconda/anaconda2/lib/python2.7/pickle.py", line 306, in save
    rv = reduce(self.proto)
  File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1133, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 63, in deco
    return f(*a, **kw)
  File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py", line 323, in get_return_value
    format(target_id, ".", name, value))
Py4JError: An error occurred while calling o57.__getnewargs__. Trace:
py4j.Py4JException: Method __getnewargs__([]) does not exist
        at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
        at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
        at py4j.Gateway.invoke(Gateway.java:272)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:214)
        at java.lang.Thread.run(Thread.java:745)

I am not sure what the error is, Thanks for the help in advance.

EDIT: The error is not helpful here, i tried following aswell.

>>> sqlContext.registerJavaFunction("udf", "com.test.oneid.my_class.add_udf")
18/09/06 18:20:56 ERROR UDFRegistration: Can not load class com.test.oneid.my_class.add_udf, please make sure it is on the classpath

I need to register this, so that i can use it in case statements in Spark Sql but following is not helpful in that regard - Register UDF to SqlContext from Scala to use in PySpark