流分析通过事件中心从 Python 反序列化 JSON

我已经设置了一个 Azure 事件中心，我正在从 Python 脚本以 JSON 格式发送 AMQP 消息，并尝试使用流分析将这些消息流式传输到 Power BI。消息来自和物联网设备的非常简单的设备活动

Python 代码段是

msg = json.dumps({ "Hub": MAC, "DeviceID": id, "DeviceUID": ouid, "Signal": text, "Timestamp": dtz }, ensure_ascii=False, encoding='utf8')
message.body = msg
messenger.put(message)
messenger.send()

我在 MS 教程中使用了示例 C# 消息阅读器从事件中心读回数据，没有问题，输出为：

Message received.  Partition: '2', Data: '??{"DeviceUID": "z_70b3d515200002e7_0", "Signal": "/on?1", "DeviceID": "1", "Hub": "91754623489", "Timestamp": "2016-07-15T07:56:50.277440Z"}'

但是，当我尝试从事件中心测试流分析输入时，出现错误

诊断：无法将输入事件反序列化为 Json。一些可能的原因：1）格式错误的事件 2）输入源配置了不正确的序列化格式

我不确定格式错误的事件是什么意思 - 我假设流分析可以处理通过 AMQP 发送到事件中心的数据？

我看不出 C# 应用程序收到的 JSON 有什么问题 - 除非 BOM 符号导致问题？

这是我第一次尝试这一切，我已经搜索了任何类似的帖子，但无济于事，所以如果有人能指出我正确的方向，我将不胜感激。

干杯抢

这是

由客户端 API 不兼容引起的。Python 使用 Proton 在 AMQP 值消息的正文中发送 JSON 字符串。正文编码为 AMQP 字符串（AMQP 类型编码字节 + utf8 编码字节的字符串）。流分析使用服务总线 .Net SDK，该 SDK 将 AMQP 消息公开为事件数据，其正文始终为字节数组。对于 AMQP 值消息，它包括 AMQP 类型的编码字节，因为没有它们，就无法解码以下值。开头的这些额外字节将导致 JSON 序列化失败。

若要在消息正文上实现互操作性，应用程序应确保发布者和使用者就其类型和编码达成一致。在这种情况下，发布者应在 AMQP 数据消息中发送原始字节。使用 Proton Python API，您可以尝试以下操作：

message.body = msg.encode('utf-8')

另一种解决方法是在应用程序属性中发送简单类型（例如字符串）。

其他人也遇到了这个问题。https://github.com/Azure/amqpnetlite/issues/117

正如@XinChen所说，该问题是由AMQP协议引起的。

根据我的经验，以下两种解决方法对于这种情况是有效的。

使用Send Event REST API而不是带有AMQP的Azure Python SDK，但其余API基于HTTP协议，性能不高。
使用 Base64 编码发送 JSON 消息，然后将收到的消息解码为 JSON 字符串。

这两

件事对我有用：

添加message.inferred = True
检查以确保指定的转储编码encoding='utf-8'不像示例中那样encoding='utf8'。

更新的OP：

msg = json.dumps({ "Hub": MAC, "DeviceID": id, "DeviceUID": ouid, "Signal": text, "Timestamp": dtz }, ensure_ascii=False, encoding='utf-8')
message.body = msg
message.inferred = True
messenger.put(message)
messenger.send()

通过添加推断的标志，我认为消息序列化程序可以正确推断正文是字节并创建 AMPQ DATA，从而解决陈@Xin的观点。

消息
的推断标志指示如何将消息内容编码到 AMQP 部分中。如果推断为 true，则消息正文中的二进制值和列表值将分别编码为 AMQP DATA 和 AMQP 序列部分。如果推断为 false，则消息正文中的所有值都将编码为 AMQP VALUE 部分，无论其类型如何。

回复：Qpid 质子文档 #inferred

re： JSON 编码器和解码器 #dumps

相关内容

最新更新

热门标签：