Vanliga frågemönster i Azure Stream Analytics

Frågor i Azure Stream Analytics uttrycks i ett SQL liknande frågespråk. Språkkonstruktionerna dokumenteras i guiden Stream Analytics referens för frågespråk.

Frågedesignen kan uttrycka enkel genomströmningslogik för att flytta händelsedata från en indataström till ett utdatalager, eller göra omfattande mönstermatchning och tidsanalys för att beräkna aggregeringar över olika tidsfönster som i guiden Skapa en IoT-lösning med hjälp av Stream Analytics. Du kan koppla data från flera indata för att kombinera direktuppspelningshändelser, och du kan göra uppslag mot statiska referensdata för att utöka händelsevärdena. Du kan också skriva data till flera utdata.

Den här artikeln beskriver lösningar på flera vanliga frågemönster baserat på verkliga scenarier.

Dataformat som stöds

Azure Stream Analytics stöder bearbetning av händelser i CSV-, JSON- och Avro-dataformat.

Både JSON och Avro kan innehålla komplexa typer, till exempel kapslade objekt (poster) eller matriser. Mer information om hur du arbetar med dessa komplexa datatyper finns i artikeln Parsa JSON- och AVRO-data.

Skicka data till flera utdata

Flera SELECT-instruktioner kan användas för att mata ut data till olika utdata-mottagare. En SELECT kan till exempel mata ut en tröskelbaserad avisering medan en annan kan mata ut händelser till Blob Storage.

Indata:

ArkivutdataUtdataUtdata:

UtdataaviseringUtdata:

Fråga:

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-satsen Stream Analytics vilka utdata som data ska skrivas till. Den första SELECT definierar en genomströmningsfråga som tar emot data från indata och skickar dem till utdata med namnet ArchiveOutput. Den andra frågan gör viss enkel aggregering och filtrering innan resultatet skickas till ett nedströms aviseringssystem med namnet AlertOutput.

Observera att WITH-satsen kan användas för att definiera flera underfrågeblock. Det här alternativet har fördelen att du öppnar färre läsare för indatakällan.

Fråga:

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

Mer information finns i WITH-satsen.

Enkel direktfråga

En enkel genomströmningsfråga kan användas för att kopiera indataströmmen till utdata. Om till exempel en dataström som innehåller fordonsinformation i realtid måste sparas i en SQL-databas för bokstavsanalys, kommer en enkel genomströmningsfråga att göra jobbet.

Indata:

Utdata:

Fråga:

SELECT
    *
INTO Output
FROM Input

En SELECT *-fråga projekterar alla fält i en inkommande händelse och skickar dem till utdata. På samma sätt kan SELECT också användas för att endast projicera obligatoriska fält från indata. I det här exemplet, om fordons make och time är de enda obligatoriska fälten som ska sparas, kan dessa fält anges i SELECT-instruktionen.

Indata:

Utdata:

Fråga:

SELECT
    Make, Time
INTO Output
FROM Input

Strängmatchning med LIKE och NOT LIKE

LIKE och NOT LIKE kan användas för att kontrollera om ett fält matchar ett visst mönster. Ett filter kan till exempel skapas för att endast returnera de registreringsskyltar som börjar med bokstaven "A" och slutar med siffran 9.

Indata:

Utdata:

Fråga:

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

Använd LIKE-instruktionen för att kontrollera License_plate-fältvärdet. Den bör börja med bokstaven "A" och sedan ha en sträng med noll eller fler tecken som slutar med siffran 9.

Beräkning över tidigare händelser

Funktionen LAG kan användas för att titta på tidigare händelser inom ett tidsfönster och jämföra dem med den aktuella händelsen. Till exempel kan den aktuella bilbilen matas ut om den skiljer sig från den sista bilen som kördes via vägtullen.

Indata:

Utdata:

Fråga:

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

Använd LAG för att titta på indataströmmen en händelse tillbaka, hämta Värdet Skapa och jämföra det med värdet Make för den aktuella händelsen och mata ut händelsen.

Mer information finns i LAG.

Returnera den sista händelsen i ett fönster

