Распространенные шаблоны запросов в Azure Stream Analytics и Fabric Eventstream

Замечание

Компонент Eventstream Fabric основан на той же среде выполнения, что и Azure Stream Analytics. Поэтому основные понятия, описанные в этой статье, применимы как к Azure Stream Analytics, так и к потоку событий Fabric.

Запросы в Azure Stream Analytics выражаются на языке запросов, например SQL. Конструкции языка описаны в справочном руководстве по языку запросов Stream Analytics .

Конструктор запросов может выразить простую сквозную логику для перемещения данных событий из одного входного потока в выходное хранилище данных или может выполнять насыщенное сопоставление шаблонов и темпоральный анализ для вычисления статистических выражений за различные временные окна, как в руководстве по созданию IoT решения с помощью Stream Analytics. Вы можете присоединить данные из нескольких входных данных для объединения событий потоковой передачи и выполнять поиск по статическим эталонным данным для обогащения значений событий. Вы также можете записывать данные в несколько выходных данных.

В этой статье описываются решения для нескольких распространенных шаблонов запросов на основе реальных сценариев.

Поддерживаемые форматы данных

Azure Stream Analytics поддерживает обработку событий в форматах данных CSV, JSON и Avro. Форматы JSON и Avro могут содержать сложные типы, такие как вложенные объекты (записи) или массивы. Дополнительные сведения о работе с этими сложными типами данных см. в разделе Анализ данных JSON и AVRO.

Отправка данных на несколько выходов

Несколько инструкций SELECT можно использовать для вывода данных в разные приемники выходных данных. Например, одна инструкция SELECT может выводить оповещение на основе порогового значения, а другая может выводить события в хранилище блобов.

Рассмотрим следующие входные данные:

| Make | Time |
| --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |
| Make1 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:01.0000000Z |
| Make2 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:03.0000000Z |

И вам нужны следующие два выходных данных из запроса:

ArchiveOutput:

| Make | Time |
| --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |
| Make1 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:01.0000000Z |
| Make2 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:03.0000000Z |

AlertOutput:

| Make | Time | Count |
| --- | --- | --- |
| Make2 |2023-01-01T00:00:10.0000000Z |3 |

Запрос с двумя операторами SELECT с выходными данными архива и выходными данными оповещений в качестве выходных данных:

SELECT
	*
INTO
	ArchiveOutput
FROM
	Input TIMESTAMP BY Time

SELECT
	Make,
	System.TimeStamp() AS Time,
	COUNT(*) AS [Count]
INTO
	AlertOutput
FROM
	Input TIMESTAMP BY Time
GROUP BY
	Make,
	TumblingWindow(second, 10)
HAVING
	[Count] >= 3

Предложение INTO сообщает службе Stream Analytics, в какой из выходов направлять данные. Первый SELECT определяет сквозной запрос, который получает данные из входных данных и отправляет его в выходные данные с именем ArchiveOutput. Второй запрос агрегирует и фильтрует данные перед отправкой результатов в нижестоящий вывод системы оповещений с именем AlertOutput.

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

Запрос:

WITH ReaderQuery AS (
	SELECT
		*
	FROM
		Input TIMESTAMP BY Time
)

SELECT * INTO ArchiveOutput FROM ReaderQuery

SELECT 
	Make,
	System.TimeStamp() AS Time,
	COUNT(*) AS [Count] 
INTO AlertOutput 
FROM ReaderQuery
GROUP BY
	Make,
	TumblingWindow(second, 10)
HAVING [Count] >= 3

Дополнительные сведения см. в предложении WITH.

Простой сквозной запрос

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

Рассмотрим следующие входные данные:

| Make | Time | Weight |
| --- | --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |"1000" |
| Make1 |2023-01-01T00:00:02.0000000Z |"2000" |

Вы хотите, чтобы выходные данные были одинаковыми для входных данных:

| Make | Time | Weight |
| --- | --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |"1000" |
| Make1 |2023-01-01T00:00:02.0000000Z |"2000" |

Ниже приведен запрос:

SELECT
	*
INTO Output
FROM Input

Этот запрос SELECT * проектирует все поля входящего события и отправляет их в выходные данные. Вместо этого можно проецировать только необходимые поля в инструкции SELECT. В следующем примере оператор SELECT проецирует только поля Make и Time из входных данных.

Рассмотрим следующие входные данные:

