Běžné vzory dotazů v Azure Stream Analytics

dotazy v Azure Stream Analytics jsou vyjádřeny v dotazovacím jazyku podobném SQL. Jazykové konstrukce jsou zdokumentovány v referenční příručce jazyka Stream Analytics dotazů .

Návrh dotazu může vyjádřit jednoduchou předávací logiku pro přesun dat událostí z jednoho vstupního datového proudu do výstupního úložiště dat, nebo může provádět bohatou porovnávání vzorů a dočasnou analýzu pro výpočet agregovaných hodnot v různých časových oknech, jako v sestavení řešení IoT pomocí průvodce Stream Analytics . K kombinování událostí streamování můžete spojit data z několika vstupů a můžete provádět vyhledávání na základě statických referenčních dat a rozšířit hodnoty událostí. Můžete také zapisovat data do více výstupů.

Tento článek popisuje řešení několika běžných vzorů dotazů založených na scénářích reálného světa.

Podporované formáty dat

Azure Stream Analytics podporuje zpracování událostí v datových formátech CSV, JSON a Avro.

JSON a Avro mohou obsahovat komplexní typy, jako jsou například vnořené objekty (záznamy) nebo pole. Další informace o práci s těmito komplexními datovými typy najdete v článku Analýza JSON a data Avro .

Odesílání dat do více výstupů

Pro výstup dat do různých výstupních umyvadel lze použít vícenásobné příkazy Select . Například jedna možnost vybrat může vygenerovat výstrahu na základě prahové hodnoty, zatímco jiná může vygenerovat události do úložiště objektů BLOB.

Vstup:

Výstupní ArchiveOutput:

Výstupní AlertOutput:

Dotaz:

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

Klauzule into oznamuje Stream Analytics, na které výstupy mají zapisovat data. První Výběr definuje předávací dotaz, který přijímá data ze vstupu a odesílá je do výstupu s názvem ArchiveOutput. Druhý dotaz provede několik jednoduchých agregací a filtrování před odesláním výsledků do výstupního systému výstrah pro příjem dat s názvem AlertOutput.

Všimněte si, že klauzuli with lze použít k definování více bloků dílčího dotazu. Tato možnost je výhodou otevření menšího počtu čtenářů ke vstupnímu zdroji.

Dotaz:

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

Další informace najdete v tématu s klauzulí with.

Jednoduchý předávací dotaz

Pomocí jednoduchého předávacího dotazu můžete zkopírovat data vstupního datového proudu do výstupu. například pokud se datový proud dat, který obsahuje informace o vozidlech v reálném čase, musí uložit do SQL databáze pro účely analýzy, bude úloha provádět jednoduchý předávací dotaz.

Vstup:

Výstup:

Dotaz:

SELECT
    *
INTO Output
FROM Input

Dotaz Select * vyprojektech všechna pole příchozí události a odesílá je do výstupu. Stejný způsob, jak Vybrat , lze také použít pouze k projektu povinných polí ze vstupu. V tomto příkladu, pokud je vytvářená a Časová a časová pole, která se mají uložit, je možné tato pole zadat v příkazu Select .

Vstup:

Výstup:

Dotaz:

SELECT
    Make, Time
INTO Output
FROM Input

Porovnávání řetězců s like a NOT LIKE

Like a NOT LIKE lze použít k ověření, jestli pole odpovídá určitému vzoru. Můžete například vytvořit filtr, který vrátí pouze registrační desky, které začínají písmenem A a končí číslem 9.

Vstup:

Výstup:

Dotaz:

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

Pomocí příkazu LIKE zkontrolujte hodnotu License_plate pole. Měl by začít písmenem "A", pak by měl mít libovolný řetězec nula nebo více znaků, který končí číslem 9.

Výpočet minulých událostí

Funkci LAG můžete použít k zobrazení minulých událostí v časovém okně a jejich porovnání s aktuální událostí. Například aktuální značka auta může být ve výstupu, pokud se liší od posledního auta, které prošly mýtem.

Vstup:

Výstup:

Dotaz:

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

Pomocí funkce LAG si prohlédněte vstupní stream o jednu událost zpět, načítáte hodnotu Make a porovnáváte ji s hodnotou Make aktuální události a výstupem události.

Další informace najdete v tématu LAG.