Eftersom händelser förbrukas av systemet i realtid finns det ingen funktion som kan avgöra om en händelse blir den sista som tas emot under den tidsperioden. För att uppnå detta måste indataströmmen vara ansluten till en annan där tiden för en händelse är den maximala tiden för alla händelser i det fönstret.

Indata:

Utdata:

Fråga:

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

Det första steget i frågan hittar den maximala tidsstämpeln i 10-minutersfönster, det vill säga tidsstämpeln för den sista händelsen för det fönstret. Det andra steget sammanfogar resultatet av den första frågan med den ursprungliga dataströmmen för att hitta den händelse som matchar de senaste tidsstämplarna i varje fönster.

DATEDIFF är en datumspecifik funktion som jämför och returnerar tidsskillnaden mellan två DateTime-fält. Mer information finns i datumfunktionerna.

Mer information om hur du ansluter dataströmmar finns i JOIN.

Dataaggregering över tid

För att beräkna information över ett tidsfönster kan data aggregeras tillsammans. I det här exemplet beräknas ett antal under de senaste 10 sekunderna för varje specifik bil.

Indata:

Utdata:

Fråga:

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

Den här aggregeringen grupperar bilar efter Make och räknar dem var 10:e sekund. Utdata har bil make och count som gick via vägtullen.

TumblingWindow är en fönsterfunktion som används för att gruppera händelser. En aggregering kan tillämpas på alla grupperade händelser. Mer information finns i fönsterfunktioner.

Mer information om aggregering finns i mängdfunktioner.

Utdatavärden med jämna mellanrum

Vid oregelbundna eller saknade händelser kan utdata med regelbundna intervall genereras från en mer gles datainmatning. Du kan till exempel generera en händelse var femte sekund som rapporterar den senast sedda datapunkten.

Indata:

Utdata (de första 10 raderna):

Fråga:

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)

Den här frågan genererar händelser var femte sekund och matar ut den senaste händelsen som togs emot tidigare. HOPPINGWINDOW-varaktigheten avgör hur långt tillbaka frågan ser ut för att hitta den senaste händelsen.

Mer information finns i Hoppande fönster.

Korrelera händelser i en dataström

Korrelering av händelser i samma dataström kan göras genom att titta på tidigare händelser med hjälp av FUNKTIONEN LAG. Till exempel kan utdata genereras varje gång två bilar i följd från samma Make går igenom avgiften för de senaste 90 sekunderna.

Indata:

Utdata:

Fråga:

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

Funktionen LAG kan titta på indataströmmen en händelse tillbaka och hämta värdet Make (Skapa) och jämföra det med make-värdet för den aktuella händelsen. När villkoret är uppfyllt kan data från den föregående händelsen projiceras med hjälp av LAG i SELECT-instruktionen.

Mer information finns i LAG.

Identifiera varaktigheten mellan händelser

Varaktigheten för en händelse kan beräknas genom att titta på den senaste starthändelsen när en sluthändelse tas emot. Den här frågan kan vara användbar för att avgöra hur lång tid en användare tillbringar på en sida eller en funktion.

Indata:

Utdata:

Fråga:

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'

Funktionen LAST kan användas för att hämta den sista händelsen inom ett visst villkor. I det här exemplet är villkoret en händelse av typen Start, där sökningen partitioneras efter PARTITIONERAD AV användare och funktion. På så sätt behandlas alla användare och funktioner oberoende av varandra när du söker efter Start-händelsen. BEGRÄNSA VARAKTIGHET begränsar sökningen tillbaka i tiden till 1 timme mellan slut- och starthändelserna.

Räkna unika värden

COUNT och DISTINCT kan användas för att räkna antalet unika fältvärden som visas i dataströmmen inom ett tidsfönster. Du kan skapa en fråga för att beräkna hur många unika makes of cars som skickas via avgiftsbelagda bås i ett fönster på 2 sekunder.

Indata:

Produktionen:

Fråga:

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

COUNT(DISTINCT Make) returnerar antalet distinkta värden i kolumnen Skapa inom ett tidsfönster. Mer information finns i mängdfunktionen COUNT.

Hämta den första händelsen i ett fönster

IsFirst kan användas för att hämta den första händelsen i ett tidsfönster. Du kan till exempel mata ut den första bilinformationen vid varje 10-minutersintervall.

Indata:

Utdata:

Fråga:

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

