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