Vrácení poslední události v okně

Vzhledem k tomu, že systém využívá události v reálném čase, neexistuje žádná funkce, která by určila, jestli bude událost poslední, která v tomto časovém období dorazí jako poslední. Aby toho bylo možné dosáhnout, musí být vstupní datový proud připojen k jinému, kde čas události je maximální doba pro všechny události v tomto okně.

Vstup:

Výstup:

Dotaz:

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

První krok dotazu najde maximální časové razítko v 10minutových časových oknech, které je časovým razítkem poslední události pro toto okno. Druhý krok spojí výsledky prvního dotazu s původním datovým proudem a najde událost, která odpovídá posledním časovým razítkům v každém okně.

DATEDIFF je funkce specifická pro datum, která porovnává a vrací časový rozdíl mezi dvěma poli DateTime. Další informace najdete v tématu o datech.

Další informace o spojování datových proudů najdete v tématu JOIN.

Agregace dat v průběhu času

K výpočtu informací v časovém období je možné data agregovat dohromady. V tomto příkladu se počet vypočítá za posledních 10 sekund času pro každé konkrétní auto.

Vstup:

Výstup:

Dotaz:

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

Tato agregace seskupuje automobily pomocí a počítá je každých 10 sekund. Výstup má na začátku a na počtu vozidel, která se provedla prostřednictvím mýtné.

TumblingWindow je funkce Window, která slouží k seskupení událostí dohromady. Agregaci lze použít pro všechny seskupené události. Další informace najdete v tématu funkce pro okna.

Další informace o agregaci naleznete v tématu agregační funkce.

Periodické výstupní hodnoty

V případě nepravidelných nebo chybějících událostí lze výstup pravidelného intervalu vygenerovat z zhuštěného datového vstupu. Vygenerujte například událost každých 5 sekund, která hlásí poslední zjištěný datový bod.

Vstup:

Výstup (prvních 10 řádků):

Dotaz:

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)

Tento dotaz generuje události každých 5 sekund a vypíše poslední událost, která byla dříve přijata. Doba trvání HOPPINGWINDOW určuje, jak daleko se má dotaz Hledat v poslední události.

Další informace najdete v okně skákající.

Korelace událostí v datovém proudu

Korelace událostí ve stejném datovém proudu se dá udělat tak, že si prohlížíte minulé události pomocí funkce Lag . Výstup lze například vygenerovat pokaždé, když dvě po sobě jdoucí automobily ze stejné linky projdou za posledních 90 sekund.

Vstup:

Výstup:

Dotaz:

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

Funkce LAG se může podívat na vstupní stream o jednu událost zpět a načíst hodnotu Make a porovnat ji s hodnotou Make aktuální události. Po splnit tuto podmínku je možné data z předchozí události projektovat pomocí funkce LAG v příkazu SELECT.

Další informace najdete v tématu LAG.

Zjištění doby trvání mezi událostmi

Dobu trvání události můžete vypočítat pohledem na poslední událost Start po přijetí události End. Tento dotaz může být užitečný k určení času, který uživatel stráví na stránce nebo funkci.

Vstup:

Výstup:

Dotaz:

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'

Funkci LAST lze použít k načtení poslední události v rámci konkrétní podmínky. V tomto příkladu je podmínkou událost typu Start, která rozděluje hledání podle uživatele a funkce PARTITION BY. Tímto způsobem se s každým uživatelem a funkcí zachází nezávisle při hledání události Start. LIMIT DURATION omezuje hledaný čas zpět na 1 hodinu mezi událostmi End a Start.

Počet jedinečných hodnot

Funkce COUNT a DISTINCT se dají použít k počítání jedinečných hodnot polí, které se zobrazí ve streamu v časovém okně. Můžete vytvořit dotaz, který vypočítá, kolik jedinečných automobilů prošlo mýtem ve 2sekudových oknech.

Vstup:

Výstup:

Dotaz:

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

FUNKCE COUNT(DISTINCT Make) vrátí počet jedinečných hodnot ve sloupci Vytvořit v časovém okně. Další informace najdete v tématu Agregační funkce COUNT.

Načtení první události v okně

IsFirst lze použít k načtení první události v časovém okně. Příkladem je vystupování informací o prvním automobilu v intervalu 10 minut.

