PyFlink - Issue using Scala UDF in JAR

Question

I'm trying to register a Scala UDF in Pyflink using an external JAR as follows, but get below error.

Scala UDF:

package com.dummy

import org.apache.flink.table.functions.ScalarFunction

class dummyTransform(factor: Int) extends ScalarFunction {
  def eval(s: String): Int = {
    s.hashCode()
  }
}

build.sbt:

name := "hello_scala_for_flink"
version := "0.1"
scalaVersion := "2.12.11"
libraryDependencies += "org.apache.flink" % "flink-table-common" % "1.11.2" % "provided"

assembly.sbt:

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.5")

Scala project structure:

hello_scala_for_flink/
  project/
  src/
    main/
      resources/
        scala/
          com.dummy/
            dummyTransform
            hello_scala
    test
  target/
  build.sbt

shell:

cd hello_scala_for_flink
sbt assembly
cp ./target/scala-2.12/hello_scala_for_flink-assembly-0.1.jar /Users/py-r/opt/anaconda3/envs/venv_pyflink_37/lib/python3.7/site-packages/pyflink/lib/

Python:

from pyflink.dataset import ExecutionEnvironment
exec_env = ExecutionEnvironment.get_execution_environment()
#exec_env = StreamExecutionEnvironment.get_execution_environment()

exec_env.set_parallelism(1)
t_config = TableConfig()

table_env = BatchTableEnvironment.create(exec_env, t_config)
table_env.register_java_function("hash_code","com.dummy.dummyTransform")

Error (incl. after restarting Anaconda):

Py4JJavaError: An error occurred while calling o12.newInstance.
: java.lang.InstantiationException: com.dummy.dummyTransform
    at java.base/java.lang.Class.newInstance(Class.java:598)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:564)
    at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
    at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
    at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.lang.NoSuchMethodException: com.dummy.dummyTransform.<init>()
    at java.base/java.lang.Class.getConstructor0(Class.java:3427)
    at java.base/java.lang.Class.newInstance(Class.java:585)
11 more

Versions:

jdk = 1.8.0_151.jdk
scala = 2.12.11
python = 3.7
apache-beam = 2.19.0
apache-flink = 1.11.2

Any idea what the issue might be ?

Thanks for your support

Answer 1

Looks like I found the issue myself. Apparently only a class instantiation was required in above code:

class dummyTransform(factor: Int) extends ScalarFunction {
  def eval(s: String): Int = {
    s.hashCode() * factor
  }
  def this() = this(1)
}

Also, I changed to Scala 2.11.12 due to another error. Now everything seems to work: awesome !