| Make | Time | Weight |
| --- | --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |1000 |
| Make1 |2023-01-01T00:00:02.0000000Z |2000 |
| Make2 |2023-01-01T00:00:04.0000000Z |1500 |

Выходные данные должны иметь только поля Make and Time:

| Make | Time |
| --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |
| Make1 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:04.0000000Z |

Вот запрос, который выводит только необходимые поля.

SELECT
	Make, Time
INTO Output
FROM Input

Сопоставление строк с LIKE и NOT LIKE

LIKE и NOT LIKE можно использовать для проверки соответствия поля определенному шаблону. Например, можно использовать фильтр, чтобы вернуть только номерные знаки, начинающиеся с буквы A и заканчивающиеся номером 9.

Рассмотрим следующие входные данные:

| Make | License_plate | Time |
| --- | --- | --- |
| Make1 |ABC-123 |2023-01-01T00:00:01.0000000Z |
| Make2 |AAA-999 |2023-01-01T00:00:02.0000000Z |
| Make3 |ABC-369 |2023-01-01T00:00:03.0000000Z |

Вы хотите, чтобы выходные данные имели номерные знаки, начинающиеся с буквы A и заканчивающиеся номером 9:

| Make | License_plate | Time |
| --- | --- | --- |
| Make2 |AAA-999 |2023-01-01T00:00:02.0000000Z |
| Make3 |ABC-369 |2023-01-01T00:00:03.0000000Z |

Ниже приведен запрос , использующий оператор LIKE:

SELECT
	*
FROM
	Input TIMESTAMP BY Time
WHERE
	License_plate LIKE 'A%9'

Используйте инструкцию LIKE , чтобы проверить значение поля License_plate . Он должен начинаться с буквы A, а затем иметь любую строку от нуля или более символов, заканчивая числом 9.

Вычисление за прошлые события

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

Пример входных данных:

| Make | Time |
| --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |
| Make2 |2023-01-01T00:00:02.0000000Z |

Пример выходных данных:

| Make | Time |
| --- | --- |
| Make2 |2023-01-01T00:00:02.0000000Z |

Пример запроса:

SELECT
	Make,
	Time
FROM
	Input TIMESTAMP BY Time
WHERE
	LAG(Make, 1) OVER (LIMIT DURATION(minute, 1)) <> Make

Используйте LAG для просмотра входного потока одного события назад, получения значения Make и сравнения его со значением Make текущего события и вывода события.

Дополнительные сведения см. в разделе LAG.

Возврат последнего события в окне

Так как события используются системой в режиме реального времени, функция не может определить, является ли событие последним для этого периода времени. Для этого входной поток должен быть присоединен к другому, где время события — максимальное время для всех событий в этом окне.

Пример входных данных:

| License_plate | Make | Time |
| --- | --- | --- |
| DXE 5291 |Make1 |2023-07-27T00:00:05.0000000Z |
| YZK 5704 |Make3 |2023-07-27T00:02:17.0000000Z |
| RMV 8282 |Make1 |2023-07-27T00:05:01.0000000Z |
| YHN 6970 |Make2 |2023-07-27T00:06:00.0000000Z |
| VFE 1616 |Make2 |2023-07-27T00:09:31.0000000Z |
| QYF 9358 |Make1 |2023-07-27T00:12:02.0000000Z |
| MDR 6128 |Make4 |2023-07-27T00:13:45.0000000Z |

Пример выходных данных с информацией о последних автомобилях в двух десятиминутных временных окнах:

| License_plate | Make | Time |
| --- | --- | --- |
| VFE 1616 |Make2 |2023-07-27T00:09:31.0000000Z |
| MDR 6128 |Make4 |2023-07-27T00:13:45.0000000Z |

Пример запроса:

WITH LastInWindow AS
(
	SELECT 
		MAX(Time) AS LastEventTime
	FROM 
		Input TIMESTAMP BY Time
	GROUP BY 
		TumblingWindow(minute, 10)
)

SELECT 
	Input.License_plate,
	Input.Make,
	Input.Time
FROM
	Input TIMESTAMP BY Time 
	INNER JOIN LastInWindow
	ON DATEDIFF(minute, Input, LastInWindow) BETWEEN 0 AND 10
	AND Input.Time = LastInWindow.LastEventTime

