Блог

В моем задании Azure Stream Analytics я хочу создать окно для каждого устройства (при получении данных для устройства). Я использую окно сеанса для поддержания отдельного окна для каждого устройства. Я хочу получать предупреждение о нарушении температурного порога каждые 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)
  

InputJSON
Я отправляю данные устройства с интервалом в 1 минуту.

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

Вывод

Для устройства — 9327848923ABE я получил первое предупреждение через 10 минут, после этого все предупреждения через 5 минут.

Azure Stream Analytics — это способ управления окнами сеанса:

Окна сеанса группируют события, которые поступают в одинаковое время, отфильтровывая периоды времени, когда нет данных. Функция окна сеанса имеет три основных параметра: тайм-аут, максимальную продолжительность и ключ разделения (необязательно).

Следующая диаграмма иллюстрирует поток с серией событий и то, как они отображаются в окнах сеанса с таймаутом 5 минут и максимальной продолжительностью 10 минут.

Окно сеанса начинается при возникновении первого события. Если в течение указанного времени ожидания с момента последнего принятого события происходит другое событие, окно расширяется, чтобы включить новое событие. В противном случае, если в течение времени ожидания не происходит никаких событий, окно закрывается по истечении времени ожидания.

Если события продолжают происходить в течение указанного времени ожидания, окно сеанса будет продолжать расширяться до достижения максимальной продолжительности. Пожалуйста, обратите внимание, что интервалы проверки максимальной продолжительности установлены такими же, как и указанная максимальная продолжительность. Например, если максимальная продолжительность равна 10, то проверки на превышение максимальной продолжительности окна будут выполняться при t = 0, 10, 20, 30 и т.д.

Таким образом, математически наше окно сеанса завершается, если выполняется следующее условие:

Время ожидания окна сеанса Stream Analytics составляет 5 минут, максимум 10 минут

При предоставлении ключа раздела события группируются по ключу, и окно сеанса применяется к каждой группе независимо. Это полезно в случаях, когда вам нужны разные окна сеанса для разных пользователей или устройств.

Вот синтаксис:

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

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

Объяснение:

время ожидания

Большое целое число, описывающее размер промежутка в окне сеанса. Данные, которые встречаются в пределах размера разрыва, группируются в одном окне.

maxdurationsize

Если общий размер окна превышает указанный maxDurationSize в контрольной точке, то окно закрывается и в той же точке открывается новое окно. В настоящее время размер интервала проверки равен 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 минут и ключ разделения
идентификатор пользователя. Это означает, что будут созданы независимые окна сеанса для
каждый идентификатор пользователя. Для каждого окна этот запрос будет генерировать выходные данные, которые
содержит идентификатор пользователя, время запуска окна (window_start),
окончание окна (window_end) и общая продолжительность работы пользователя
сеанс (длительность_в_секундах).

Это очень просто, просто начните с самого начала, если вы застряли.