流分析到事件中心 - 意外连接事件答案

【问题标题】：Stream Analytics to Event Hub - Unexpectedly concatenating events流分析到事件中心 - 意外连接事件
【发布时间】：2017-07-01 23:47:20
【问题描述】：

我有一个流分析作业，它使用 avro 消息的事件中心（我们称之为 RawEvents），转换/展平消息并将它们触发到单独的事件中心（我们称之为 FormattedEvents）。

RawEvents 中的每个 EventData 实例都由一个顶级 json 对象组成，该对象具有更详细的事件数组。这是一个人为的例子：

[{ "事件": [{ "dataOne": 123.0, "dataTwo": 234.0, “subEventCode”：3，“dateTimeLocal”：1482170771，“dateTimeUTC”： 1482192371 }，{“dataOne”：456.0，“dataTwo”：789.0， “subEventCode”：20，“dateTimeLocal”：1482170771，“dateTimeUTC”： 1482192371 }]，“messageType”：“myDeviceType-Events”，“deviceID”： “我的设备”，}]

流分析作业将结果展平并解压缩 subEventCode，这是一个位掩码。结果如下所示：

{"messagetype":"myDeviceType-Event","deviceid":"myDevice",eventid:1,"dataone":123,"datatwo":234,"subeventcode":6,"flag1":0, "flag2":1,"flag3":1,"flag4":0,"flag5":0,"flag6":0,"flag7":0,"flag8":0,"flag9":0,"flag10 ":0,"flag11":0,"flag12":0,"flag13":0,"flag14":0,"flag15":0,"flag16":0,"eventepochlocal":"2016-12-06T17 :33:11.0000000Z","eventepochutc":"2016-12-06T23:33:11.0000000Z"} {"messagetype":"myDeviceType-Event","deviceid":"myDevice",eventid:2,"dataone" :456,"datatwo":789,"subeventcode":8,"flag1":0,"flag2":0,"flag3":0,"flag4":1,"flag5":0,"flag6":0 ,"flag7":0,"flag8":0,"flag9":0,"flag10":0,"flag11":0,"flag12":0,"flag13":0,"flag14":0," flag15":0,"flag16":0,"eventepochlocal":"2016-12-06T17:33:11.0000000Z","eventepochutc":"2016-12-06T23:33:11.0000000Z"}

当我从 FormattedEvents 事件中心提取消息时，我希望看到两个 EventData 实例。我得到的是一个 EventData，在同一条消息中包含两个“扁平化”事件。这是针对 blob 存储或数据湖时的预期行为，但在针对事件中心时令人惊讶。我的期望是类似于服务总线的行为。

这是预期的行为吗？如果是这样，是否有强制行为的配置选项？

【问题讨论】：

标签： azure azure-eventhub azure-stream-analytics

【解决方案1】：

是的，这是目前的预期行为。目的是提高吞吐量，尝试在 EventHub 消息（EventData）中发送尽可能多的事件。

不幸的是，截至今天，没有配置选项可以覆盖此行为。一种可能值得尝试的方法是将输出分区键的概念用于超级独特的东西（即将此列添加到您的查询中 - GetMetadataPropertyValue(ehInput, "EventId") as outputpk ）。现在在输出 EventHub 的 ASA 设置中将“outputpk”指定为 PartitionKey。

如果有帮助，请告诉我。

干杯车坦

【讨论】：

如您所说，我们确认行为是 MS 设计的。如果我们想处理单个事件，我们要么必须以服务总线为目标，要么手动拆分主体。有点令人失望，但你去吧。

【解决方案2】：

我遇到了同样的问题。感谢您手动格式化输入消息的答案。我和我的同事用几行代码解决了这个问题，去掉了换行和回车。然后我将“}{”替换为“},{”，并通过在两端添加“[”和“]”使其成为一个数组。

string modifiedMessage = myEventHubMessage.Replace("\n","").Replace("\r","");    
modifiedMessage = "[" + modifiedMessage.Replace("}{","},{") + "]";

然后根据其数据结构将输入作为对象列表：

List<TelemetryDataPoint> newDataPoints = new List<TelemetryDataPoint>();
try
{
    newDataPoints = Newtonsoft.Json.JsonConvert.DeserializeObject<List<TelemetryDataPoint>>(modifiedMessage);

.... ....

【讨论】：