AbsaOSS / AbsaOSS/ABRiS

Unable to deserialize the records via Abris Pyspark

Abierto
#366 0 comentarios 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Scala
Estrellas
242
Forks
84
Merge medio
10 h 12 min
PR fusionados (30 d)
2

Descripción

I am trying to run this code where I am trying to install the packages during runtime

`from pyspark import SparkContext
from pyspark.sql.column import Column, _to_java_column
from pyspark.sql.session import SparkSession

spark = SparkSession \
.builder \
.appName("Kafka_Test") \
.config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0,org.apache.spark:spark-avro_2.12:3.3.0,za.co.absa:abris_2.13:6.4.0") \
.getOrCreate()

def from_avro(col, config):
"""
avro deserialize

:param col (PySpark column / str): column name "key" or "value"
:param config (za.co.absa.abris.config.FromAvroConfig): abris config, generated from abris_config helper function
:return: PySpark Column
"""
jvm_gateway = SparkContext._active_spark_context._gateway.jvm
abris_avro = jvm_gateway.za.co.absa.abris.avro

return Column(abris_avro.functions.from_avro(_to_java_column(col), config))

def from_avro_abris_config(config_map, topic, is_key):
"""
Create from avro abris config with a schema url

:param config_map (dict[str, str]): configuration map to pass to deserializer, ex: {'schema.registry.url': 'http://localhost:8081'}
:param topic (str): kafka topic
:param is_key (bool): boolean
:return: za.co.absa.abris.config.FromAvroConfig
"""
jvm_gateway = SparkContext._active_spark_context._gateway.jvm
scala_map = jvm_gateway.PythonUtils.toScalaMap(config_map)

return jvm_gateway.za.co.absa.abris.config \
.AbrisConfig \
.fromConfluentAvro() \
.downloadReaderSchemaByLatestVersion() \
.andTopicNameStrategy(topic, is_key) \
.usingSchemaRegistry(scala_map)

def to_avro(col, config):
"""
avro serialize
:param col (PySpark column / str): column name "key" or "value"
:param config (za.co.absa.abris.config.ToAvroConfig): abris config, generated from abris_config helper function
:return: PySpark Column
"""
jvm_gateway = SparkContext._active_spark_context._gateway.jvm
abris_avro = jvm_gateway.za.co.absa.abris.avro

return Column(abris_avro.functions.to_avro(_to_java_column(col), config))

def to_avro_abris_config(config_map, topic, is_key):
"""
Create to avro abris config with a schema url

:param config_map (dict[str, str]): configuration map to pass to the serializer, ex: {'schema.registry.url': 'http://localhost:8081'}
:param topic (str): kafka topic
:param is_key (bool): boolean
:return: za.co.absa.abris.config.ToAvroConfig
"""
jvm_gateway = SparkContext._active_spark_context._gateway.jvm
scala_map = jvm_gateway.PythonUtils.toScalaMap(config_map)

return jvm_gateway.za.co.absa.abris.config \
.AbrisConfig \
.toConfluentAvro() \
.downloadSchemaByLatestVersion() \
.andTopicNameStrategy(topic, is_key) \
.usingSchemaRegistry(scala_map)

df = spark.read.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", "test01").load()

from_avro_abris_settings = from_avro_abris_config({'schema.registry.url': 'http://schema-registry:8081'}, 'test01', False)
df2 = df.withColumn("parsed", from_avro("value", from_avro_abris_settings))
df2.show()`

But it is giving me the below error:

