Azure流分析会话窗口异常行为

Question

在我的Azure流分析作业中，我想为每个设备创建一个窗口（当收到设备的数据时）。我正在使用会话窗口为每个设备维护一个单独的窗口。我希望每5分钟发出一次温度阈值突破警报，但有时候行为会有所不同。下面是我的ASA JOB查询

SELECT
t1.DeviceId,MIN(t1.Temperature)MinTemperature,     
MAX(t1.Temperature)MaxTemperature,t2.ThresholdLimit,System.Timestamp as 
EventProcessedUtcTime,t3.EventProcessedUtcTime as LastAlertTime
,MIN(CAST(t1.MessageTime AS DateTime)) AS Window_start,
System.Timestamp AS Window_end,Count(t1.DeviceId)EventCount
INTO
[alertOutputsb]
FROM
[tsfInput] t1
INNER JOIN [device-threashold-Input] t2
ON t1.DeviceId = t2.DeviceId
LEFT OUTER JOIN [temperature-alerts-sql-Input] t3
ON t1.DeviceId = t3.DeviceId   
WHERE t1.Temperature >= Cast(t2.ThresholdLimit as float)      
GROUP BY t1.DeviceId,t2.ThresholdLimit,t3.EventProcessedUtcTime, 
SESSIONWINDOW(minute, 4, 5) OVER (PARTITION BY t1.DeviceId)
HAVING MIN(t1.Temperature) >= Cast(t2.ThresholdLimit as float)
AND MAX(t1.Temperature) >= Cast(t2.ThresholdLimit as float)

InputJSON我每1分钟发送一次设备数据。

{
"DeviceId": "9327848923ABE",
"Temperature": 300.5,
"MessageTime": "2019-03-02T05:50:05.362Z"
}

输出

对于设备 - 9327848923ABE我在10分钟内收到第一个警报，之后在5分钟窗口内发出所有警报。

Answer 1

Azure Stream Analytics是管理会话窗口的方法：

会话窗口将在相似时间到达的事件分组，过滤掉没有数据的时间段。会话窗口功能有三个主要参数：超时，最大持续时间和分区键（可选）。

下图说明了包含一系列事件的流以及它们如何映射到5分钟超时和最长10分钟的会话窗口。

会话窗口在第一个事件发生时开始。如果在上次摄取事件的指定超时内发生另一个事件，则窗口将扩展为包含新事件。否则，如果超时内没有事件发生，则窗口在超时时关闭。

如果事件在指定的超时内持续发生，则会话窗口将持续延长，直到达到最大持续时间。请注意，最大持续时间检查间隔设置为与指定的最大持续时间相同。例如，如果最大持续时间为10，则检查窗口是否超过最大持续时间将在t = 0,10,20,30等处发生。

因此，在数学上，如果满足以下条件，则会话窗口结束：

流分析会话窗口超时5分钟，最长10分钟

当提供分区键时，通过键将事件分组在一起，并且会话窗口独立地应用于每个组。这对于您需要针对不同用户或设备的不同会话窗口的情况非常有用。

这是语法：

SESSIONWINDOW(timeunit, timeoutSize, maxDurationSize) [OVER (PARTITION BY partitionKey)]

SESSIONWINDOW(Timeout(timeunit , timeoutSize), MaxDuration(timeunit, maxDurationSize)) [OVER (PARTITION BY partitionKey)]

说明：

timeoutsize

一个大整数，描述会话窗口的间隙大小。间隙大小内发生的数据在同一窗口中组合在一起。

maxdurationsize

如果总窗口大小超过检查点处的指定maxDurationSize，则关闭窗口并在同一点打开新窗口。目前，检查间隔的大小等于maxDurationSize。

PartitionKey

一个可选参数，指定会话窗口操作的键。如果指定，则窗口将仅将相同键的事件组合在一起。

示例：JSON：

[
  // time: the timestamp when the user clicks on the link
  // user_id: the id of the user
  // url: the url the user clicked on
  {
    "time": "2017-01-26T00:00:00.0000000z",
    "user_id": 0,
    "url": "www.example.com/a.html"
  },
  {
    "time": "2017-01-26T00:00:20.0000000z",
    "user_id": 0,
    "url": "www.example.com/b.html"
  },
  {
    "time": "2017-01-26T00:00:55.0000000z",
    "user_id": 1,
    "url": "www.example.com/c.html"
  },
  // ...
]

要衡量每个用户会话的时长，可以使用以下查询：

CREATE TABLE localinput(time DATETIME, user_id BIGINT, url NVARCHAR(MAX))
SELECT
    user_id,
    MIN(time) AS window_start,
    System.Timestamp AS window_end,
    DATEDIFF(s, MIN(time), System.Timestamp) AS duration_in_seconds
FROM localinput TIMESTAMP BY time
GROUP BY user_id, SessionWindow(minute, 2, 60) OVER (PARTITION BY user_id)

上述查询创建的会话窗口的超时时间为2分钟，最长持续时间为60分钟，分区键为user_id。这意味着将为每个user_id创建独立的会话窗口。对于每个窗口，此查询将生成包含user_id，窗口的开始时间（window_start），窗口结束（window_end）和用户会话的总持续时间（duration_in_seconds）的输出。

这很简单，如果你遇到困难就从头开始。