confluent avro conversion failure

[abhisam]

val commonRegistryConfig = Map(
SchemaManager.PARAM_SCHEMA_REGISTRY_TOPIC -> "test",
SchemaManager.PARAM_SCHEMA_REGISTRY_URL -> "http://testkafkaschema.jganalytics.local",
SchemaManager.PARAM_SCHEMA_NAME_FOR_RECORD_STRATEGY -> "test",
SchemaManager.PARAM_SCHEMA_NAMESPACE_FOR_RECORD_STRATEGY -> "sensor.machine.data"
)

val keyRegistryConfig = commonRegistryConfig +
(SchemaManager.PARAM_KEY_SCHEMA_NAMING_STRATEGY -> SchemaManager.SchemaStorageNamingStrategies.TOPIC_RECORD_NAME,
SchemaManager.PARAM_VALUE_SCHEMA_ID -> "latest")

val valueRegistryConfig = commonRegistryConfig +
(SchemaManager.PARAM_VALUE_SCHEMA_NAMING_STRATEGY -> SchemaManager.SchemaStorageNamingStrategies.TOPIC_RECORD_NAME,
SchemaManager.PARAM_VALUE_SCHEMA_ID -> "latest")

import za.co.absa.abris.avro.functions.to_confluent_avro

SchemaManager.configureSchemaRegistry(commonRegistryConfig)

val result: DataFrame = df
.selectExpr("serialno as key", "to_json(struct(*)) as value")
.select(to_confluent_avro(col("key"), keyRegistryConfig) as 'key,
to_confluent_avro(col("value"), valueRegistryConfig) as 'value)

ERROR 2020-02-26 01:01:25,156 7247 org.apache.spark.executor.Executor [Executor task launch worker for task 0] Exception in task 0.0 in stage 0.0 (TID 0)
java.lang.NoSuchMethodError: org.apache.avro.Schema.createUnion([Lorg/apache/avro/Schema;)Lorg/apache/avro/Schema;
at org.apache.spark.sql.avro.SchemaConverters$.toAvroType(SchemaConverters.scala:185)
at za.co.absa.abris.avro.sql.SchemaProvider$$anonfun$apply$2.apply(SchemaProvider.scala:91)
at za.co.absa.abris.avro.sql.SchemaProvider$$anonfun$apply$2.apply(SchemaProvider.scala:87)
at za.co.absa.abris.avro.sql.SchemaProvider.lazyLoadSchemas(SchemaProvider.scala:41)
at za.co.absa.abris.avro.sql.SchemaProvider.originalSchema(SchemaProvider.scala:53)
at za.co.absa.abris.avro.sql.CatalystDataToAvro.serializer$lzycompute(CatalystDataToAvro.scala:41)
at za.co.absa.abris.avro.sql.CatalystDataToAvro.serializer(CatalystDataToAvro.scala:40)
at za.co.absa.abris.avro.sql.CatalystDataToAvro.nullSafeEval(CatalystDataToAvro.scala:44)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:410)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:256)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:836)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)

[cerveada]

Hello,
which version of Spark are you using? And which version of Abris?

[abhisam]

I am using spark 2.4 and abris 3.1.1

[cerveada]

There is no such Abris version. Do you mean 3.0.3?

It looks like there is a wrong version of Avro library on your claspath. Could you check the version? It should be 1.8.0 or higher.