Vstup:

Výstup:

Dotaz:

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

IsFirst může také rozdělit data a vypočítat první událost na každé konkrétní auto Make nalezené v intervalu 10 minut.

Výstup:

Dotaz:

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

Další informace najdete v tématu IsFirst.

Odebrání duplicitních událostí v okně

Při provádění operace, jako je výpočet průměrů nad událostmi v daném časovém okně, by se měly filtrovat duplicitní události. V následujícím příkladu je druhá událost duplikátem první události.

Vstup:

Výstup:

Dotaz:

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)

FUNKCE COUNT(DISTINCT Time) vrátí počet jedinečných hodnot ve sloupci Čas v časovém okně. Výstup prvního kroku se pak může použít k výpočtu průměru na zařízení zahozením duplicit.

Další informace najdete v tématu COUNT(DISTINCT Time).

Určení logiky pro různé případy/hodnoty (příkazy CASE)

Příkazy CASE mohou poskytovat různé výpočty pro různá pole na základě konkrétního kritéria. Například přiřaďte dráhu "A" automobilům značky Make1 a dráhu "B" k jakékoli jiné značky.

Vstup:

Výstup:

Řešení:

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

Výraz CASE porovná výraz se sadu jednoduchých výrazů, aby určil jeho výsledek. V tomto příkladu se vozidla make1 odbaví do dráhy "A", zatímco vozidla všech ostatních zozidel budou mít přiřazenou dráhu "B".

Další informace najdete v tématu o výrazu případu.

Převod dat

Data lze přetypovat v reálném čase pomocí metody CAST. Například Tloušťka auta může být převedena z typu nvarchar (max) na typ bigint a lze ji použít pro číselný výpočet.

Vstup:

Výstup:

Dotaz:

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

Použijte příkaz cast k určení jeho datového typu. Podívejte se na seznam podporovaných datových typů u datových typů (Azure Stream Analytics).

Další informace o funkcích pro převod dat.

Zjištění doby trvání podmínky

Pro podmínky, které jsou v rozsahu více událostmi, lze funkci Lag použít k identifikaci doby trvání této podmínky. Předpokládejme například, že chyba byla způsobena tím, že všechna auta mají nesprávnou váhu (nad 20 000 libry) a že se musí vypočítat doba trvání této chyby.

Vstup:

Výstup:

Dotaz:

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

První příkaz Select koreluje aktuální váhu s předchozí měřením a propojuje je s aktuální měření. Druhý Výběr se vrátí k poslední události, kde je previous_weight menší než 20000, přičemž aktuální váha je menší než 20000 a previous_weight aktuální události byla větší než 20000.

End_fault je aktuální nepoškozená událost, u které došlo k chybě předchozí události, a Start_fault je poslední nepoškozená událost.

Zpracování událostí s nezávislým časem (podproudy)

Události mohou docházet pozdě nebo mimo pořadí z důvodu výpadků hodin mezi výrobci událostí, hodinovým zkosením mezi oddíly nebo latencí sítě. Například hodiny zařízení pro TollID 2 jsou pět sekund za TollID 1 a hodiny zařízení pro TollID 3 jsou deset sekund po hodnotě TollID 1. Výpočet může být nezávisle u každého placená linka a zvažuje pouze vlastní data o hodinách jako časové razítko.

Vstup:

Výstup:

Dotaz:

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

Klauzule TIMESTAMP OVER BY se na časovou osu jednotlivých zařízení dívá nezávisle pomocí dílčích streamů. Výstupní událost pro každé TOLLID se vygeneruje při výpočtu, což znamená, že události jsou seřazené s ohledem na jednotlivá idllid místo toho, aby se přeuspořádaly, jako by všechna zařízení byla ve stejných hodinách.

Další informace najdete v tématu TIMESTAMP BY OVER.

Relace Windows

Okno relace je okno, které se stále rozšiřuje, když dojde k událostem, a zavře se pro výpočet, pokud se po určité době neobdrží žádná událost nebo pokud okno dosáhne maximální doby trvání. Toto okno je zvlášť užitečné při výpočtu dat interakce uživatele. Okno se spustí, když uživatel začne pracovat se systémem, a zavře se, když se nezaznamenají žádné další události, což znamená, že uživatel přestal interagovat. Uživatel například pracuje s webovou stránkou, na které se zaprotokoluje počet kliknutí, a pomocí okna relace můžete zjistit, jak dlouho uživatel s webem interaguje.