Первый шаг запроса находит максимальную метку времени в 10-минутных окнах, то есть метку времени последнего события для этого окна. Второй шаг объединяет результаты первого запроса с исходным потоком, чтобы найти событие, соответствующее последним меткам времени в каждом окне.

DATEDIFF — это функция, которая сравнивает и возвращает разницу времени между двумя полями DateTime, дополнительные сведения см. в функциях дат.

Дополнительные сведения о присоединении потоков см. в статье JOIN.

Агрегирование данных с течением времени

Чтобы вычислить информацию за период времени, можно агрегировать данные. В этом примере инструкция вычисляет количество за последние 10 секунд времени для каждого конкретного вида автомобиля.

Пример входных данных:

| Make | Time | Weight |
| --- | --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |1000 |
| Make1 |2023-01-01T00:00:02.0000000Z |2000 |
| Make2 |2023-01-01T00:00:04.0000000Z |1500 |

Пример выходных данных:

| Make | Count |
| --- | --- |
| Make1 | 2 |
| Make2 | 1 |

Запрос:

SELECT
	Make,
	COUNT(*) AS Count
FROM
	Input TIMESTAMP BY Time
GROUP BY
	Make,
	TumblingWindow(second, 10)

Эта агрегатная функция группирует автомобили по марке и подсчитывает их каждые 10 секунд. Выходные данные содержат марку и количество автомобилей, которые прошли через пункт оплаты.

TumblingWindow — это функция окна, используемая для группировки событий вместе. Агрегирование можно применить ко всем сгруппированным событиям. Дополнительные сведения см. в разделе "Функции окна".

Дополнительные сведения об агрегации см. в разделе агрегатных функций.

Периодически выводимые значения

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

Пример входных данных:

| Time | Value |
| --- | --- |
| "2014-01-01T06:01:00" |1 |
| "2014-01-01T06:01:05" |2 |
| "2014-01-01T06:01:10" |3 |
| "2014-01-01T06:01:15" |4 |
| "2014-01-01T06:01:30" |5 |
| "2014-01-01T06:01:35" |6 |

Пример выходных данных (первые 10 строк):

| Window_end | Last_event.Time | Last_event.Value |
| --- | --- | --- |
| 2014-01-01T14:01:00.000Z |2014-01-01T14:01:00.000Z |1 |
| 2014-01-01T14:01:05.000Z |2014-01-01T14:01:05.000Z |2 |
| 2014-01-01T14:01:10.000Z |2014-01-01T14:01:10.000Z |3 |
| 2014-01-01T14:01:15.000Z |2014-01-01T14:01:15.000Z |4 |
| 2014-01-01T14:01:20.000Z |2014-01-01T14:01:15.000Z |4 |
| 2014-01-01T14:01:25.000Z |2014-01-01T14:01:15.000Z |4 |
| 2014-01-01T14:01:30.000Z |2014-01-01T14:01:30.000Z |5 |
| 2014-01-01T14:01:35.000Z |2014-01-01T14:01:35.000Z |6 |
| 2014-01-01T14:01:40.000Z |2014-01-01T14:01:35.000Z |6 |
| 2014-01-01T14:01:45.000Z |2014-01-01T14:01:35.000Z |6 |

Пример запроса:

SELECT
	System.Timestamp() AS Window_end,
	TopOne() OVER (ORDER BY Time DESC) AS Last_event
FROM
	Input TIMESTAMP BY Time
GROUP BY
	HOPPINGWINDOW(second, 300, 5)

Этот запрос создает события каждые 5 секунд и выводит последнее событие, полученное ранее. Длительность HOPPINGWINDOW определяет, насколько далеко назад запрос ищет последнее событие.

Для получения дополнительной информации см. окно перескока.

Сопоставление событий в потоке

Сопоставление событий в одном потоке можно сделать, просмотрев прошлые события с помощью функции LAG . Например, результат можно создавать каждый раз, когда два последовательных автомобиля из одного и того же Make проходят через этот пункт взимания платы за последние 90 секунд.

Пример входных данных:

| Make | License_plate | Time |
| --- | --- | --- |
| Make1 |ABC-123 |2023-01-01T00:00:01.0000000Z |
| Make1 |AAA-999 |2023-01-01T00:00:02.0000000Z |
| Make2 |DEF-987 |2023-01-01T00:00:03.0000000Z |
| Make1 |GHI-345 |2023-01-01T00:00:04.0000000Z |