`:: problems summary ::
:::: WARNINGS
module not found: io.confluent#kafka-avro-serializer;6.2.1

==== local-m2-cache: tried

file:/Users/sid/.m2/repository/io/confluent/kafka-avro-serializer/6.2.1/kafka-avro-serializer-6.2.1.pom

-- artifact io.confluent#kafka-avro-serializer;6.2.1!kafka-avro-serializer.jar:

file:/Users/sid/.m2/repository/io/confluent/kafka-avro-serializer/6.2.1/kafka-avro-serializer-6.2.1.jar

==== local-ivy-cache: tried

/Users/sid/.ivy2/local/io.confluent/kafka-avro-serializer/6.2.1/ivys/ivy.xml

-- artifact io.confluent#kafka-avro-serializer;6.2.1!kafka-avro-serializer.jar:

/Users/sid/.ivy2/local/io.confluent/kafka-avro-serializer/6.2.1/jars/kafka-avro-serializer.jar

==== central: tried

https://repo1.maven.org/maven2/io/confluent/kafka-avro-serializer/6.2.1/kafka-avro-serializer-6.2.1.pom

-- artifact io.confluent#kafka-avro-serializer;6.2.1!kafka-avro-serializer.jar:

https://repo1.maven.org/maven2/io/confluent/kafka-avro-serializer/6.2.1/kafka-avro-serializer-6.2.1.jar

==== spark-packages: tried

https://repos.spark-packages.org/io/confluent/kafka-avro-serializer/6.2.1/kafka-avro-serializer-6.2.1.pom

-- artifact io.confluent#kafka-avro-serializer;6.2.1!kafka-avro-serializer.jar:

https://repos.spark-packages.org/io/confluent/kafka-avro-serializer/6.2.1/kafka-avro-serializer-6.2.1.jar

module not found: io.confluent#kafka-schema-registry-client;6.2.1

==== local-m2-cache: tried

file:/Users/sid/.m2/repository/io/confluent/kafka-schema-registry-client/6.2.1/kafka-schema-registry-client-6.2.1.pom

-- artifact io.confluent#kafka-schema-registry-client;6.2.1!kafka-schema-registry-client.jar:

file:/Users/sid/.m2/repository/io/confluent/kafka-schema-registry-client/6.2.1/kafka-schema-registry-client-6.2.1.jar

==== local-ivy-cache: tried

/Users/sid/.ivy2/local/io.confluent/kafka-schema-registry-client/6.2.1/ivys/ivy.xml

-- artifact io.confluent#kafka-schema-registry-client;6.2.1!kafka-schema-registry-client.jar:

/Users/sid/.ivy2/local/io.confluent/kafka-schema-registry-client/6.2.1/jars/kafka-schema-registry-client.jar

==== central: tried

https://repo1.maven.org/maven2/io/confluent/kafka-schema-registry-client/6.2.1/kafka-schema-registry-client-6.2.1.pom

-- artifact io.confluent#kafka-schema-registry-client;6.2.1!kafka-schema-registry-client.jar:

https://repo1.maven.org/maven2/io/confluent/kafka-schema-registry-client/6.2.1/kafka-schema-registry-client-6.2.1.jar

==== spark-packages: tried

https://repos.spark-packages.org/io/confluent/kafka-schema-registry-client/6.2.1/kafka-schema-registry-client-6.2.1.pom

-- artifact io.confluent#kafka-schema-registry-client;6.2.1!kafka-schema-registry-client.jar:

https://repos.spark-packages.org/io/confluent/kafka-schema-registry-client/6.2.1/kafka-schema-registry-client-6.2.1.jar

::::::::::::::::::::::::::::::::::::::::::::::

:: UNRESOLVED DEPENDENCIES ::

::::::::::::::::::::::::::::::::::::::::::::::

:: io.confluent#kafka-avro-serializer;6.2.1: not found

:: io.confluent#kafka-schema-registry-client;6.2.1: not found

::::::::::::::::::::::::::::::::::::::::::::::

:: USE VERBOSE OR DEBUG MESSAGE LEVEL FOR MORE DETAILS
Exception in thread "main" java.lang.RuntimeException: [unresolved dependency: io.confluent#kafka-avro-serializer;6.2.1: not found, unresolved dependency: io.confluent#kafka-schema-registry-client;6.2.1: not found]
at org.apache.spark.deploy.SparkSubmitUtils$.resolveMavenCoordinates(SparkSubmit.scala:1456)
` Please help me in this.

Guía de contribución

No hay ninguna guía de contribución indexada para este repositorio

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.