Вопросы преобразования Avro Kafka между scala и Python

Question

Наш проект имеет как scala, так и код python, и нам нужно отправлять/потреблять avro-кодированные сообщения в kafka.

Я отправляю avro кодирует сообщения в kafka, используя python и scala. У меня есть продюсер в Ла Скала код, который отправить Avro закодированные сообщения, используя библиотеку биекция Twitter следующим образом:

val resourcesPath = getClass.getResource("/avro/url_info_schema.avsc") 
val schemaFile = scala.io.Source.fromURL(resourcesPath).mkString 
val schema = parser.parse(schemaFile) 
val recordInjection = GenericAvroCodecs[GenericRecord](schema) 
val avroRecord = new GenericData.Record(schema) 
avroRecord.put("url_sha256", row._1) 
avroRecord.put("url", row._2._1) 
avroRecord.put("timestamp", row._2._2) 
val recordBytes = recordInjection.apply(avroRecord) 
kafkaProducer.value.send("topic", recordBytes) 

Avro схема выглядит

{ 
    "namespace": "com.rm.avro", 
    "type": "record", 
    "name": "url_info", 
    "fields":[ 
    { 
     "name": "url_sha256", "type": "string" 
    }, 
    { 
     "name": "url", "type": "string" 
    }, 
    { 
     "name": "timestamp", "type": ["long"] 
    } 
] 

}

Я могу расшифровать его успешно KafkaConsumer in scala

val resourcesPath = getClass.getResource("/avro/url_info_schema.avsc") 
val schemaFile = scala.io.Source.fromURL(resourcesPath).mkString 


kafkaInputStream.foreachRDD(kafkaRDD => { 
    kafkaRDD.foreach(

    avroRecord => { 
     val parser = new Schema.Parser() 
     val schema = parser.parse(schemaFile) 
     val recordInjection = GenericAvroCodecs[GenericRecord](schema) 
     val record = recordInjection.invert(avroRecord.value()).get 
     println(record) 
    } 
) 

} 

Однако я не могу декодировать сообщения эс в питоне я получаю следующее исключение

'utf8' codec can't decode byte 0xe4 in position 16: invalid continuation byte 

питона кодов выглядит следующим образом: schema_path = "Avro/url_info_schema.avsc" схема = avro.schema.parse (открытое (schema_path) .read())

for msg in consumer: 
    bytes_reader = io.BytesIO(msg.value) 
    decoder = avro.io.BinaryDecoder(bytes_reader) 
    reader = avro.io.DatumReader(schema) 
    decoded_msg = reader.read(decoder) 
    print(decoded_msg) 

Также сообщения производителя python avro не понятны потребителю scala avro. Там я получаю исключение. Производитель Python Avro выглядит следующим образом:

datum_writer = DatumWriter(schema) 
bytes_writer = io.BytesIO() 

datum_writer = avro.io.DatumWriter(schema) 
encoder = avro.io.BinaryEncoder(bytes_writer) 
datum_writer.write(data, encoder) 
raw_bytes = bytes_writer.getvalue() 
producer.send(topic, raw_bytes) 

Как я могу оставаться последовательным на python и scala? Любые указатели будут замечательными

Answer 1

Я использовал двоичный кодер в питоне и ничего в Scala. Просто пришлось изменить одну строку из

val recordInjection = GenericAvroCodecs[GenericRecord](schema) 

в

val recordInjection = GenericAvroCodecs.toBinary[GenericRecord](schema) 

Я надеюсь, что другие считают это полезным. Нет изменений, необходимых для кода python.