Пример выходных данных:

| Make | Time | Current_car_license_plate | First_car_license_plate | First_car_time |
| --- | --- | --- | --- | --- |
| Make1 |2023-01-01T00:00:02.0000000Z |AAA-999 |ABC-123 |2023-01-01T00:00:01.0000000Z |

Пример запроса:

SELECT
	Make,
	Time,
	License_plate AS Current_car_license_plate,
	LAG(License_plate, 1) OVER (LIMIT DURATION(second, 90)) AS First_car_license_plate,
	LAG(Time, 1) OVER (LIMIT DURATION(second, 90)) AS First_car_time
FROM
	Input TIMESTAMP BY Time
WHERE
	LAG(Make, 1) OVER (LIMIT DURATION(second, 90)) = Make

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

Дополнительные сведения см. в разделе LAG.

Определение длительности между событиями

Длительность события можно вычислить, просмотрев последнее событие "Start" после получения события "End". Этот запрос может быть полезен для определения времени, которое пользователь тратит на страницу или функцию.

Пример входных данных:

| User | Feature | Event | Time |
| --- | --- | --- | --- |
| user@location.com |RightMenu |Start |2023-01-01T00:00:01.0000000Z |
| user@location.com |RightMenu |End |2023-01-01T00:00:08.0000000Z |

Пример выходных данных:

| User | Feature | Duration |
| --- | --- | --- |
| user@location.com |RightMenu |7 |

Пример запроса:

SELECT
	[user],
	feature,
	DATEDIFF(
		second,
		LAST(Time) OVER (PARTITION BY [user], feature LIMIT DURATION(hour, 1) WHEN Event = 'start'),
		Time) as duration
FROM input TIMESTAMP BY Time
WHERE
	Event = 'end'

Функцию LAST можно использовать для получения последнего события в определенном условии. В этом примере условие — это событие типа Start, секционирование поиска по пользователю и компоненту PARTITION BY . Таким образом, каждый пользователь и функция обрабатываются независимо при поиске события Start. LIMIT DURATION ограничивает обратный поиск до 1 часа между событиями Start и End.

Подсчет уникальных значений

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

Пример входных данных:

| Make | Time |
| --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |
| Make1 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:01.0000000Z |
| Make2 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:03.0000000Z |

Пример выходных данных:

| Count_make | Time |
| --- | --- |
| 2 |2023-01-01T00:00:02.000Z |
| 1 |2023-01-01T00:00:04.000Z |

Пример запроса:

SELECT
     COUNT(DISTINCT Make) AS Count_make,
     System.TIMESTAMP() AS Time
FROM Input TIMESTAMP BY TIME
GROUP BY 
     TumblingWindow(second, 2)

COUNT(DISTINCT Make) возвращает количество уникальных значений в столбце Make в течение периода времени. Дополнительные сведения см. в разделе "Агрегатная функция COUNT".

Извлечение первого события в окне

Вы можете использовать IsFirst для получения первого события в окне времени. Например, вывод первых данных автомобиля каждые 10 минут.

Пример входных данных:

| License_plate | Make | Time |
| --- | --- | --- |
| DXE 5291 |Make1 |2023-07-27T00:00:05.0000000Z |
| YZK 5704 |Make3 |2023-07-27T00:02:17.0000000Z |
| RMV 8282 |Make1 |2023-07-27T00:05:01.0000000Z |
| YHN 6970 |Make2 |2023-07-27T00:06:00.0000000Z |
| VFE 1616 |Make2 |2023-07-27T00:09:31.0000000Z |
| QYF 9358 |Make1 |2023-07-27T00:12:02.0000000Z |
| MDR 6128 |Make4 |2023-07-27T00:13:45.0000000Z |

Пример выходных данных:

| License_plate | Make | Time |
| --- | --- | --- |
| DXE 5291 |Make1 |2023-07-27T00:00:05.0000000Z |
| QYF 9358 |Make1 |2023-07-27T00:12:02.0000000Z |

Пример запроса:

SELECT 
	License_plate,
	Make,
	Time
FROM 
	Input TIMESTAMP BY Time
WHERE 
	IsFirst(minute, 10) = 1

IsFirst также может секционировать данные и вычислять первое событие для каждого конкретного автомобиля Make , найденного каждые 10 минут.

Пример выходных данных:

