Hi again, I am unable to deserialize Avro record. There is a similar question here with no answer 😕 https://stackoverflow.com/questions/74825256/failed-to-deserialize-avro-record-getting-arrayindexoutofboundsexception

deserialization_schema = AvroRowDeserializationSchema(
        avro_schema_string="""
            {
                "type": "record",
                "name": "TestObject",
                "namespace": "ca.dataedu",
                "fields": [{
                    "name": "count",
                    "type": ["null", "int"],
                    "default": null
                }, {
                    "name": "word",
                    "type": ["null", "string"],
                    "default": null
                }]
            }
            """
    )


kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers('0.0.0.0:9092') \
    .set_topics("test_topic") \
    .set_group_id("my-group") \
    .set_starting_offsets(KafkaOffsetsInitializer.latest()) \
    .set_value_only_deserializer(deserialization_schema) \
    .build()

ds = st_env.from_source(kafka_source, WatermarkStrategy.no_watermarks(), "Kafka Source")

Error traceback

Caused by: java.lang.ArrayIndexOutOfBoundsException: Index -62 out of bounds for length 2
        at org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:460)
        at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:283)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:188)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
        at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:260)
        at org.apache.avro.specific.SpecificDatumReader.readField(SpecificDatumReader.java:142)
        at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:248)
        at org.apache.avro.specific.SpecificDatumReader.readRecord(SpecificDatumReader.java:123)
        at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:180)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:161)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:154)
        at org.apache.flink.formats.avro.AvroRowDeserializationSchema.deserialize(AvroRowDeserializationSchema.java:164)
        ... 18 more

Python code on the producer side

from random_word import RandomWords
import json
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                         value_serializer=lambda m: json.dumps(m).encode('utf-8'))

r = RandomWords()

for i in range(10):
    producer.send('test_topic', {'word': r.get_random_word(), 'count' : i})

Copy code

value_serializer=lambda m: json.dumps(m).encode('utf-8')

It looks like you’re serializing it as JSON, but trying to deserialize as Avro

Thanks for the hint. I tried with JSONSchema and now I am getting AttributeError : encode

json_schema = """

{
  "$schema": "<http://json-schema.org/draft-04/schema#>",
  "type": "object",
  "properties": {
    "word": {
      "type": "string"
    },
    "count": {
      "type": "integer"
    }
  },
  "required": [
    "word",
    "count"
  ]
}

"""
deserialization_schema = JsonRowDeserializationSchema.Builder() \
        .json_schema(json_schema) \
        .build()

Error Traceback

File "/home/user/anaconda3/envs/flink/lib/python3.8/site-packages/pyflink/common/types.py", line 225, in __getattr__
    idx = self._fields.index(item)
ValueError: 'encode' is not in list

During handling of the above exception, another exception occurred:

This error implies when it tries to parse string but a list is given, have you checked the result of

r.get_random_word()

is not a list?

Thanks for your comment. Today I had a look at it again.

r.get_random_word()

returns a string but I have made a mistake in my code in ds.map after it. The issue has been resolved now.