[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:

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<string, object[]>((str) => new object[] { 1, "abc" });

// 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;
}