| License_plate | Make | Time |
| --- | --- | --- |
| DXE 5291 |Make1 |2023-07-27T00:00:05.0000000Z |
| YZK 5704 |Make3 |2023-07-27T00:02:17.0000000Z |
| YHN 6970 |Make2 |2023-07-27T00:06:00.0000000Z |
| QYF 9358 |Make1 |2023-07-27T00:12:02.0000000Z |
| MDR 6128 |Make4 |2023-07-27T00:13:45.0000000Z |

Пример запроса:

SELECT 
	License_plate,
	Make,
	Time
FROM 
	Input TIMESTAMP BY Time
WHERE 
	IsFirst(minute, 10) OVER (PARTITION BY Make) = 1

Дополнительные сведения см. в разделе IsFirst.

Удаление повторяющихся событий в окне

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

Пример входных данных:

| DeviceId | Time | Attribute | Value |
| --- | --- | --- | --- |
| 1 |2018-07-27T00:00:01.0000000Z |Temperature |50 |
| 1 |2018-07-27T00:00:01.0000000Z |Temperature |50 |
| 2 |2018-07-27T00:00:01.0000000Z |Temperature |40 |
| 1 |2018-07-27T00:00:05.0000000Z |Temperature |60 |
| 2 |2018-07-27T00:00:05.0000000Z |Temperature |50 |
| 1 |2018-07-27T00:00:10.0000000Z |Temperature |100 |

Пример выходных данных:

| AverageValue | DeviceId |
| --- | --- |
| 70 | 1 |
|45 | 2 |

Пример запроса:

WITH Temp AS (
	SELECT Value, DeviceId
	FROM Input TIMESTAMP BY Time
	GROUP BY Value, DeviceId, System.Timestamp()
)
 

SELECT
	AVG(Value) AS AverageValue, DeviceId
INTO Output
FROM Temp
GROUP BY DeviceId,TumblingWindow(minute, 5)

При выполнении первой инструкции повторяющиеся записи объединяются в одну, так как поля в операторе GROUP BY совпадают. Таким образом, он удаляет дубликаты.

Укажите логику для разных случаев и значений (оператор CASE)

Операторы CASE могут обеспечивать различные вычисления для разных полей на основе определенного критерия. Например, назначьте полосу A для автомобилей Make1, а полосу B для любой другой марки.

Пример входных данных:

| Make | Time |
| --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |
| Make2 |2023-01-01T00:00:02.0000000Z |
| Make2 |2023-01-01T00:00:03.0000000Z |

Пример выходных данных:

| Make |Dispatch_to_lane | Time |
| --- | --- | --- |
| Make1 |"A" |2023-01-01T00:00:01.0000000Z |
| Make2 |"B" |2023-01-01T00:00:02.0000000Z |

Пример запроса:

SELECT
	Make
	CASE
		WHEN Make = "Make1" THEN "A"
		ELSE "B"
	END AS Dispatch_to_lane,
	System.TimeStamp() AS Time
FROM
	Input TIMESTAMP BY Time

Выражение CASE сравнивает выражение с набором простых выражений для определения результата. В этом примере транспортные средства марки Make1 отправляются в полосу A, в то время как транспортные средства любой другой марки будут назначены по полосе B.

Дополнительные сведения см. в выражении case.

Преобразование данных

Данные можно преобразовать в режиме реального времени с помощью метода CAST. Например, вес автомобиля можно преобразовать из типа nvarchar(max) в тип bigint и использовать в числовых вычислениях.

Пример входных данных:

| Make | Time | Weight |
| --- | --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |"1000" |
| Make1 |2023-01-01T00:00:02.0000000Z |"2000" |

Пример выходных данных:

| Make | Weight |
| --- | --- |
| Make1 |3000 |

Пример запроса:

SELECT
	Make,
	SUM(CAST(Weight AS BIGINT)) AS Weight
FROM
	Input TIMESTAMP BY Time
GROUP BY
	Make,
	TumblingWindow(second, 10)

Используйте инструкцию CAST , чтобы указать его тип данных. См. список поддерживаемых типов данных в типах данных (Azure Stream Analytics).

Дополнительные сведения о функциях преобразования данных.

Определение/Измерение длительности состояния

