LAG (Azure Stream Analytics)

Artikel
08/02/2024

MED OPERATORN LAG-analys kan en slå upp en "tidigare" händelse i en händelseström, inom vissa begränsningar. Det är mycket användbart för att beräkna tillväxttakten för en variabel, identifiera när en variabel överskrider ett tröskelvärde eller när ett villkor startar eller slutar vara sant.

I Stream Analytics är omfånget för LAG (dvs. hur långt tillbaka i historiken från den aktuella händelsen den behöver se ut) alltid begränsat till ett begränsat tidsintervall med hjälp av LIMIT DURATION-satsen. LAG kan också begränsas till att endast överväga händelser som matchar den aktuella händelsen på en viss egenskap eller ett visst villkor med hjälp av PARTITION BY- och WHEN-satserna.

LAG påverkas inte av predikat i WHERE-satsen, kopplingsvillkor i JOIN-satsen eller grupperingsuttryck i GROUP BY-satsen i den aktuella frågan eftersom den utvärderas före dessa satser.

Syntax

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

Till exempel:

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

Argument

scalar_expression

Värdet som ska returneras baserat på den angivna förskjutningen. Det är antingen ett uttryck av valfri typ som returnerar ett enskilt (skalärt) värde eller jokertecknets uttryck '*'. För "*" returneras hela händelsen enligt den angivna förskjutningen och kommer att finnas i resultathändelsen (kapslad post).
scalar_expression får inte innehålla andra analysfunktioner eller externa funktioner.

offset

Antalet händelser från den aktuella händelsen som ett värde ska hämtas från. Om det inte anges är standardvärdet 1, vilket innebär att den returnerar föregående händelse. Förskjutningen måste vara ett heltal som är större än eller lika med 1. Händelser bearbetas i tidsordning. Om det finns flera händelser med samma tidsstämpelhändelser bearbetas i ankomstordningen.

standard

Det värde som ska returneras när det inte finns någon händelse vid den angivna förskjutningen. Om ett standardvärde inte anges returneras NULL. "Ingen händelse vid den angivna förskjutningen" kan vara fallet 1) om antalet motsvarande händelser som hittills har setts är mindre än den angivna förskjutningen eller 2) om händelsen vid den angivna förskjutningen överskrids enligt den angivna limit_duration_clause 3) händelser finns men inte matchar booleskt villkor som anges i when_clause.

Om händelsen vid den angivna förskjutningen finns och värdet för scalar_expression är NULL är null
returneras. standardvärdet kan vara en kolumn, en underfråga eller ett annat uttryck, men det får inte innehålla andra
analysfunktioner eller externa funktioner. standard måste ha exakt samma typ som
scalar_expression.

ÖVER ( [ partition_by_clause ] limit_duration_clause [when_clause])

partition_by_clause PARTITION BY-partitionsnyckelsatsen <> begär att endast händelser vars värde för
<partitionsnyckeln> är densamma som för den aktuella händelsen. Till exempel,

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

returnerar den tidigare avläsningen av samma sensor som den aktuella händelsen (om sådan inträffat inom de föregående 1 timmarna).

limit_duration-sats DURATION(<enhet>, <längd>)

Anger hur mycket av historiken från den aktuella händelsen som måste beaktas. Se DATEDIFF för en detaljerad beskrivning av enheter som stöds och deras förkortningar. Om det inte finns tillräckligt med matchande händelser inom varaktighetsintervallet <returneras standardvärdet> .

when_clause
Anger booleskt villkor för de händelser som ska beaktas i LAG-beräkningen. Om det inte finns tillräckligt med matchande händelser inom varaktighetsintervallet <returneras standardvärdet> . When_clause är valfritt.

Returtyper

Datatypen för den angivna scalar_expression. NULL returneras om scalar_expression

Allmänna kommentarer

LAG är icke-terministiskt. Händelser bearbetas i tidsordning. Om det finns flera händelser med samma tidsstämpelhändelser bearbetas i ankomstordningen.

Om LAG tillämpas på resultatuppsättningen för en fönsterfunktion kan oväntade resultat uppstå. Fönsterfunktioner ändrar tidsstämpeln för händelser, eftersom varje fönsteråtgärd matar ut händelsen i slutet av fönstret. Den aktuella tidsstämpeln för en händelse kan nås med system.timestamp(), efter en fönsteråtgärd skiljer den sig från det ursprungliga händelsetidsattributet. Om LAG inte kan flyttas före fönsteråtgärden bör du överväga att använda CollectTop och beställa efter den ursprungliga händelsetiden.

Exempel

Beräkna tillväxttakten per sensor:

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

Hitta tidigare sensoravläsningar som inte är null:

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

Hitta tidigare sensoravläsningar som inte är null för en viss sensortyp:

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

Avgöra när en variabel överskrider ett tröskelvärde:

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

Se även

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

Dela via