Spark: [FEATURE REQUEST]: UDF to return custom business objects

[guruvonline]

I have a C# component which is presently running in Azure Data lake and i am planning to move to Spark and reuse the same component.
My example scenario

C# takes an input of Manager Dataset like

| mgrId | name |
| ------- | ------ |
| 11 | ABC |
| 22 | DEF |

C# component returns a List of Reportee, where Reportee is Defined as
Class { public int EmpId; public string Name; public string Role; public int MgrId; }

Reportee dataset

| empId | name | role | mgrId |
| ----- | ----- | ----- | ----- |
| 100 | pqr | admin | 11 |
| 200 | stu | reader | 11 |
| 300 | wxy | reader | 22 |

intended UDF
var udf = Udf<int, List<Reportee>>((mgrId) => return component.Execute(mgrId); });

for each row in my Manager dataset, i have to call UDF to get final result in spark as

| mgrId | mgrname | empname | empid | Role |
| --- | --- | ---- | ---- | ---- |
| 11 | ABC | pqr | 100 | admin |
| 11 | ABC | stu | 200 | reader |
| 22 | DEF | wxy | 300 | reader |

[imback82]

The UDT (user-defined type) as a return type of UDF will not be supported. (UDT API in Spark became private since 2.0, and not much traction in PR, etc.)

However, we plan to achieve something similar using StructType. This is how it's done in PySpark.

I did a quick prototype, and it looks like the following:

```C#
var schema = new StructType(new[] {
new StructField("col1", new IntegerType()),
new StructField("col2", new StringType()) });
// The schema is hard-coded inside Udf<> for POC, but "Row" class will be used to embed
// schema and object[].
var udf = Udf

// Assume that we have a df that has:
//+----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
var udfDf = df.Select(udf(df["name"]).As("udf_col"));

// PrintSchema() prints:
// root
// |-- udf_col: struct (nullable = true)
// | |-- col1: integer (nullable = false)
// | |-- col2: string (nullable = true)
udfDf.PrintSchema();

// Show() prints:
// +--------+
// | udf_col|
// +--------+
// |[1, abc]|
// |[1, abc]|
// |[1, abc]|
// +--------+
udfDf.Show();

// Flatten nested column as follows:
// +----+----+
// |col1|col2|
// +----+----+
// | 1| abc|
// | 1| abc|
// | 1| abc|
// +----+----+
udfDf.Select(udfDf["udf_col.col1"], udfDf["udf_col.col2"]).Show();
// or
udfDf.Select("udf_col.*").Show();
```

This feature be available in coming weeks.

[jatcar95]

I'm a bit confused on how the schema variable is being used above. Do you mean it will be possible to pass it to the Udf call in the future release?

[imback82]

Yes, that was a proof of concept. You will be able to pass the schema when you register the udf.

[suhsteve]

Proposed usage for UDT like return type using GenericRow:

var schema = new StructType(new[] {
    new StructField("col1", new IntegerType()),
    new StructField("col2", new StringType()) });

// schema will be last parameter of the Udf helper class)
var udf = Udf<string, GenericRow>((str) => new GenericRow(new object[] { 1, "abc" }), schema);

// Assume that we have a df that has:
//+----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+
var udfDf = df.Select(udf(df["name"]).As("udf_col"));

// PrintSchema() prints:
// root
// |-- udf_col: struct (nullable = true)
// |    |-- col1: integer (nullable = false)
// |    |-- col2: string (nullable = true)
udfDf.PrintSchema();

// Show() prints:
// +--------+
// | udf_col|
// +--------+
// |[1, abc]|
// |[1, abc]|
// |[1, abc]|
// +--------+
udfDf.Show();

Column col1 = udfDf.Select("udf_col.col1") // or udfDf.Select("udf_col").Select("col1")

You may add a extra parameter schema to the Udf<> helper class and set it to null by default. Push the schema object all the way, from Functions.Udf<T1, ..., TResult> => ... => Functions.CreateUdf (where it creates the Json ReturnType). From there we can check if the schema == null and either get the json or use the UdfUtils.GetReturnType method.

[imback82]