Для условий, охватывающих несколько событий, функцию LAG можно использовать для определения длительности этого условия. Например, предположим, что ошибка вызвала некорректный вес у всех автомобилей (более 20 000 фунтов), и необходимо вычислить длительность этой ошибки.

Пример входных данных:

| Make | Time | Weight |
| --- | --- | --- |
| Make1 |2023-01-01T00:00:01.0000000Z |2000 |
| Make2 |2023-01-01T00:00:02.0000000Z |25000 |
| Make1 |2023-01-01T00:00:03.0000000Z |26000 |
| Make2 |2023-01-01T00:00:04.0000000Z |25000 |
| Make1 |2023-01-01T00:00:05.0000000Z |26000 |
| Make2 |2023-01-01T00:00:06.0000000Z |25000 |
| Make1 |2023-01-01T00:00:07.0000000Z |26000 |
| Make2 |2023-01-01T00:00:08.0000000Z |2000 |

Пример выходных данных:

| Start_fault | End_fault |
| --- | --- |
| 2023-01-01T00:00:02.000Z |2023-01-01T00:00:07.000Z |

Пример запроса:

WITH SelectPreviousEvent AS
(
SELECT
	*,
	LAG([time]) OVER (LIMIT DURATION(hour, 24)) as previous_time,
	LAG([weight]) OVER (LIMIT DURATION(hour, 24)) as previous_weight
FROM input TIMESTAMP BY [time]
)

SELECT 
	LAG(time) OVER (LIMIT DURATION(hour, 24) WHEN previous_weight < 20000 ) [Start_fault],
	previous_time [End_fault]
FROM SelectPreviousEvent
WHERE
	[weight] < 20000
	AND previous_weight > 20000

Первая инструкция SELECT сопоставляет текущее измерение веса с предыдущим измерением, проецируя его вместе с текущим измерением. Второй SELECT возвращается к последнему событию, в котором previous_weight меньше 20000, где текущий вес меньше 20000, и previous_weight текущего события был больше 20000.

End_fault является текущим неисправным событием, где предыдущее событие было неисправным, а Start_fault является последним безошибочным событием до него.

Обработка событий с независимым временем (Подпотоки)

События могут поступать поздно или вне порядка из-за отклонений часов между источниками событий, отклонений часов между разделами или задержки в сети. Например, часы устройства для TollID 2 отстают на пять секунд от TollID 1, а часы устройства для TollID 3 отстают на 10 секунд от TollID 1. Вычисления могут выполняться независимо для каждой платы, используя только время по собственным часам в качестве метки времени.

Пример входных данных:

| LicensePlate | Make | Time | TollID |
| --- | --- | --- | --- |
| DXE 5291 |Make1 |2023-07-27T00:00:01.0000000Z | 1 |
| YHN 6970 |Make2 |2023-07-27T00:00:05.0000000Z | 1 |
| QYF 9358 |Make1 |2023-07-27T00:00:01.0000000Z | 2 |
| GXF 9462 |Make3 |2023-07-27T00:00:04.0000000Z | 2 |
| VFE 1616 |Make2 |2023-07-27T00:00:10.0000000Z | 1 |
| RMV 8282 |Make1 |2023-07-27T00:00:03.0000000Z | 3 |
| MDR 6128 |Make3 |2023-07-27T00:00:11.0000000Z | 2 |
| YZK 5704 |Make4 |2023-07-27T00:00:07.0000000Z | 3 |

Пример выходных данных:

| TollID | Count |
| --- | --- |
| 1 | 2 |
| 2 | 2 |
| 1 | 1 |
| 3 | 1 |
| 2 | 1 |
| 3 | 1 |

Пример запроса:

SELECT
      TollId,
      COUNT(*) AS Count
FROM input
      TIMESTAMP BY Time OVER TollId
GROUP BY TUMBLINGWINDOW(second, 5), TollId

Предложение TIMESTAMP OVER BY рассматривает каждую временную шкалу устройства независимо, используя подпотоки. Выходные события для каждого объекта TollID создаются по мере их вычисления, то есть события упорядочены относительно каждого значения TollID , а не были переупорядочены, как если бы все устройства работали по одному и тому же времени.

Дополнительные сведения см. в разделе TIMESTAMP BY OVER.

Сеансовые окна