IsFirst kan också partitionera data och beräkna den första händelsen för varje specifik bil Make som finns var 10:e minut.

Utdata:

Fråga:

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

Mer information finns i IsFirst.

Ta bort dubbletthändelser i ett fönster

När du utför en åtgärd, till exempel att beräkna medelvärden över händelser i ett visst tidsfönster, ska duplicerade händelser filtreras. I följande exempel är den andra händelsen en dubblett av den första.

Indata:

Utdata:

Fråga:

With Temp AS (
SELECT
    COUNT(DISTINCT Time) AS CountTime,
    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)

COUNT(DISTINCT Time) returnerar antalet distinkta värden i kolumnen Tid inom ett tidsfönster. Utdata från det första steget kan sedan användas för att beräkna medelvärdet per enhet genom att ta bort dubbletter.

Mer information finns i COUNT(DISTINCT Time).

Ange logik för olika ärenden/värden (CASE-instruktioner)

CASE-instruktioner kan ge olika beräkningar för olika fält, baserat på specifika kriterier. Du kan till exempel tilldela band "A" till bilar med Make1 och band "B" till andra bilar.

Indata:

Utdata:

Lösning:

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-uttrycket jämför ett uttryck med en uppsättning enkla uttryck för att fastställa dess resultat. I det här exemplet skickas fordon i Make1 till band "A" medan fordon av någon annan make tilldelas band "B".

Mer information finns i ärendeuttrycket.

Datakonvertering

Data kan castas i realtid med hjälp av CAST-metoden. Bilvikt kan till exempel konverteras från typen nvarchar(max) till bigint och användas i en numerisk beräkning.

Indata:

Utdata:

Fråga:

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

Använd en CAST-instruktion för att ange dess datatyp. Se listan över datatyper som stöds för Datatyper (Azure Stream Analytics).

Mer information om datakonverteringsfunktionerfinns i .

Identifiera varaktigheten för ett villkor

För villkor som sträcker sig över flera händelser kan LAG-funktionen användas för att identifiera varaktigheten för villkoret. Anta till exempel att en bugg resulterade i att alla bilar hade en felaktig vikt (över 20 000 pund) och att buggens varaktighet måste beräknas.

Indata:

Utdata:

Fråga:

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

Den första SELECT-instruktionen korrelerar den aktuella viktmätningen med föregående mätning och projicerar den tillsammans med den aktuella mätningen. Den andra SELECT söker tillbaka till den senaste händelsen där previous_weight är mindre än 20000, där den aktuella vikten är mindre än 20000 och previous_weight för den aktuella händelsen var större än 20000.

Den End_fault är den aktuella icke-felaktiga händelsen där den tidigare händelsen var felaktig och Start_fault är den sista icke-felaktiga händelsen före den.

Bearbeta händelser med oberoende tid (underströmmar)

Händelser kan komma sent eller i fel ordning på grund av klockskevhet mellan händelseproducenter, klockskevhet mellan partitioner eller nätverksfördröjning. Enhetsklockan för TollID 2 är till exempel fem sekunder efter TollID 1 och enhetsklockan för TollID 3 är tio sekunder efter TollID 1. En beräkning kan ske oberoende av varje vägtull, med bara sina egna klockdata som en tidsstämpel.

Indata:

Utdata:

Fråga:

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

TIMESTAMP OVER BY-satsen tittar på varje enhets tidslinje oberoende av underströmmar. Utdatahändelsen för varje TollID genereras när de beräknas, vilket innebär att händelserna är i ordning med avseende på varje TollID i stället för att ordnas om som om alla enheter hade samma klocka.

Mer information finns i TIMESTAMP BY OVER.

Sessions-Windows

Ett sessionsfönster är ett fönster som expanderar allt eftersom händelser inträffar och stängs för beräkning om ingen händelse tas emot efter en viss tid eller om fönstret når sin maximala varaktighet. Det här fönstret är särskilt användbart när du beräknar användarinteraktionsdata. Ett fönster startar när en användare börjar interagera med systemet och stängs när inga fler händelser observeras, vilket innebär att användaren har slutat interagera. En användare interagerar till exempel med en webbsida där antalet klick loggas. Ett sessionsfönster kan användas för att ta reda på hur länge användaren har interagerat med webbplatsen.

