[Question] Is there a way to use the GenericRow to load a complex type in a MapType?

[thedanfields]

Hi!
I'm currently using v1 of the Microsoft.Spark nuget packages and the microsoft-spark-2-4_2.11-1.0.0.jar.

I've currently got the need to have a DataFrame schema which looks similar to this:

new StructType(
    new List<StructField> {
        new StructField("Identifier", new LongType()),
        new StructField("NestedComplexMap", new MapType(new IntegerType(),
            new StructType(new List<StructField>() {
                new StructField("FirstProp", new LongType()),
                new StructField("SecondProp", new LongType()),
                new StructField("ThirdProp", new LongType()),
            }))),
    }
)

I'm trying to write tests by generating data using a GenericRow to hydrate a DataFrame like so:

var nestedComplexMap = new Dictionary<int, Dictionary<string, long>>() {
                    { 1, new Dictionary<string, long> { { "FirstProp", 1L }, { "SecondProp", 1L }, { "ThirdProp", 1L } } }
                };
var complexMapRow = new GenericRow(new object[] { identifier, nestedComplexMap });

var complexMapData = new List<GenericRow>();
complexMapData.Add(complexMapRow);

var complexMapFrame = session.CreateDataFrame(complexMapData, new StructType(
                            new List<StructField> {
                                new StructField("Identifier", new LongType()),
                                new StructField("NestedComplexMap", new MapType(new IntegerType(),
                                    new StructType(new List<StructField>() {
                                        new StructField("FirstProp", new LongType()),
                                        new StructField("SecondProp", new LongType()),
                                        new StructField("ThirdProp", new LongType()),
                                    }))),
                            }));
complexMapFrame.PrintSchema();
complexMapFrame.Show();

During execution, I receive run time errors stating:

20/11/17 10:50:44 ERROR DotnetBackendHandler: Failed to execute 'createDataFrame' on 'org.apache.spark.sql.SparkSession' with args=([Type=scala.collection.convert.Wrappers.SeqWrapper, Value: [[1,{1={FirstProp=1, SecondProp=1, ThirdProp=1}}]]], [Type=org.apache.spark.sql.types.StructType, Value: StructType(StructField(Identifier,LongType,true), StructField(NestedComplexMap,MapType(IntegerType,StructType(StructField(FirstProp,LongType,true), StructField(SecondProp,LongType,true), StructField(ThirdProp,LongType,true)),true),true))])
[2020-11-17T15:50:44.7081200Z] [] [Error] [JvmBridge] JVM method execution failed: Nonstatic method 'createDataFrame' failed for class '5' when called with 2 arguments ([Index=1, Type=List`1, Value=System.Collections.Generic.List`1[Microsoft.Spark.Sql.GenericRow]], [Index=2, Type=JvmObjectReference, Value=9], )
[2020-11-17T15:50:44.7082803Z] [] [Error] [JvmBridge] java.lang.IllegalArgumentException: The value ({FirstProp=1, SecondProp=1, ThirdProp=1}) of the type (scala.collection.convert.Wrappers.MapWrapper) cannot be converted to struct<FirstProp:bigint,SecondProp:bigint,ThirdProp:bigint>

Is there a way to do accomplish creating this kind of data using the GenericRow?
Thanks so much for your time and efforts!

[Niharikadutta]

Hi @thedanfields, thank you for your question!
Can you try using this as the schema you pass to CreateDataFrame (The nested Dictionary corresponds to the MapType and not StructType):

var schema = new StructType(
                new List<StructField> {
                    new StructField("Identifier", new LongType()),
                    new StructField("NestedComplexMap", new MapType(new IntegerType(),
                        new MapType(new StringType(), new LongType()))),
                }
            );

It showed me the following result for PrintSchema and Show:

[thedanfields]

@Niharikadutta !!! Thanks so much.

Out of curiosity, what c# type should I use for a StructType, if I wanted to keep the original schema I provided?

Thanks again for your time and attention! Really appreciate it.

[Niharikadutta]

I'm not sure I understood your question, but if you want to use StructType as the original schema, you would need to change your nestedComplexMap to use StructType instead of Dictionary in the nested type.

[thedanfields]

Hey @Niharikadutta, thanks for your response.

I see why I was unclear.

Ultimately I do not have control of the schema, it's being dictated by a process upstream. My goal is to be able to use the GenericRow to create some fake data to mock the result of that upstream process.

I am stuck with this schema:

new StructType(
    new List<StructField> {
        new StructField("Identifier", new LongType()),
        new StructField("NestedComplexMap", new MapType(new IntegerType(),
            new StructType(new List<StructField>() {
                new StructField("FirstProp", new LongType()),
                new StructField("SecondProp", new LongType()),
                new StructField("ThirdProp", new LongType()),
            }))),
    }
)

I've tried a number of ways of using the GenericRow to adhere to the schema, but I can't seem to figure out the right way.

This example:

 var row = new GenericRow(new object[] { 
                1,  // Identifier
                new Dictionary<int, GenericRow> { // ?? NestedComplexMap ??
                    { 1, new GenericRow(new object[] {1L, 2L, 3L }) },
                    { 2, new GenericRow(new object[] {1L, 2L, 3L }) }
                }
            });
            var dataFrame = _session.CreateDataFrame(new [] { row }, schema);

            var srows = dataFrame.Collect().ToArray();