var udf = Udf<string, Row>((str) => new GenericRow(new object[] { 1, "abc" }), schema);

Any reason why one is GenericRow and the other is Row?

You may add a extra parameter schema to the Udf<> helper class and set it to null by default.

You have to always specify the schema for the Row. Is the plan to check the return type and if it's a Row type then schema cannot be null?

[elvaliuliuliu]

Any reason why one is GenericRow and the other is Row?

Since Row constructor is internal, we could not call it when writing examples in Basic.cs for instance.

You have to always specify the schema for the Row. Is the plan to check the return type and if it's a Row type then schema cannot be null?

Yes, schema is null by default, and if return type is RowType, schema cannot be null otherwise throw exception?

[imback82]

Any reason why one is GenericRow and the other is Row?

Since Row constructor is internal, we could not call it when writing examples in Basic.cs for instance.

I think this would confuse the user.

You have to always specify the schema for the Row. Is the plan to check the return type and if it's a Row type then schema cannot be null?

Yes, schema is null by default, and if return type is RowType, schema cannot be null otherwise throw exception?

What if we expose a function that is specialized for returning Row types? Did you explore this option as well?

[elvaliuliuliu]

Any reason why one is GenericRow and the other is Row?

Since Row constructor is internal, we could not call it when writing examples in Basic.cs for instance.

I think this would confuse the user.

Yes, I agree, but we have internal Row(object[] values, StructType schema).

You have to always specify the schema for the Row. Is the plan to check the return type and if it's a Row type then schema cannot be null?

Yes, schema is null by default, and if return type is RowType, schema cannot be null otherwise throw exception?

What if we expose a function that is specialized for returning Row types? Did you explore this option as well?

Yes, I initially thought of exposing a function within UdfUtils.GetReturnType, but it seems like this function is mostly for return types that either generic types or are implemented from generic types. Should we make a separate return type function only for Row types like UdfUtils.GetReturnTypeAsRowType outside of UdfUtils.GetReturnType?

[imback82]

Why not something like Udf_SomeOtherName<T>(T t, Schema s) for UDFs that return Row? It's awkward to allow schema for almost all other types that don't even need it.

So we have two options:
1) Have a different name for UDF that returns Row type. The API is clear, but the downside is we need a separate set; you cannot specialize on generic types in C#.
2) Allow specifying schema for UDF, but we need to make sure there is no confusion on Row and GenericRow, especially if you are going from an object with less info to more info (GenericRow -> Row, how can it logically be possible?)

We can explore both options and get some early feedback.

[suhsteve]

var udf = Udf<string, Row>((str) => new GenericRow(new object[] { 1, "abc" }), schema);

Any reason why one is GenericRow and the other is Row?

Unless we need the schema from the Row object when pickling (maybe it can be fetched from the defined returnType of the UDF), I wanted to avoid something like:

var udf = Udf<string, Row>((str) => new Row(new object[] { 1, "abc" }, schema), schema);

You may add a extra parameter schema to the Udf<> helper class and set it to null by default.

You have to always specify the schema for the Row. Is the plan to check the return type and if it's a Row type then schema cannot be null?