Indata:

Utdata:

Fråga:

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 projekterar de data som är relevanta för användarinteraktion, tillsammans med interaktionens varaktighet. Gruppering av data efter användare och en SessionWindow som stängs om ingen interaktion sker inom 1 minut, med en maximal fönsterstorlek på 60 minuter.

Mer information om SessionWindow finns i Sessionsfönster.

Språk utökningsbarhet med användardefinierad funktion i JavaScript och C

Azure Stream Analytics kan utökas med anpassade funktioner som skrivits antingen i JavaScript- eller C#-språk. Användardefinierade funktioner (UDF) är anpassade/komplexa beräkningar som inte enkelt kan uttryckas med hjälp av SQL språk. Dessa användardefinierade filer kan definieras en gång och användas flera gånger inom en fråga. En UDF kan till exempel användas för att konvertera ett hexadecimalt nvarchar(max)-värde till ett bigint-värde.

Indata:

Utdata:

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

Den användardefinierade funktionen beräknar bigint-värdet från HexValue för varje händelse som förbrukas.

Mer information finns i JavaScript och C#.

Avancerad mönstermatchning med MATCH_RECOGNIZE

MATCH_RECOGNIZE är en avancerad metod för mönstermatchning som kan användas för att matcha en sekvens med händelser med ett väldefinierat mönster för reguljära uttryck. Till exempel övervakas en atm i realtid för fel. Under drift av ATM om det finns två på varandra följande varningsmeddelanden måste administratören meddelas.

Indata:

Utdata:

SELECT *
FROM input TIMESTAMP BY time OVER ATM_id
MATCH_RECOGNIZE (
    LIMIT DURATION(minute, 1)
    PARTITON 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 Succes.Return_Code = 'Success',
        Warning AS Warning.Return_Code <> 'Success'
) AS patternMatch

Den här frågan matchar minst två efterföljande felhändelser och genererar ett larm när villkoren uppfylls. PATTERN definierar det reguljära uttryck som ska användas för matchningen, i det här fallet minst två efterföljande varningar efter minst en lyckad åtgärd. Success och Warning definieras Return_Code värdet och när villkoret har uppfyllts projiceras MÅTTen med ATM_id, den första varningsåtgärden och första varningstiden.

Mer information finns i MATCH_RECOGNIZE.

Geofencing och geospatiala frågor

Azure Stream Analytics innehåller inbyggda geospatiala funktioner som kan användas för att implementera scenarier som vagnparkshantering, delning av resor, anslutna bilar och tillgångsspårning. Geospatiala data kan matas in i antingen GeoJSON- eller WKT-format som en del av händelseströmmen eller referensdata. Ett företag som till exempel är specialiserade på tillverkningsdatorer för att skriva ut pass, leasar sina datorer till myndigheter och myndigheter. Platsen för dessa datorer är kraftigt kontrollerad för att undvika felplattform och möjlig användning för förfalskade pass. Varje dator är försedd med en GPS-spårare. Informationen vidarebefordras tillbaka till ett Azure Stream Analytics jobb. En tillverkare vill hålla reda på var datorerna finns och få aviseringar om någon av dem lämnar ett godkänt område. På så sätt kan de fjärraktivera, varna myndigheter och hämta utrustningen.

Indata:

Indatareferens:

Utdata:

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_currenct_location, referenceInput.Equipment_lease_location) = 1

Frågan gör att tillverkaren kan övervaka datorernas plats automatiskt och få aviseringar när en dator lämnar den tillåtna geofence-platsen. Med den inbyggda geospatiala funktionen kan användarna använda GPS-data i frågan utan bibliotek från tredje part.

Mer information finns i artikeln Geofencing och geospatiala aggregeringsscenarier Azure Stream Analytics artikeln.

Få hjälp

Om du vill ha mer hjälp kan du prova vår Microsoft Q&A-frågesida för att Azure Stream Analytics.

Nästa steg

• Introduktion till Azure Stream Analytics
• Komma igång med Azure Stream Analytics
• Skala Azure Stream Analytics-jobb
• Referens för Azure Stream Analytics-frågespråket
• Referens för Azure Stream Analytics Management REST-API:et