Results in an Exception stating: System.Threading.SemaphoreFullException : Adding the specified count to the semaphore would cause it to exceed its maximum count.

I've also naively tried substituting the Dictionary<int, GenericRow> with Dictionary<int, NestedComplexType> where NestedComplexType is some class i've defined which contains the FirstProp, SecondProp and ThirdProp properties. I then receive an error related to Spark not supporting the type NestedComplexType, which makes sense.

So given that schema, do you know how I can fabricate a DataFrame ?

Thanks again for your attention here. I really appreciate the support.

[Niharikadutta]

@thedanfields Thanks for explaining your scenario and apologies for the late reply. It is currently not possible to create a DataFrame for the schema you have provided (or nested DataFrames in general) because we do not support GenericRow as an argument as of yet. We can talk about supporting this in the next release though. @imback82 , @suhsteve thoughts?

[imback82]

@Niharikadutta, Can you check if we can make the following work?

 var row = new GenericRow(new object[] { 
                1, 
                new Dictionary<int, GenericRow> {
                    { 1, new GenericRow(new object[] {1L, 2L, 3L }) },
                    { 2, new GenericRow(new object[] {1L, 2L, 3L }) }
                }
            });
var dataFrame = _session.CreateDataFrame(new [] { row }, schema);

[thedanfields]

@Niharikadutta, thanks again for the attention to this.

Just to follow up, While I wasn't able to directly create the needed data, I managed to do it indirectly with two DataFrames and some joins.

[Niharikadutta]

@Niharikadutta, Can you check if we can make the following work?

 var row = new GenericRow(new object[] { 
                1, 
                new Dictionary<int, GenericRow> {
                    { 1, new GenericRow(new object[] {1L, 2L, 3L }) },
                    { 2, new GenericRow(new object[] {1L, 2L, 3L }) }
                }
            });
var dataFrame = _session.CreateDataFrame(new [] { row }, schema);

Yes @imback82 I did test this and it failed with [Exception] [JvmBridge] Type Microsoft.Spark.Sql.GenericRow not supported yet, so was wondering if we want to add the support here, will test it out.

[truveug123]

@Niharikadutta, Can you check if we can make the following work?

 var row = new GenericRow(new object[] { 
                1, 
                new Dictionary<int, GenericRow> {
                    { 1, new GenericRow(new object[] {1L, 2L, 3L }) },
                    { 2, new GenericRow(new object[] {1L, 2L, 3L }) }
                }
            });
var dataFrame = _session.CreateDataFrame(new [] { row }, schema);

Yes @imback82 I did test this and it failed with [Exception] [JvmBridge] Type Microsoft.Spark.Sql.GenericRow not supported yet, so was wondering if we want to add the support here, will test it out.

@Niharikadutta I'm facing a similar issue here with using nested type. Essentially, I want to add a new row to a dataframe with a schema that has a nested type. But I'm also seeing an error with "not supported yet".

Here's my code (running on Synapse spark 3.2)

var innerSchema = new StructType(
    new List<StructField>()
    {
        new StructField("inner", new StringType()),
        new StructField("inner2", new StringType()),
    }
);
var arrayType = new ArrayType( innerSchema );
var arraySchema = new StructType(
    new List<StructField>()
    {
        new StructField("array", arrayType),
    }
);

public struct MyStruct
{
    public string A { get; }
    public string B { get; }

    public MyStruct(string a, string b)
    {
        A = a;
        B = b;
    }
} 

var arrayRow = new List<GenericRow>{ 
    new GenericRow( new object[]{
        new List<object>() { new MyStruct("inner1.1", "inner1.2"), 
        new MyStruct("inner2.1", "inner2.2") }
     })
};

var testDF = spark.CreateDataFrame( arrayRow, arraySchema );

error message:

[Exception] [JvmBridge] Type Submission#66+MyStruct not supported yet
   at Microsoft.Spark.Interop.Ipc.PayloadHelper.GetTypeId(Type type)
   at Microsoft.Spark.Interop.Ipc.PayloadHelper.ConvertArgsToBytes(MemoryStream destination, Object[] args, Boolean addTypeIdPrefix)
   at Microsoft.Spark.Interop.Ipc.PayloadHelper.ConvertArgsToBytes(MemoryStream destination, Object[] args, Boolean addTypeIdPrefix)
   at Microsoft.Spark.Interop.Ipc.PayloadHelper.ConvertArgsToBytes(MemoryStream destination, Object[] args, Boolean addTypeIdPrefix)
   at Microsoft.Spark.Interop.Ipc.PayloadHelper.BuildPayload(MemoryStream destination, Boolean isStaticMethod, Int32 processId, Int32 threadId, Object classNameOrJvmObjectReference, String methodName, Object[] args)
   at Microsoft.Spark.Interop.Ipc.JvmBridge.CallJavaMethod(Boolean isStatic, Object classNameOrJvmObjectReference, String methodName, Object[] args)
System.Threading.SemaphoreFullException: Adding the specified count to the semaphore would cause it to exceed its maximum count.

[Niharikadutta]

@Niharikadutta, thanks again for the attention to this.

Just to follow up, While I wasn't able to directly create the needed data, I managed to do it indirectly with two DataFrames and some joins.

@thedanfields could you elaborate a little more on how you created it indirectly with two DataFrames? Thanks!