如何使用Python解碼/反序列化Kafka Avro字符串

我正在從遠程服務器接收Kafka Avro郵件（使用Confluent Kafka Python庫的使用者），它使用帶有字段的用戶代理，位置表示點擊流數據，url等。這裏是一條消息的樣子：如何使用Python解碼/反序列化Kafka Avro字符串

b'\x01\x00\x00\xde\x9e\xa8\xd5\x8fW\xec\x9a\xa8\xd5\x8fW\x1axxx.xxx.xxx.xxx\x02:https://website.in/rooms/\x02Hhttps://website.in/wellness-spa/\x02\xaa\x14\x02\x9c\n\x02\xaa\x14\x02\xd0\x0b\x02V0:j3lcu1if:rTftGozmxSPo96dz1kGH2hvd0CREXmf2\x02V0:j3lj1xt7:YD4daqNRv_Vsea4wuFErpDaWeHu4tW7e\x02\x08null\x02\nnull0\x10pageview\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\x10Thailand\x02\xa6\x80\xc4\x01\x02\x0eBangkok\x02\x8c\xba\xc4\x01\x020*\xa9\x13\xd0\[email protected]\x02\xec\xc09#J\[email protected]\x02\x8a\x02Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/58.0.3029.96 Chrome/58.0.3029.96 Safari/537.36\x02\x10Chromium\x02\x10Chromium\x028Google Inc. and contributors\x02\x0eBrowser\x02\x1858.0.3029.96\x02"Personal computer\x02\nLinux\x02\x00\x02\x1cCanonical Ltd.'

如何解碼？我試過bson解碼，但字符串不被識別爲UTF-8，因爲它是我猜測的特定Avro編碼。我發現https://github.com/verisign/python-confluent-schemaregistry但它只支持Python 2.7。理想情況下，我希望與Python 3.5+和MongoDB一起處理數據並將其存儲爲當前的基礎結構。

來源

2017-06-07 Alex Pa

我以爲Avro庫只是爲了讀取Avro文件，但它實際上解決了解碼Kafka消息的問題，如下所示：我首先導入庫並將模式文件作爲參數，然後創建一個函數來解碼消息轉換成字典，我可以在消費者循環中使用它。

from confluent_kafka import Consumer, KafkaError 
from avro.io import DatumReader, BinaryDecoder 
import avro.schema 

schema = avro.schema.Parse(open("data_sources/EventRecord.avsc").read()) 
reader = DatumReader(schema) 

def decode(msg_value): 
    message_bytes = io.BytesIO(msg_value) 
    decoder = BinaryDecoder(message_bytes) 
    event_dict = reader.read(decoder) 
    return event_dict 

c = Consumer() 
c.subscribe(topic) 
running = True 
while running: 
    msg = c.poll() 
    if not msg.error(): 
     msg_value = msg.value() 
     event_dict = decode(msg_value) 
     print(event_dict) 
    elif msg.error().code() != KafkaError._PARTITION_EOF: 
     print(msg.error()) 
     running = False

來源

2017-06-12 11:40:53

如何使用Python解碼/反序列化Kafka Avro字符串

回答

相關問題