Add UDF support #106

nonpool · 2021-08-24T03:05:28Z

This pr is a complement to #67

solution

After many attempts, I gave up on hack RowEncoder, because it always has inexplicable bugs.
When register UDF, I directly convert DataTypeWithClass to spark DataType, and this solution seems to be working well.

problem

Please see UDFRegisterTest.kt line134, when use a complex type as the return value of UDF, we will got an exception(same as line134 link). This exception also exists in java(i test in my local on spark 3.1.1).So this bug we needs to wait for the official spark to fix or we try to hack related classes to fix it(maybe more effort)

Possible improvements

Add more unit test. Actually, I don’t know some common cases of UDF. you can give some help for this

…>()) as DataTypeWithClass).dt().json())

nonpool · 2021-08-24T03:05:50Z

@asm0dey @FelixEngl
After you review these commits, I will add 3.0 support and related documents.

asm0dey · 2021-08-24T06:32:46Z

Looks like returning complex classes works in Scala: https://stackoverflow.com/a/59293056

Are you sure it may not be supported in Kotlin?

asm0dey · 2021-08-24T06:36:02Z

kotlin-spark-api/2.4/src/main/kotlin/org/jetbrains/kotlinx/spark/api/UDFRegister.kt

+ */
+@OptIn(ExperimentalStdlibApi::class)
+inline fun <reified R> UDFRegistration.register(name: String, noinline func: () -> R): UDFWrapper0 {
+    register(name, UDF0(func), schema(typeOf<R>()).unWrapper())


It looks like it makes sense to work not with our generated schema, but with schema from an encoder, isn't it?

Yes. Actually, it is better than hack RowEncoder to adapter our generated schema when register a UDF. reason: 1. RowEncoder only need a correctly DataType, not our generated schema other info. 2. some spark action (eg: .as<Int>() action) to use UDF result is not though RowEncoder,if still use hack class way, we will hack many class and there is no guarantee that hacked classes will work well.

nonpool · 2021-08-24T07:27:49Z

Why don't you extract the schema from the generated encoder?

I will remove this to other file

nonpool · 2021-08-24T07:32:08Z

Looks like returning complex classes works in Scala: https://stackoverflow.com/a/59293056

Are you sure it may not be supported in Kotlin?

I think this is a big plan. If we don’t have the java implementation as a reference, this future should hack many class. i suggest separate this future to other PR

asm0dey · 2021-08-24T09:18:20Z

I will remove this to other file

no-no, I meant that we have encoder method, which allows you to use the schema, which was already processed by spark.

asm0dey · 2021-08-24T09:20:44Z

I think this is a big plan. If we don’t have the java implementation as a reference, this future should hack many class. i suggest separate this future to other PR

But if it works for the Scala API should we really provide something so limited? I mean people will just use Scala API because they will get weird exceptions in runtime which sounds as extremely bad UX. Like "we support UDFs, but only for couple of classes". And BTW do we throw exception if provided class is just regular data class or Java bean class? Or they will work?

nonpool · 2021-08-24T15:42:57Z

I think this is a big plan. If we don’t have the java implementation as a reference, this future should hack many class. i suggest separate this future to other PR

But if it works for the Scala API should we really provide something so limited? I mean people will just use Scala API because they will get weird exceptions in runtime which sounds as extremely bad UX. Like "we support UDFs, but only for couple of classes". And BTW do we throw exception if provided class is just regular data class or Java bean class? Or they will work?

I created a simple example for this in my repo spark-udf-explore, and got some conclusion:

UDF return type is not support java bean in java spark or scala spark
UDF return type is not support normal scala class
UDF return type is only support case class in scala spark

Nevertheless, I still want to try my best to let users have a complete API. I will try to hack related classes to support return data class in kotlin API

nonpool · 2021-08-24T15:49:32Z

I will remove this to other file

no-no, I meant that we have encoder method, which allows you to use the schema, which was already processed by spark.

But our encoder result is wrapper by KDataTypeWrapper KComplexTypeWrapper and so on, and these wrapper class is not support by RowEncoder or other spark native class which handles Datatype. If we use wrapper class register UDF we will hack more class and so on he same reason as I said above

asm0dey · 2021-08-24T19:59:48Z

And that's why we potentially should use not our schema, but schema, which can be obtained from encoder, I think

nonpool · 2021-08-25T00:28:25Z

And that's why we potentially should use not our schema, but schema, which can be obtained from encoder, I think

What the schema can be obtain from encoder means? I think i have used our encoder to obtained schema, just unwrapper our schema when register UDF.
Could you provide some pseudo-code to show your intentions?

asm0dey · 2021-08-25T08:26:05Z

You're calling schema(typeOf<R>()), but you could call encoder<R>().schema()

nonpool · 2021-08-25T09:20:56Z

You're calling schema(typeOf<R>()), but you could call encoder<R>().schema()

A..... Because these codes are copied, and I always thought that this is the encoder used here. ^-^

asm0dey · 2021-08-25T09:33:57Z

Copying code is usually just a bad idea I believe. It increases WTFps dramatically

nonpool · 2021-09-01T03:31:48Z

After testing, Call encoder<R>().schema() to register UDF is not a good way. encoder<R>() method included two action:

call schema(typeOf<R>()) to get schema(type is DataTypeWithClass)
generate serializer and deserializer from schema and construct the ExpressionEncoder

so encoder<R>().schema() is call the schema() method of the ExpressionEncoder which is :

  // The schema after converting `T` to a Spark SQL row. This schema is dependent on the given
  // serializer.
  val schema: StructType = StructType(serializer.map { s =>
    StructField(s.name, s.dataType, s.nullable)
  })

it always wrapper a StructField in StructType. if call encoder<R>().schema() for String it return {"type":"struct","fields":[{"name":"value","type":"string","nullable":true,"metadata":{}}]}, but UDF register only need StringType. Not only string type and primitive type are like this, complex type also not work.

so, i think we should call schema(typeOf<R>()). Unless there are other changes in the complex type.

asm0dey · 2021-09-01T06:02:11Z

On the other hand, we could unwrap this struct, right? And this way we could avoid Spark dealing with our own DataTypeWithClass instances

nonpool · 2021-09-01T06:38:12Z

On the other hand, we could unwrap this struct, right? And this way we could avoid Spark dealing with our own DataTypeWithClass instances

Spark does not directly use our DataTypeWithClass instances. Because i unwrap thought this:

fun DataType.unWrapper(): DataType {
    return when (this) {
        is DataTypeWithClass -> DataType.fromJson(dt().json())
        else -> this
    }
}

asm0dey · 2021-09-01T06:41:17Z

Should it work with deeply nested structures?

nonpool · 2021-09-01T06:45:40Z

List<String> and List< primitive type> also work well.
The Complex type seems to be right when register UDF.

asm0dey · 2021-09-01T06:57:09Z

OK, I'm fine with design, let's port it to 3.0 now!

…ement UDF return data class

asm0dey · 2021-09-05T20:08:18Z

Wowzer, let's just suppress "unused" inspections

asm0dey · 2021-09-05T20:49:39Z

Or just ignore it and et me know when you're ready

Jolanrensen · 2022-06-21T19:30:54Z

This was revamped in #152

See https://github.com/Kotlin/kotlin-spark-api/tree/main#user-defined-functions
and https://github.com/Kotlin/kotlin-spark-api/wiki/UDF.

The original notation was kept intact, just built around it :)