I was thinking of something like the following for the public Udf API

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf)
{
    return CreateUdf<TResult>(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf)).Apply2;
}

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow
{
    return CreateUdf<TResult>(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply2;
}

[imback82]

As I explained above, it doesn't make sense to do

var udf = Udf<string, Row>((str) => new GenericRow(new object[] { 1, "abc" }), schema);

since Row is supposed to have more info (schema) than GenericRow. Can't you just use GenericRow here instead? I remember getting the prototype working just using object[].

[suhsteve]

As I explained above, it doesn't make sense to do

var udf = Udf<string, Row>((str) => new GenericRow(new object[] { 1, "abc" }), schema);

since Row is supposed to have more info (schema) than GenericRow. Can't you just use GenericRow here instead? I remember getting the prototype working just using object[].

Sure I agree, we can change it to something like:

var udf = Udf<string, GenericRow>((str) => new GenericRow(new object[] { 1, "abc" }), schema);

Although choosing to go this route will depend on if we need the schema info when pickling the row and if we do, if this can be fetched from the returnType of the UDF (I haven't looked into that part yet). Otherwise, we may need to do something like the following instead:

var udf = Udf<string, Row>((str) => new Row(new object[] { 1, "abc" }, schema), schema);

[elvaliuliuliu]

I have a prototype working as follows using Option 1 - Have a different name for UDF that returns GenericRow type.
I am using object[] for now, once we have GenericRow merged, I can change it to GenericRow instead.

```C#
var spark = SparkSession.Builder().GetOrCreate();
var df = spark.Range(0, 5);

Func udfReturnRowTypeTest = UdfReturnRowType r => new object[] { r + 100 },
new StructType(new[]
{
new StructField("id", new IntegerType())
}));

// Assume that we have a df that has:
//+---+
// | id |
//+---+
// | 0 |
// | 1 |
// | 2 |
// | 3 |
// | 4 |
//+---+

var udfDf = df.Select(udfReturnRowTypeTest(df["id"]).As("udf_col"));

// PrintSchema() prints:
// root
// |-- udf_col: struct (nullable = true)
// | |-- id: integer (nullable = true)

udfDf.PrintSchema();

// Show() prints:
// +--------+
// | udf_col|
// +--------+
// | [100] |
// | [101] |
// | [102] |
// | [103] |
// | [104] |
// +--------+
udfDf.Show();

// Flatten nested column as follows:
//+---+
// | id |
//+---+
// | 100 |
// | 101 |
// | 102 |
// | 103 |
// | 104 |
//+---+
udfDf.Select(udfDf["udf_col.col1"]).Show();
// or
udfDf.Select("udf_col.*").Show();

And I have exposed `UdfReturnRowType` like the following:
```C#
public static Func<Column, Column> UdfReturnRowType<T, TResult>(Func<T, TResult> udf, StructType returnType)
{
    return CreateUdfReturnRowType<TResult>(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply1;
}

@imback82 @suhsteve : Please let me know what you think and we can move forward from there. Thanks!

[suhsteve]

@elvaliuliuliu can you get it to work with returning GenericRow/Row instead of object[] ? Also can you get it to work with Udf instead of UdfReturnRowType ?

[elvaliuliuliu]

@elvaliuliuliu can you get it to work with returning GenericRow/Row instead of object[] ? Also can you get it to work with Udf instead of UdfReturnRowType ?

1) UdfReturnRowType will work with GenericRow and object[] not Row here, since UdfReturnRowType takes schema outside. Moreover, if we are going to explore with UdfReturnRowType method, should it be aiming at taking GenericRow instead of Row? Please correct me if I misunderstand anything here.
2) Work with Udf instead of UdfReturnRowType: Do you mean the second option which @imback82 mentioned earlier?

[suhsteve]

@elvaliuliuliu

        Elva Liu
        FTE can you get it to work with returning `GenericRow`/`Row` instead of `object[]` ? Also can you get it to work with `Udf` instead of `UdfReturnRowType` ?

1. UdfReturnRowType will work with GenericRow and object[] not Row here, since UdfReturnRowType takes schema outside. Moreover, if we are going to explore with UdfReturnRowType method, should it be aiming at taking GenericRow instead of Row? Please correct me if I misunderstand anything here.

2. Work with Udf instead of UdfReturnRowType: Do you mean the second option which @imback82

          Terry Kim
          FTE mentioned earlier?
   

I'm not a big fan of introducing a new API name just for a row return type. I think we can try out either adding a constraint to the type parameter, or only accepting funcs with an output type of Row. ie:

public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

// or

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

We can defer to @imback82 and see which approach he would like to take.

[elvaliuliuliu]

I'm not a big fan of introducing a new API name just for a row return type. I think we can try out either adding a constraint to the type parameter, or only accepting funcs with an output type of Row. ie:

public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

// or

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

How would the above two options you proposed be for other types e.g. ArrayType, MapType when they are using the same Udf API?

[imback82]

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

I think this should be good.

[suhsteve]

I'm not a big fan of introducing a new API name just for a row return type. I think we can try out either adding a constraint to the type parameter, or only accepting funcs with an output type of Row. ie:

public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

// or

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

How would the above two options you proposed be for other types e.g. ArrayType, MapType when they are using the same Udf API?

Are you planning to have separate Udf methods for all these types ? UdfRowReturnType, UdfArrayReturnType, UdfMapReturnType ?

[elvaliuliuliu]

I'm not a big fan of introducing a new API name just for a row return type. I think we can try out either adding a constraint to the type parameter, or only accepting funcs with an output type of Row. ie:

public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

// or

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

How would the above two options you proposed be for other types e.g. ArrayType, MapType when they are using the same Udf API?

Are you planning to have separate Udf methods for all these types ? UdfRowReturnType, UdfArrayReturnType, UdfMapReturnType ?

Nope, Udf already has the functionality to take return type as ArrayType, MapType per UdfUtils.GetReturnType as mentioned earlier. I believe RowType is different from them since it's not implemented from generic type based on previous discussion.

[suhsteve]

I'm not a big fan of introducing a new API name just for a row return type. I think we can try out either adding a constraint to the type parameter, or only accepting funcs with an output type of Row. ie:

public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

// or

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

How would the above two options you proposed be for other types e.g. ArrayType, MapType when they are using the same Udf API?

Are you planning to have separate Udf methods for all these types ? UdfRowReturnType, UdfArrayReturnType, UdfMapReturnType ?

Nope, Udf already has the functionality to take return type as ArrayType, MapType per UdfUtils.GetReturnType as mentioned earlier. I believe RowType is different from them since it's not implemented from generic type based on previous discussion.

If the current Udf implementation already works with ArrayType and MapType then I don't understand your previous question.

[suhsteve]

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

I think this should be good.

Actually, it looks like this approach will not work when GenericRow is a sealed class. Seems like the current options would be

1) Go with the new API name, UdfReturnRowType
2) public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

