如何从二进制原型中提取JSON？

Question

将Apache Spark 2.2.0结构化流视为：

jsonStream.printSchema()
root
 |-- body: binary (nullable = true)

正文中的数据类型为Protocol Buffers v2和嵌套的JSON。看起来像

syntax = "proto2";

message Data {
  required string data = 1;
}

message List {
  repeated Data entry = 1;
}

如何提取Spark中的数据以“进一步”处理它？

我查看了ScalaPB，但当我在Jupyter中运行代码时，无法将“.proto”代码内联。我也不知道如何将数据帧转换为流上的RDD。正在尝试。rdd因流媒体源而失败。

更新1：我发现了如何使用ScalaPB的控制台工具从protobuf规范生成Scala文件。由于“类型不匹配”，我仍然无法导入它们。




Answer 1

匿名用户

tl； dr编写一个用户定义函数（UDF）将二进制字段（带有JSON的协议）反序列化为JSON。

将序列化的正文（二进制格式）视为一个表列。暂时忘掉结构化流媒体（以及流媒体数据集）。

那么，让我把这个问题重新表述如下：

如何转换（又名铸）二进制值[这里你的格式]？

有些格式是直接cast-able的，这使得将二进制文件转换为字符串非常容易，如下所示：

$"body" cast "string"

如果字符串是JSON或unixtime，您可以使用内置的“转换器”，即from_json或from_unixtime等函数。

介绍应该给你一个提示，如何进行像你这样的转换。

body中的数据类型为Protocol Buffers v2和嵌套的JSON。

要处理此类字段（protobuf json），您必须编写一个Scala函数，将“有效负载”解码为json，并使用UDF创建一个用户定义函数（UDF）：

UDF（f： UDF1[_，_]， returType： DataType）：用户定义函数将JavaUDF1实例定义为用户定义函数（UDF）。调用者必须指定输出数据类型，并且没有自动输入类型强制。默认情况下，返回的UDF是确定性的。要将其更改为非确定性，请调用APIUserDefedFunction.as非确定性（）。

然后使用像from_json或get_json_object这样的函数。

为了简化您的案例，请编写一个单参数函数进行转换，并使用UDF函数将其包装成一个UDF。

由于流媒体源，尝试. rdd失败。

使用Dataset.foreach或foreach分区。

foreach（f：（T）⇒ Unit）：Unit将函数f应用于所有行。

foreach分区（f：（Iterator[T]）¶Unit）： Unit将函数f应用于此数据集的每个分区。