nonpool added 8 commits August 22, 2021 15:54

copy code from Kotlin#67

5f7a0e2

remove hacked RowEncoder.scala

2c161cd

replace all schema(typeOf<R>()) to DataType.fromJson((schema(typeOf<R…

8bbea9a

…>()) as DataTypeWithClass).dt().json())

add return udf data class test

a4bbbf9

change test

59cec6b

add in dataset test for calling the UDF-Wrapper

460350f

add the same exception link

f250694

refactor unWrapper

42b5aa3

asm0dey reviewed Aug 24, 2021

View reviewed changes

add test for udf return a List

456a29e

make the test simpler

6fe2c6b

nonpool added 7 commits September 1, 2021 20:37

add License

e77c58d

add UDFRegister for 3.0

12ea1bd

remove useless import

02169bd

resolved deprecated method

e72acee

[experimental] add CatalystTypeConverters.scala for hacked it to impl…

3398210

…ement UDF return data class

[experimental] implement UDF return data class

fc85d47

fix code inspection issue

e98f6d1

Adds suppre unused

56cd280

asm0dey merged commit 05a3dca into Kotlin:main Sep 13, 2021

Add UDF support #106

Add UDF support #106

Uh oh!

Conversation

nonpool commented Aug 24, 2021

solution

problem

Possible improvements

Uh oh!

nonpool commented Aug 24, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asm0dey commented Aug 24, 2021

Uh oh!

asm0dey Aug 24, 2021

Choose a reason for hiding this comment

Uh oh!

nonpool Aug 24, 2021

Choose a reason for hiding this comment

Uh oh!

nonpool commented Aug 24, 2021

Uh oh!

nonpool commented Aug 24, 2021

Uh oh!

asm0dey commented Aug 24, 2021

Uh oh!

asm0dey commented Aug 24, 2021

Uh oh!

nonpool commented Aug 24, 2021

Uh oh!

nonpool commented Aug 24, 2021

Uh oh!

asm0dey commented Aug 24, 2021

Uh oh!

nonpool commented Aug 25, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asm0dey commented Aug 25, 2021

Uh oh!

nonpool commented Aug 25, 2021

Uh oh!

asm0dey commented Aug 25, 2021

Uh oh!

nonpool commented Sep 1, 2021

Uh oh!

asm0dey commented Sep 1, 2021

Uh oh!

nonpool commented Sep 1, 2021

Uh oh!

asm0dey commented Sep 1, 2021

Uh oh!

nonpool commented Sep 1, 2021

Uh oh!

asm0dey commented Sep 1, 2021

Uh oh!

asm0dey commented Sep 5, 2021

Uh oh!

asm0dey commented Sep 5, 2021

Uh oh!

Jolanrensen commented Jun 21, 2022

Uh oh!

Uh oh!

nonpool commented Aug 24, 2021 •

edited

Loading

nonpool commented Aug 25, 2021 •

edited

Loading