Блог

У меня возникли некоторые проблемы с декодированием сообщения Avro из Kafka в Python с использованием kafka-python. Чтобы свести это к минимуму, я сосредоточусь на простом декодировании сообщения с использованием пакета avro. Я написал тест со схемой и примером из официальных документов avro: https://avro.apache.org/docs/current/gettingstartedpython.html .

repl.it

 from avro.io import DatumWriter, DatumReader, BinaryEncoder, BinaryDecoder
import avro.schema
from io import BytesIO

schema = avro.schema.parse("""
    {
        "type": "record",
        "name": "User",
        "namespace": "example.avro",
        "fields": [
            {
                "name": "name",
                "type": "string"
            },
            {
                "name": "favorite_number",
                "type": [
                    "int",
                    "null"
                ]
            },
            {
                "name": "favorite_color",
                "type": [
                    "string",
                    "null"
                ]
            }
        ]
    }
""")

wb = BytesIO()
encoder = BinaryEncoder(wb)
writer = DatumWriter(schema)
writer.write('{"name":"Alyssa","favorite_number":256,"favorite_color":"blue"}', encoder)

rb = BytesIO(wb.getvalue())
decoder = BinaryDecoder(rb)
reader = DatumReader(schema)
msg = reader.read(decoder)

print(msg)
  

Я получаю сообщение об ошибке, которое the datum {"name":"Alyssa","favorite_number":256,"favorite_color":"blue"} is not an example of the schema . Что я делаю не так, учитывая, что эта схема и база данных взяты прямо из официальных документов Avro для Python?

 Traceback (most recent call last):
  File "main.py", line 36, in <module>
    writer.write('{"name":"Alyssa","favorite_number":256,"favorite_color":"blue"}', encoder)
  File "/opt/virtualenvs/python3/lib/python3.8/site-packages/avro/io.py", line 979, in write
    raise AvroTypeException(self.writers_schema, datum)
avro.io.AvroTypeException: The datum {"name":"Alyssa","favorite_number":256,"favorite_color":"blue"} is not an example of the schema {
  "type": "record",
  "name": "User",
  "namespace": "example.avro",
  "fields": [
    {
      "type": "string",
      "name": "name"
    },
    {
      "type": [
        "int",
        "null"
      ],
      "name": "favorite_number"
    },
    {
      "type": [
        "string",
        "null"
      ],
      "name": "favorite_color"
    }
  ]
}
  

В настоящее время у вас есть

 writer.write('{"name":"Alyssa","favorite_number":256,"favorite_color":"blue"}', encoder)
  

Итак, предоставляемая вами база данных представляет собой строку. Если вы измените его на словарь, подобный этому:

 writer.write({"name":"Alyssa","favorite_number":256,"favorite_color":"blue"}, encoder)
  

Тогда это работает.