scala和Python之间的Avro Kafka转换问题

我们的项目同时具有scala和python代码，我们需要向kafka发送/使用avro编码的消息。scala和Python之间的Avro Kafka转换问题

我使用python和scala发送avro编码消息给kafka。我在Scala代码制片其发送使用Twitter一一对应库的Avro编码的消息如下：

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成功解码在斯卡拉

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) 
    } 
) 

}

但是，我无法解码messag在蟒蛇上课我得到以下异常

'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之间保持一致？任何指针将是伟大的

来源

2017-02-11 Abhishek

找出解决方案。很快就会发布解决方案。它可以帮助其他人。 – Abhishek

我在Python中使用二进制编码器，并没有在斯卡拉。刚刚从

val recordInjection = GenericAvroCodecs[GenericRecord](schema)

改变一个行

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

我希望其他人发现它是有用的。 python代码不需要更改

来源

2017-02-11 04:44:15 Abhishek

scala和Python之间的Avro Kafka转换问题

回答

相关问题