Nullability in spark 3

[amilosevic-grid]

Hi everyone,
I have been trying to use Abris with Schema registry and spark 3. From what I read in the docs, starting from abris 5.x, Spark3 using newer avro versions should be able to handle nullable fields. However, I do not get that behaviour.

this is the schema of a json I have been using:

 |-- Address: struct (nullable = true)
 |    |-- code: string (nullable = true)
 |    |-- latitude: double (nullable = true)
 |    |-- longitude: double (nullable = true)
 |    |-- postal_code: long (nullable = true)
 |-- Age: string (nullable = true)
 |-- Location: string (nullable = true)
 |-- name: struct (nullable = true)
 |    |-- first: string (nullable = true)
 |    |-- second: string (nullable = true)

if I use the AvroSchemaUtils to convert it I get:

scala> AvroSchemaUtils.toAvroSchema(df, "Address")
res1: org.apache.avro.Schema = [{"type":"record","name":"topLevelRecord","fields":[{"name":"code","type":["string","null"]},{"name":"latitude","type":["double","null"]},{"name":"longitude","type":["double","null"]},{"name":"postal_code","type":["long","null"]}]},"null"]

This schema will not work when decoding; it can be registered on the Schema Registry; however when trying to deserialize:

Caused by: org.apache.spark.sql.avro.IncompatibleSchemaException: Cannot convert Avro type [{"type":"record","name":"Address","namespace":"hbaseTable","fields":[{"name":"code","type":["string","null"]},{"name":"latitude","type":["double","null"]},{"name":"longitude","type":["double","null"]},{"name":"postal_code","type":["long","null"]}]},"null"] to SQL type STRUCT<code: STRING, latitude: DOUBLE, longitude: DOUBLE, postal_code: BIGINT>.
        at org.apache.spark.sql.avro.AvroDeserializer.liftedTree1$1(AvroDeserializer.scala:102)
      ...
Caused by: org.apache.spark.sql.avro.IncompatibleSchemaException: Attempting to treat union as a RECORD, but it was: UNION

Is there a way to use these methods, without explicitly casting the fields to be non-nullable before registering the schema?

what I see even from spark-avro_2.12:3.2.1, it will always return a union:
SchemaConverters:201

    if (nullable && catalystType != NullType) {
      Schema.createUnion(schema, nullSchema)
    } else {
      schema
    }

which is called from Abris at AvroSchemaUtils using the fields regular nullability value

 SchemaConverters.toAvroType(field.dataType, field.nullable, recordName, nameSpace)

[cerveada]

Can you change the Avro schema before registering it non-nulable types?

[cerveada]

In your case this may suffice:

val newSchema = 
  if(schema.isUnion) schema.getTypes.get(0)
  else schema

[amilosevic-grid]

Hi again @cerveada,

thank you for your help. In the end, this snippet helped me to get to a schema that I wanted(currently without validation on union types, but thats not hard to add), so leaving it here in case somebody encounters the same scenario:

          val df = setNullableStateOfColumn(dataFrame, recordName, false)
          val schema = AvroSchemaUtils.toAvroSchema(df, recordName, recordName, recordNamespace)

          // reverse types
          val fields = schema.getFields.toArray.map(f => f.asInstanceOf[Field])

          val newFields = fields.map(f => {
            val reverseTypes = List(f.schema().getTypes.get(1), f.schema().getTypes.get(0)).asJava
            val newTypes: Schema = Schema.createUnion(reverseTypes)
            new Field(f.name(), newTypes, f.doc(), Schema.Field.NULL_DEFAULT_VALUE)
          })

          val newSchema = Schema.createRecord(recordName, schema.getDoc, recordNamespace, false , newFields.toList.asJava)

Its a case of schema evolution where Id want to pass in a default value; in case a field is added in the future(convention is for that field to be null if its missing). So I had to reverse from ["string","null"] in the union to the opposite ["null", "string"], in order to be able to set null as a default value. All the fields of the schema have to be changed in this way so a new record has to be created.
the schema at the end looks something like this:

{"type":"record","name":"Address","namespace":"testTable","fields":[{"name":"code","type":["null","string"],"default":null},{"name":"latitude","type":["null","double"],"default":null},{"name":"longitude","type":["null","double"],"default":null},{"name":"postal_code","type":["null","long"],"default":null}]}

starting from

[{"type":"record","name":"Address","namespace":"testTable","fields":[{"name":"code","type":["string","null"]},{"name":"latitude","type":["double","null"]},{"name":"longitude","type":["double","null"]},{"name":"postal_code","type":["long","null"]}]},"null"]

it would be nice to have some kind of control over default fields after creating the schema from a spark column; especially in streaming if a field evolves. I will mark this as answered, but if you ever decide to open an issue on this topic ping me, I would like to help 👍🏻

keep up the good work!

[cerveada]

Thanks, for sharing the solution.

[kevinwallimann]

Hi @amilosevic-grid If you would like more control over the creation of the Avro schema, you can implement your own logic to convert a Spark schema to an Avro schema. Abris relies on Spark's SchemaConverters class, but there is not one definitive way how to convert from a Spark to an Avro schema. Especially the handling of default values and metadata may not be sufficient to your needs.

• One option may be to modify an Avro schema after creating it with SchemaConverters, as done e.g. here: https://github.com/AbsaOSS/hyperdrive/blob/143c569dcfec09a0dc8e32b7ffd3d4ad90d65cbd/ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/utils/AbrisConfigUtil.scala#L111
• A more radical option is to rewrite the SchemaConverters class altogether, as done here: https://github.com/AbsaOSS/hyperdrive/blob/v4.7.0/ingestor-default/src/main/scala/za/co/absa/hyperdrive/ingestor/implementation/transformer/avro/confluent/AdvancedSparkToAvroConverter.scala

[edward-capriolo-db]

I had to modify schema converters for just this issue.
I put in a PR back in the day but it got canceled b because I didn't follow up.