问题标签 [scalapb]

我正在使用 spark scala 中的 scalapb 解析通过 protobuf 序列化的 bahir mqtt 有效负载，但解析的 json 仅包含第一个 json 对象，其他对象为空。

• 火花版本：2.3.0
• 斯卡拉版本：2.11.8
• Protobuf 版本：2
• sparksql-scalapb 版本：0.8.0

原型文件

我得到的结果

但预期的结果是

更新 1

序列化之前的有效载荷传入消息如下所示：

MQTT 服务器正在使用 eclipse tahu 项目，该项目使用 protobuf 序列化有效负载。

更新 2

这是代码：

流数据的模式和样本

感谢您的意见。

我正在尝试创建一个使用 ProtoBuf 编码的 Kafka 消息的 Spark Streaming。

这是我最近几天尝试的

使用 scalapb，我设法解码二进制 proto 文件并转换为数据帧。但是对于流式传输，我在解析行的编译时得到了这个异常：

谁能给点提示？

我想使用 scalaPB 在 scala 中生成以下代数数据类型。

我在protobuf下面定义了myevent.proto-

我的定义的问题Flush是生成为case classnot case object。

另外，我不知道如何使字段可选并具有默认值。

我正在尝试解析 spark 2.4 中的 protobuf (protobuf3) 数据，但 ByteString 类型遇到了一些问题。我使用 ScalaPB 库创建了案例类，并将 jar 加载到 spark shell 中。我也尝试为该类型创建一个隐式编码器，但是我仍然收到以下错误；

java.lang.UnsupportedOperationException: No Encoder found for com.google.protobuf.ByteString

到目前为止，这是我尝试过的；

任何帮助表示赞赏

我正在尝试使用 ScalaPB 和 Maven 创建一个简单的 gRPC 服务。以下是我的目录结构：

我的 simlpe.proto 看起来像这样：

和 pom.xml ：

当我这样做时，mvn compile或者mvn scala:compile我可以看到在目标目录中生成了 protobuf 类，但与 gRPC 相关的任何类除外。

我错过了什么吗？

我正在使用 google protobuf 库在文件中写入消息。但是，这些消息对我来说并不像二进制消息。我使用的原型很简单，它有姓名、电子邮件和手机号码。我得到的文件是：

如果我回读，我可以读取它，但这看起来不是二进制数据，我想要的是将二进制写入文件，然后在读取时将其解析回来。我浏览了 API，发现我们有多种可用于parseFrom()的方法，例如：

在谷歌 protobuf 教程中，他们使用的是parseFrom(InputStream)。对于二进制消息，我认为我需要parseFrom(Byte[])。但我不知道如何编写二进制消息。帮助表示赞赏。我的最终目标是使用 scalapb 读取 spark 数据帧中的二进制消息。

我有一个持久的actor，它可以接收一种类型的命令Persist(event)，其中事件是一种类型trait Event（它有很多实现）。成功后，这会回复Persisted(event)发件人。

事件本身是可序列化的，因为这是我们存储在持久性存储中的数据，并且序列化是使用自定义序列化程序实现的，该序列化程序在内部使用从 google protobuf.proto文件生成的类。并且这个自定义序列化器被配置application.conf并绑定到 base trait Event。这已经很好了。

注意：的实现Event不是protobuf 生成的类。它们是普通的 scala 类，它们也有一个等价的 protobuf，但它是通过绑定到基本 Event 类型的自定义序列化程序映射的。这是我的前辈为版本控制完成的（这可能不是必需的，因为这也可以使用普通的 protobuf 类 + 自定义序列化组合来处理，但这是另一回事），我不想更改那个 atm。

我们现在正在尝试为这个参与者实现集群分片，这也意味着我的命令（即Persist和Persisted）也需要可序列化，因为它们可能会被转发到其他节点。

这是域模型：

问题是，我没有看到一种使其可序列化的优雅方法。以下是我考虑过的选项

方法 1.Persist为和定义一个新的 proto 文件Persisted，但是我使用什么作为数据类型event呢？我没有找到一种方法来定义这样的东西：

这样我就可以使用现有的 Scala 特征Event作为数据类型。如果这可行，我想（虽然它很牵强）我可以将生成的代码（在编译这个 proto 文件之后）绑定到 akka 的谷歌 protobuf 的内置序列化程序，它可能会工作。上面的注释解释了为什么我不能oneof在我的 proto 文件中使用构造。

方法 2。这是我已经实现的并且有效（但我不喜欢它）

基本上，我为命令编写了一个新的序列化程序，并将部分命令的序列化和反序列化委托event给现有的序列化程序。

这种方法的问题是我def manifest在def fromBinary. 在序列化和反序列化时，我必须确保我拥有命令的清单以及事件的清单。因此，我不得不将其~用作分隔符——类似于我对清单信息的自定义序列化技术。

是否有更好的或正确的方法来实现这一点？

对于上下文：我使用 ScalaPB 从.proto文件和经典的 akka 演员生成 scala 类。

非常感谢任何形式的指导！

在服务实现的方法可以访问元数据的情况下，我可以为 ScalaPB 代码生成设置什么选项？

默认选项生成如下内容：

我想要类似的东西：

--- 2021 年 12 月 30 日更新 ---

一个原因是我可能想从元数据中检索客户端的令牌，因此我可以使用该令牌来调用另一个服务。

我已经定义了扩展以下的scala对象（比如MyObject）

当我在对象上调用 parseFrom 方法时，出现以下错误：

引起：java.lang.NoSuchMethodError:....MyObject$.parseFrom([B)Lscalapb/GeneratedMessage;

我尝试了 scalapb-runtime_2.11 和 scalapb-runtime_2.12。

编辑：问题已解决。这是依赖不匹配的情况。

使用https://scalapb.github.io/generated-code.htmlPerson中的和Address消息定义

构建设置是：

然后我创建了一个测试：

测试#1 通过。TEST #2 编译失败

错误：(98, 40) value optionalStreet is not a member of scalapb.lenses.Lens[trexo.scalapb.myprotos.Address,trexo.scalapb.myprotos.Address] val emptyAddr = bogusAddr.update(_.optionalStreet := None)

问题：在 ScalaPB 文档中，可选字段据说有一种更新可选字段的方法（为方便起见，在下面转载）。_.optionalX 更新方式是否已更改为不同的语法？