Окно сеанса — это окно, которое продолжает расширяться по мере возникновения событий и закрывается для вычислений, если событие не получено после определенного периода времени или если окно достигает максимальной длительности. Это окно особенно полезно при вычислении данных взаимодействия с пользователем. Окно начинается, когда пользователь начинает взаимодействовать с системой и закрывается, когда не наблюдается больше событий, то есть пользователь перестал взаимодействовать. Например, пользователь взаимодействует с веб-страницей, в которой регистрируется количество щелчков, можно использовать окно сеанса, чтобы узнать, сколько времени пользователь взаимодействовал с сайтом.

Пример входных данных:

| User_id | Time | URL |
| --- | --- | --- |
| 0 | 2017-01-26T00:00:00.0000000Z | "www.example.com/a.html" |
| 0 | 2017-01-26T00:00:20.0000000Z | "www.example.com/b.html" |
| 1 | 2017-01-26T00:00:55.0000000Z | "www.example.com/c.html" |
| 0 | 2017-01-26T00:01:10.0000000Z | "www.example.com/d.html" |
| 1 | 2017-01-26T00:01:15.0000000Z | "www.example.com/e.html" |

Пример выходных данных:

| User_id | StartTime | EndTime | Duration_in_seconds |
| --- | --- | --- | --- |
| 0 | 2017-01-26T00:00:00.0000000Z | 2017-01-26T00:01:10.0000000Z | 70 |
| 1 | 2017-01-26T00:00:55.0000000Z | 2017-01-26T00:01:15.0000000Z | 20 |

Пример запроса:

SELECT
	user_id,
	MIN(time) as StartTime,
	MAX(time) as EndTime,
	DATEDIFF(second, MIN(time), MAX(time)) AS duration_in_seconds
FROM input TIMESTAMP BY time
GROUP BY
	user_id,
	SessionWindow(minute, 1, 60) OVER (PARTITION BY user_id)

SELECT отображает данные, относящиеся к взаимодействию с пользователем, включая длительность взаимодействия. Группирование данных по пользователю и SessionWindow, которое закрывается, если взаимодействие не происходит в течение 1 минуты, с максимальным размером окна в 60 минут.

Дополнительные сведения о SessionWindow см. в окне сеанса .

Определяемые пользователем функции в JavaScript и C#

Замечание

Этот раздел не относится к потоку событий Fabric.

Язык запросов Azure Stream Analytics можно расширить с помощью пользовательских функций, написанных на языке JavaScript или C#. Определяемые пользователем функции (UDF) — это пользовательские и сложные вычисления, которые нельзя легко выразить с помощью языка SQL . Эти UDF (определяемые пользователем функции) определяются один раз и используются несколько раз в запросе. Например, UDF можно использовать для преобразования значения nvarchar(max) из шестнадцатеричного в значение bigint.

Пример входных данных:

| Device_id | HexValue |
| --- | --- |
| 1 | "B4" |
| 2 | "11B" |
| 3 | "121" |

Пример выходных данных:

| Device_id | Decimal |
| --- | --- |
| 1 | 180 |
| 2 | 283 |
| 3 | 289 |

function hex2Int(hexValue){
	return parseInt(hexValue, 16);
}

public static class MyUdfClass {
	public static long Hex2Int(string hexValue){
		return int.Parse(hexValue, System.Globalization.NumberStyles.HexNumber);
	}
}

SELECT
	Device_id,
	udf.Hex2Int(HexValue) AS Decimal
From
	Input

Функция User-Defined вычисляет значение bigint из HexValue для каждого потребляемого события.

Дополнительные сведения см. в статье JavaScript и C#.

Расширенное сопоставление шаблонов с MATCH_RECOGNIZE

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

Входные данные:

| ATM_id | Operation_id | Return_Code | Time |
| --- | --- | --- | --- |
| 1 | "Entering Pin" | "Success" | 2017-01-26T00:10:00.0000000Z |
| 2 | "Opening Money Slot" | "Success" | 2017-01-26T00:10:07.0000000Z |
| 2 | "Closing Money Slot" | "Success" | 2017-01-26T00:10:11.0000000Z |
| 1 | "Entering Withdraw Quantity" | "Success" | 2017-01-26T00:10:08.0000000Z |
| 1 | "Opening Money Slot" | "Warning" | 2017-01-26T00:10:14.0000000Z |
| 1 | "Printing Bank Balance" | "Warning" | 2017-01-26T00:10:19.0000000Z |

Выходные данные:

