scala - Spark将原始字节解组为可读格式

Question

我的 Spark 应用程序从 Kafka 接收二进制数据。数据作为字节原始消息发送到 Kafka。原型消息是：

message Any {
  string type_url=1;
  bytes value=2;
}

使用 ScalaPB 库，我可以将 Any 消息反序列化为其原始格式。如何将值从字节反序列化为可读格式？SerializationUtils 不起作用。这就是 Any 消息在反序列化后的样子。

#+-----------------------------------------|
#| type_url           | value              |
#+-----------------------------------------|
#|type.googleapis.c...|[0A 8D D8 04 0A 1...|
#+-----------------------------------------|

该值仍为其字节格式。使用 SerializationUtils 对其进行反序列化后，数据不正确。

#+-----------------+
#|value            |
#+-----------------+
#|2020-09-04T10:...|
#+-----------------+

还有其他选择吗？有没有办法将字节反序列化为字符串、结构或字符串 Json？

我使用带有 udf 的 ScalaPBs 示例将字节反序列化为 Any 消息。

val parseCloud = ProtoSQL.udf { bytes: Array[Byte] => CloudEvent.parseFrom(bytes) }

带有 SerializationUtils 字节值的 udf 如下所示。

val parseBytes = ProtoSQL.udf {bytes: Array[Byte] => deserialize(bytes)}

标签： scalaapache-sparkapache-kafkaprotocol-buffersscalapb

Answer 1

如果您知道 中的消息类型，则Any可以使用 unpack 方法进行反序列化。

val unpackAny = ProtoSQL.udf { any: com.google.protobuf.any.Any => any.unpack[MyMessage] }