[elvaliuliuliu]

public static Func<Column, Column, Column> Udf<T1, T2, TResult>(Func<T1, T2, TResult> udf, StructType returnType)
    where TResult : GenericRow

I think this should be good.

Actually, it looks like this approach will not work when GenericRow is a sealed class. Seems like the current options would be

1. Go with the new API name, UdfReturnRowType
2. public static Func<Column, Column, Column> Udf<T1, T2>(Func<T1, T2, GenericRow> udf, StructType returnType)

Yes, I agree. The following won't work since GenericRow is a sealed class.
```C#
public static Func Udf(Func udf, StructType returnType)
where TResult : GenericRow

I have working prototypes for the following two options. And I think they both have pros and cons.

Option1 - New API (Separate API for `RowType` only)
```C#
public static Func<Column, Column> UdfReturnRowType<T, TResult>(Func<T, TResult> udf, StructType returnType)
{
    return CreateUdfReturnRowType<TResult>(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply1;
}

Option2 - Take only inputs and accept output as GenericRow (Different structure since user doesn't have to specify output type)
C# public static Func<Column, Column> Udf<T>(Func<T, GenericRow> udf, StructType returnType) { return CreateUdfReturnRowType<GenericRow>(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply1; }

[imback82]

You don't need to specify return type for option 1 either, the reason why it has a special name. So, option 1 and 2 are essentially the same except for the name. Option 3 is marking GenericRow as non-sealed?

[elvaliuliuliu]

You don't need to specify return type for option 1 either, the reason why it has a special name. So, option 1 and 2 are essentially the same except for the name. Option 3 is marking GenericRow as non-sealed?

Yes, you are right. We don't have to specify return type for option 1 either. Option 3 as below will also work if GenericRow is a non-sealed class. From user's point of view, I would prefer option 3 which has a seamless user experience. But not sure if it would be a good idea to mark GenericRow a non-sealed class.
C# public static Func<Column, Column> Udf<T, TResult>(Func<T, TResult> udf, StructType returnType) where TResult : GenericRow { return CreateUdfReturnRowType<TResult>(udf.Method.ToString(), UdfUtils.CreateUdfWrapper(udf), returnType).Apply1; }