Vstup:

Výstup:

Dotaz:

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 pro projektuje data, která jsou relevantní pro interakci uživatele, spolu s dobou trvání interakce. Seskupení dat podle uživatele a okna RelaceWindow, které se zavře, pokud během 1 minuty ne dojde k žádné interakci, s maximální velikostí okna 60 minut.

Další informace o SessionWindow najdete v tématu Okno relace.

Rozšiřitelnost jazyka pomocí uživatelem definované funkce v JavaScriptu a C

Azure Stream Analytics jazyka lze rozšířit o vlastní funkce napsané v jazyce JavaScript nebo C#. Uživatelem definované funkce (UDF) jsou vlastní/složité výpočty, které nelze snadno vyjádřit pomocí SQL jazyka. Tyto uživatelem definované funkce je možné definovat jednou a použít v dotazu vícekrát. Funkce UDF se například může použít k převodu šestnáctkové hodnoty nvarchar(max) na hodnotu bigint.

Vstup:

Výstup:

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

Uživatelem definovaná funkce vypočítá hodnotu bigint z hodnoty HexValue při každé spotřebované události.

Další informace najdete v tématu JavaScript a C#.

Pokročilé porovnávání vzorů s MATCH_RECOGNIZE

MATCH_RECOGNIZE je pokročilý mechanismus porovnávání vzorů, který lze použít ke spárování posloupnosti událostí s dobře definovaným vzorem regulárního výrazu. Například atm se v reálném čase monitoruje v případě selhání během provozu ATM, pokud se zobrazí dvě po sobě jdoucí varovné zprávy, které musí správce oznámit.

Vstup:

Výstup:

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

Tento dotaz odpovídá alespoň dvěma po sobě jdoucím událostem selhání a při splnění podmínek vygeneruje alarm. PATTERN definuje regulární výraz, který se má použít pro porovnávání, v tomto případě alespoň dvě po sobě jdoucí upozornění po alespoň jedné úspěšné operaci. Úspěch Return_Code Upozornění se definují pomocí hodnoty Return_Code a po dokončení podmínky se míry projektují pomocí ATM_id , první operace upozornění Return_Code prvního upozornění.

Další informace najdete v tématu MATCH_RECOGNIZE.

Geofencing a geoprostorové dotazy

Azure Stream Analytics poskytuje integrované geoprostorové funkce, které je možné použít k implementaci scénářů, jako je správa vozového parku, sdílení jízd, připojené auta a sledování prostředků. Geoprostorová data je možné ingestovat ve formátech GeoJSON nebo WKT jako součást streamu událostí nebo referenčních dat. Například společnost, která se specializuje na výrobní stroje pro tisk pasů, pronajímá své stroje vládám a vládym. Umístění těchto počítačů je silně kontrolovaným způsobem, aby se zabránilo záměně a možnému použití k padělkům pasů. Každý počítač je vybavený GPS trackerem, který předává informace zpět do Azure Stream Analytics úlohy. Výrobce by chtěl sledovat polohu těchto počítačů a být upozorněn, pokud jeden z nich opustí autorizovaný prostor. Tímto způsobem může vzdáleně zakázat, upozornit autority a načíst vybavení.

Vstup:

Vstup referenčních dat:

Výstup:

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

Dotaz umožňuje výrobci automaticky monitorovat umístění počítačů a dostávat upozornění, když počítač opustí povolenou geografickou polohu. Integrovaná geoprostorová funkce umožňuje uživatelům používat data GPS v rámci dotazu bez knihoven třetích stran.

Další informace najdete v článku Scénáře geofencingu a geoprostorové agregace s Azure Stream Analytics.

Získání pomoci

Pokud chcete další pomoc, vyzkoušejte naši stránku s&Microsoft Q Azure Stream Analytics.

Další kroky

• Úvod do Azure Stream Analytics
• Začínáme používat službu Azure Stream Analytics
• Škálování služby Stream Analytics
• Referenční příručka k jazyku Azure Stream Analytics Query Language
• Referenční příručka k rozhraní REST API pro správu služby Azure Stream Analytics