pyspark-treaming protobuf消息给出错误：attributeError：＆＃x27; bytearray＆＃x27;对象没有属性＆＃x27;

Question

我正在使用py spark流读来自kafka主题的Protobuf消息。
使用msg.thichoneof给出错误 attributeError：'bytearray'对象没有属性'whonof'。
架构protofile是：

message InstrumentStatusUpdate {
    ....
    ..
}
...
..
message MarketDataEvent {
    oneof event {
        InstrumentStatusUpdate instrumentStatusUpdate = 1;
        Trade trade = 2;
    }
};

我正在使用的py-spark代码：

def parse_options_monitor_msg(msg:schema_pb2.MarketDataEvent):
    eventStr = msg.WhichOneof("event")
    if eventStr == "trade":
        trade_list = []
        trade_msg = msg.trade
        symbol = trade_msg.symbol
        ticker = symbol.split('_')[0]
        trade_list = [symbol,ticker]
        return str(trade_list)

parse_options_monitor = udf(lambda x: parse_options_monitor_msg(x), StringType())

df = spark.readStream \
            .format("kafka") \
            .options(**kafka_conf) \
            .load()
data = df.selectExpr("offset", "value") \
         .withColumn("event", parse_options_monitor(col("value")))

df2 = data.select(col("offset"),col("event"))

df2.writeStream \
            .format("console") \
            .outputMode("append") \
            .option("truncate", False) \
            .start() \
            .awaitTermination()

如何摆脱此错误？

Answer 1

我添加了解析（）方法。

def parse_options_monitor_msg(msg_bytes:schema_pb2.MarketDataEvent):
    msg = schema_pb2.MarketDataEvent()
    msg.ParseFromString(msg_bytes)
    eventStr = msg.WhichOneof("event")
    if eventStr == "trade":
        trade_list = []
        trade_msg = msg.trade
        symbol = trade_msg.symbol
        ticker = symbol.split('_')[0]
        trade_list = [symbol,ticker]
        return str(trade_list)