代码之家 › 专栏 › 技术社区 › Pankaj Rawat

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

azure-stream-analytics azure

0

Pankaj Rawat · 技术社区 · 7 年前

在我的Azure流分析作业中,我想为每个设备创建一个窗口(当接收到设备的数据时)。我使用会话窗口为每个设备维护一个单独的窗口。我想每5分钟发出一次温度阈值突破警报,但有时行为不同。以下是我的ASA工作查询

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)

输入JSON

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

输出

0 回复 | 直到 7 年前

1

0

B''H Bi'ezras -- Boruch Hashem 7 年前

这个 Azure Stream Analytics 是管理会话窗口的方法:

会话窗口组在相似时间到达的事件,筛选没有数据的时间段。会话窗口函数有三个主要参数:超时、最大持续时间和分区

下图演示了包含一系列事件和如何将它们映射到5分钟超时的会话窗口,以及最长持续时间为10分钟。

在上次摄取事件的指定超时内发生,然后窗口将扩展以包含新事件。否则,如果没有事件

如果事件在指定的超时时间内继续发生,则会话窗口将一直延长,直到达到最大持续时间。拜托与指定的最大持续时间大小相同。例如,如果持续时间是10,然后检查窗口是否超过最大值

因此,从数学上讲,如果

流分析会话窗口5分钟超时,最多10分钟

当提供分区键时,事件按键和会话窗口将独立应用于每个组。这对于需要不同会话窗口的情况非常有用不同的用户或设备。

语法 :

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

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

:

描述会话窗口间隔大小的大整数。数据出现在间隙大小内的,在相同的

最大持续时间大小

检查点,然后关闭窗口并打开新窗口在同一点上。当前,检查间隔的大小为等于maxDurationSize。

一个可选参数,指定会话窗口手术结束了。如果指定,则窗口将仅组合在一起

实例 : 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分钟,分区键为每个用户id。对于每个窗口,此查询将生成窗口结束(窗口结束)和用户的总持续时间会话(持续时间以秒为单位)。