| ATM_id | First_Warning_Operation_id | Warning_Time |
| --- | --- | --- |
| 1 | "Opening Money Slot" | 2017-01-26T00:10:14.0000000Z |

SELECT *
FROM input TIMESTAMP BY time OVER ATM_id
MATCH_RECOGNIZE (
	LIMIT DURATION(minute, 1)
	PARTITION BY ATM_id
	MEASURES
		First(Warning.ATM_id) AS ATM_id,
		First(Warning.Operation_Id) AS First_Warning_Operation_id,
		First(Warning.Time) AS Warning_Time
	AFTER MATCH SKIP TO NEXT ROW
	PATTERN (Success+ Warning{2,})
	DEFINE
		Success AS Success.Return_Code = 'Success',
		Warning AS Warning.Return_Code <> 'Success'
) AS patternMatch

Этот запрос соответствует как минимум двум последовательным событиям сбоя и создает сигнал тревоги, когда условия соблюдаются. ШАБЛОН определяет регулярное выражение, которое будет использоваться для сопоставления, в данном случае по крайней мере два последовательных предупреждения после хотя бы одной успешной операции. Состояние "Успех" и "Предупреждение" определяются с помощью значения Return_Code, как только выполняется условие. МЕРЫ** прогнозируются с ATM_id, первой операцией оповещения и первым временем предупреждения.

Дополнительные сведения см. в MATCH_RECOGNIZE.

Геофенсинг и геопространственные запросы

Azure Stream Analytics предоставляет встроенные геопространственные функции, которые можно использовать для реализации таких сценариев, как управление парком, общий доступ к поездкам, подключенные автомобили и отслеживание активов. Геопространственные данные можно получать в форматах GeoJSON или WKT в рамках потока событий или ссылочных данных. Например, компания, которая специализируется на производстве машин для печати паспортов, арендует свои машины правительствам и консульствам. Расположение этих машин строго контролируется, чтобы избежать их перемещения и возможного использования для подделки паспортов. Каждое устройство оснащено GPS-трекером, и информация передается обратно в задание Azure Stream Analytics. Производство хотело бы следить за расположением этих машин и быть оповещено, если один из них покидает авторизованную область, таким образом они могут удаленно отключать, оповещать власти и извлекать оборудование.

Входные данные:

| Equipment_id | Equipment_current_location | Time |
| --- | --- | --- |
| 1 | "POINT(-122.13288797982818 47.64082002051315)" | 2017-01-26T00:10:00.0000000Z |
| 1 | "POINT(-122.13307252987875 47.64081350934929)" | 2017-01-26T00:11:00.0000000Z |
| 1 | "POINT(-122.13308862313283 47.6406508603241)" | 2017-01-26T00:12:00.0000000Z |
| 1 | "POINT(-122.13341048821462 47.64043760861279)" | 2017-01-26T00:13:00.0000000Z |

Ввод ссылочных данных:

| Equipment_id | Equipment_lease_location |
| --- | --- |
| 1 | "POLYGON((-122.13326028450979 47.6409833866794,-122.13261655434621 47.6409833866794,-122.13261655434621 47.64061471602751,-122.13326028450979 47.64061471602751,-122.13326028450979 47.6409833866794))" |

Выходные данные:

| Equipment_id | Equipment_alert_location | Time |
| --- | --- | --- |
| 1 | "POINT(-122.13341048821462 47.64043760861279)" | 2017-01-26T00:13:00.0000000Z |

SELECT
	input.Equipment_id AS Equipment_id,
	input.Equipment_current_location AS Equipment_current_location,
	input.Time AS Time
FROM input TIMESTAMP BY time
JOIN
	referenceInput 
	ON input.Equipment_id = referenceInput.Equipment_id
	WHERE 
		ST_WITHIN(input.Equipment_current_location, referenceInput.Equipment_lease_location) = 1

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

Для получения дополнительной информации см. статью Сценарии геозонирования и геопространственной агрегации с помощью Azure Stream Analytics.

Получите помощь

За дополнительной информацией перейдите на страницу вопросов и ответов об Azure Stream Analytics.

Дальнейшие действия

• Введение в Azure Stream Analytics
• Начать использовать Azure Stream Analytics
• Масштабирование заданий в службе Azure Stream Analytics
• Справочник по языку запросов Azure Stream Analytics
• Справочник по управлению REST API Azure Stream Analytics