LAG (Azure Stream Analytics)

Článek
11/15/2023

Analytický operátor LAG umožňuje vyhledat "předchozí" událost ve streamu událostí v rámci určitých omezení. Je velmi užitečný při výpočtu rychlosti růstu proměnné, při zjišťování, kdy proměnná překročí prahovou hodnotu nebo kdy podmínka začíná nebo přestane být pravdivá.

Ve Stream Analytics je rozsah LAG (tj. jak daleko od aktuální události musí vypadat) vždy omezen na konečný časový interval pomocí klauzule LIMIT DURATION. Lag lze volitelně omezit pouze na události, které odpovídají aktuální události v určité vlastnosti nebo podmínce pomocí klauzulí PARTITION BY a WHEN.

NA LAG nemají vliv predikáty v klauzuli WHERE, podmínky spojení v klauzuli JOIN ani výrazy seskupení v klauzuli GROUP BY aktuálního dotazu, protože se vyhodnocuje před těmito klauzulemi.

Syntax

LAG(<scalar_expression >, [<offset >], [<default>])  
     OVER ([PARTITION BY <partition key>] LIMIT DURATION(<unit>, <length>) [WHEN boolean_expression])

Například:

LAG(reading) OVER (LIMIT DURATION(hour, 3))  
LAG(name, 2, 'none such') OVER (PARTITION BY userId LIMIT DURATION(minute, 2))

Argumenty

Scalar_expression

Hodnota, která má být vrácena na základě zadaného posunu. Jedná se buď o výraz libovolného typu, který vrací jednu (skalární) hodnotu, nebo výraz se zástupným znakem *. Pro *se vrátí celá událost podle zadaného posunu a bude obsažena ve výsledné události (vnořený záznam).
scalar_expression nesmí obsahovat jiné analytické funkce ani externí funkce.

Posun

Počet událostí zpět z aktuální události, ze kterých se má získat hodnota. Pokud není zadaný, výchozí hodnota je 1, což znamená, že vrátí předchozí událost. Posun musí být celé číslo větší nebo rovno 1. Události se zpracovávají v dočasném pořadí. Pokud existuje několik událostí se stejným časovým razítkem, události se zpracovávají v pořadí doručení.

default

Hodnota, která se má vrátit v případě, že v zadaném posunu není žádná událost. Pokud není zadána výchozí hodnota, vrátí se hodnota NULL. "Žádná událost při zadaném posunu" může být v případě 1) pokud je počet odpovídajících událostí dosud menší než zadaný posun, nebo 2) pokud událost v zadaném posunu vypršel časový limit podle zadaného limit_duration_clause 3) události existují, ale neodpovídají logické podmínce zadané v when_clause.

Pokud událost v zadaném posunu existuje a hodnota scalar_expression je NULL, pak null
. výchozí může být sloupec, poddotaz nebo jiný výraz, ale nesmí obsahovat jiný výraz.
analytických funkcí nebo externích funkcí. výchozí musí mít přesně stejný typ jako
Scalar_expression.

OVER ( [ partition_by_clause ] limit_duration_clause [when_clause])

partition_by_clause KLAUZULE PARTITION BY <partition key> vyžaduje, aby pouze události, jejichž hodnota
<klíč oddílu je stejný jako klíč> aktuální události, kterou je třeba vzít v úvahu. Příklad:

LAG(reading) OVER (PARTITION BY sensorId LIMIT DURATION(hour, 1))

vrátí předchozí čtení stejného senzoru jako aktuální událost (pokud k tomu došlo během předchozí 1 hodiny).

limit_duration klauzule DURATION(<jednotka>, <délka>)

Určuje, jak velkou část historie z aktuální události je třeba vzít v úvahu. Podrobný popis podporovaných jednotek a jejich zkratek najdete v tématu DATEDIFF. Pokud se v intervalu DURATION nenajde dostatek odpovídajících událostí, <vrátí se výchozí> hodnota.

when_clause
Určuje logickou podmínku pro události, které se mají vzít v úvahu ve výpočtu LAG. Pokud se v intervalu DURATION nenajde dostatek odpovídajících událostí, <vrátí se výchozí> hodnota. When_clause je nepovinný.

Návratové typy

Datový typ zadaného scalar_expression. Pokud scalar_expression, vrátí se hodnota NULL.

Obecné poznámky

LAG je nedeterministické. Události se zpracovávají v dočasném pořadí. Pokud existuje několik událostí se stejným časovým razítkem, události se zpracovávají v pořadí doručení.

Použití LAG na sadu výsledků funkce oken může vést k neočekávaným výsledkům . Funkce oken mění časové razítko událostí, protože každá operace okna vypíše událost na konci okna. Aktuální časové razítko události je přístupné pomocí system.timestamp() a po operaci okna se bude lišit od původního atributu času události. Pokud před operací okna nejde přesunout lag, zvažte použití Funkce CollectTop, která se bude řadit podle původního času události.

Příklady

Výpočet rychlosti růstu na senzor:

SELECT sensorId,  
       growth = reading -
                        LAG(reading) OVER (PARTITION BY sensorId LIMIT DURATION(hour, 1))  
FROM input

Vyhledejte předchozí čtení senzoru not-null:

SELECT  
     sensorId,  
     LAG(reading) OVER (PARTITION BY sensorId LIMIT DURATION(hour, 1) WHEN reading IS NOT NULL)  
     FROM input

Vyhledejte předchozí nenulové čtení senzoru pro konkrétní typ senzoru:

WITH filterSensor AS
(
  SELECT *
  FROM input
  WHERE input.sensorType = 4 AND sensorId IS NOT NULL
)

SELECT
  LAG(reading) OVER (PARTITION BY sensorId LIMIT DURATION(hour, 1))
FROM filterSensor

Určení, kdy proměnná překročí prahovou hodnotu:

SELECT
    sensorId, reading
FROM input
WHERE
    devicetype = 'thermostat'
    AND reading > 100
    AND LAG(reading) OVER (PARTITION BY sensorId LIMIT DURATION(hour, 1) WHEN devicetype = 'thermostat') <= 100

Viz také

ISFIRST (Azure Stream Analytics)
LAST (Azure Stream Analytics)

Sdílet prostřednictvím