Azure 流分析中的跳跃窗口

Posted 2023-05-08

【中文标题】Azure 流分析中的跳跃窗口

【英文标题】：Hopping Window in Azure Stream Analytics

【发布时间】：2020-10-20 06:51:15

【问题描述】：

我试图在 azure 流分析中了解跳跃窗口。 我将从 Azure 事件中心获取以下数据：

[
  
    "Id": "1",
    "SensorData": [
      
        "Timestamp": 1603112431,
        "Type": "LineCrossing",
        "Direction": "forward"
      ,
      
        "Timestamp": 1603112431,
        "Type": "LineCrossing",
        "Direction": "forward"
      
    ],
    "EventProcessedUtcTime": "2020-10-20T06:35:48.5890814Z",
    "PartitionId": 1,
    "EventEnqueuedUtcTime": "2020-10-20T06:35:48.3540000Z"
  ,
  
    "Id": "1",
    "SensorData": [
      
        "Timestamp": 1603112430,
        "Type": "LineCrossing",
        "Direction": "backward"
      
    ],
    "EventProcessedUtcTime": "2020-10-20T06:35:48.5890814Z",
    "PartitionId": 0,
    "EventEnqueuedUtcTime": "2020-10-20T06:35:48.2140000Z"
  
]

我的查询如下所示：

SELECT s.Id, COUNT(data.ArrayValue.Direction) as Count
FROM [customers] s TIMESTAMP BY EventEnqueuedUtcTime
CROSS APPLY GetArrayElements(s.SensorData) AS data
WHERE data.ArrayValue.Type = 'LineCrossing' 
AND data.ArrayValue.Direction = 'forward'
GROUP BY s.Id, HoppingWindow(second, 3600, 5)

我使用跳跃窗口每 5 秒获取最后一天的所有事件。 我对给定 dto 的期望是：一行 Id1 和计数 2，但我收到的是：720 行（因此 3600 除以 5），Id1 计数为 2。

这些事件不应该被 HoppingWindow 函数聚合吗？

【问题讨论】：

您是否使用本地输入进行测试？

我只是将一些事件推送到事件中心并在流分析平台中对这些数据进行采样

我认为问题出在本地输入，因为它会模拟整个窗口——这就是您看到 3600/5 事件的原因。下面我用云输入进行了回答和测试，修改了一些参数，它应该根据我对您的期望的理解工作:)

我不知道查询窗口正在模拟整个时间跨度。

我确认与本地测试（或从样本测试）和实时处理的不同之处在于，在测试中我们假设我们拥有从开始到结束的所有事件，并且 ASA 将返回每个窗口一次。除了来自 kgalic 的回答，我还可以建议添加窗口的时间戳，以便您可以看到它们随着时间的推移而一一生成： SELECT ... , System.Timestamp() as time_of_end_of_the_window

【参考方案1】：

我将您的查询构造如下：

with inputValues as (Select input.*, message.ArrayValue as Data from input CROSS APPLY GetArrayElements(input.SensorData) as message)

select inputValues.Id, count(Data.Direction) as Count
into output
from inputValues 
where Data.Type = 'LineCrossing' and Data.Direction='forward'
GROUP BY inputValues.Id, HoppingWindow(second, 3600, 5)

我已将输入设置为事件中心，并在 Visual Studio 中使用云输入开始查询。

我使用Windows Client application 将消息通过管道传输到事件中心（下图中的 2.），并观察到事件每 5 秒发生一次（下图中的 1. 和下图中的 3.）。

也许只是更改我共享的查询以反映正确的时间戳，但结果应该符合预期 - 对于过去一小时内到达的所有事件（3600 秒HoppingWindow 函数）。