Použití paralelizace dotazů ve službě Azure Stream Analytics

V tomto článku se dozvíte, jak využívat paralelizaci ve službě Azure Stream Analytics. Naučíte se škálovat úlohy Stream Analytics konfigurací vstupních oddílů a laděním definice analytického dotazu.

Jako předpoklad by bylo dobré se seznámit s pojmem streamovací jednotky, který je popsán v Pochopení a úprava streamovacích jednotek.

Jaké jsou části úlohy Stream Analytics?

Definice úlohy Stream Analytics zahrnuje alespoň jeden vstup streamování, dotaz a výstup. Vstupy jsou místem, odkud úloha čte datový proud. Dotaz se používá k transformaci vstupního datového proudu a výstup je místo, kam se odesílají výsledky úlohy.

Oddíly ve vstupech a výstupech

Dělení umožňuje rozdělit data na podmnožiny na základě klíče oddílu. Pokud je vstup (například Event Hubs) rozdělený podle klíče, doporučujeme při přidávání vstupu do úlohy Stream Analytics zadat klíč oddílu. Škálování úlohy Stream Analytics využívá particí ve vstupu a výstupu. Úloha Stream Analytics může paralelně využívat a zapisovat různé partitiony, což zvyšuje propustnost.

Vstupy

Všechny vstupy streamování Azure Stream Analytics můžou využívat dělení: Event Hubs, IoT Hub, Blob Storage, Data Lake Storage Gen2.

Poznámka:

Pro úroveň kompatibility 1.2 a vyšší je klíč oddílu nastaven jako vstupní vlastnost bez nutnosti klíčového slova PARTITION BY v dotazu. Pro úroveň kompatibility 1.1 a níže je potřeba klíč oddílu definovat pomocí klíčového slova PARTITION BY v dotazu.

Výstupy

Při práci se službou Stream Analytics můžete využít dělení ve výstupech:

• Azure Data Lake Storage
• Azure Functions
• Tabulka Azure
• Blob Storage (může explicitně nastavit klíč oddílu)
• Azure Cosmos DB (je potřeba explicitně nastavit klíč oddílu)
• Event Hubs (je potřeba explicitně nastavit klíč oddílu)
• IoT Hub (potřeba explicitně nastavit klíč oddílu)
• Sběrnice
• SQL a Azure Synapse Analytics s volitelným particionováním: viz více informací na stránce Azure SQL Database o výstupech.

Power BI nepodporuje dělení. I tak ale můžete vstup rozdělit, jak je popsáno v této části.

Další informace o oddílech najdete v následujících článcích:

• Přehled funkcí Event Hubs
• Dělení dat

Dotaz

Aby byla úloha paralelní, je potřeba klíče oddílů zarovnat mezi všemi vstupy, všemi kroky logiky dotazu a všemi výstupy. Dělení logiky dotazu určuje klíče používané pro spojení a agregace (GROUP BY). Poslední požadavek je možné ignorovat, pokud logika dotazu není klíčovaná (projekce, filtry, referenční spojení...).

• Pokud je vstup a výstup rozdělen na oddíly WarehouseId, a dotaz seskupuje podle ProductId bez WarehouseId, pak úloha není paralelní.
• Pokud jsou dva vstupy, které se mají spojit, rozdělené podle různých klíčů oddílu (WarehouseId a ProductId), pak úloha není paralelní.
• Pokud jsou dva nebo více nezávislých toků dat obsaženy v jedné úloze, každý s vlastním klíčem dělení, nedochází k paralelizaci úlohy.

Pouze pokud všechny vstupy, výstupy a kroky dotazu používají stejný klíč, úloha je paralelní.

Trapně paralelní úlohy

Nejškálovatelnější scénář v rámci služby Azure Stream Analytics je příhodně paralelní úloha. Připojí jeden oddíl vstupu k jedné instanci dotazu k jednomu oddílu výstupu. Tento paralelismus má následující požadavky:

• Pokud logika dotazu závisí na stejném klíči zpracovávaného stejnou instancí dotazu, musíte se ujistit, že události přejdou do stejného oddílu vašeho vstupu. U event Hubs nebo IoT Hubu to znamená, že data události musí mít nastavenou hodnotu PartitionKey . Alternativně můžete použít dělené odesílatele. V případě úložiště objektů blob jsou události odesílány do stejné složky oddílu. Příkladem by byla instance dotazu, která agreguje data podle ID uživatele, kde je vstupní centrum událostí rozdělené pomocí ID uživatele jako klíče oddílu. Pokud ale logika dotazu nevyžaduje zpracování stejného klíče stejnou instancí dotazu, můžete tento požadavek ignorovat. Příkladem této logiky by byl jednoduchý dotaz select-project-filter.

• Dalším krokem je zajistit, aby byl váš dotaz rozdělen na oddíly. Pro úlohy s úrovní kompatibility 1.2 nebo vyšší (doporučeno) lze vlastní sloupec použít jako klíč oddílu ve vstupním nastavení a úloha bude automaticky zpracovávána paralelně. Úlohy s úrovní kompatibility 1.0 nebo 1.1 vyžadují použití PARTITION BY PartitionId ve všech krocích dotazu. Je povoleno více kroků, ale všechny musí být rozděleny stejným klíčem.

• Většina výstupů podporovaných ve Stream Analytics může využít dělení. Pokud použijete typ výstupu, který nepodporuje dělení vaší úlohy, nebude trapně paralelní. U výstupů služby Event Hubs se ujistěte, že je sloupec klíče oddílu nastavený na stejný klíč oddílu použitý v dotazu. Další informace najdete v části výstupu.

• Počet vstupních oddílů se musí shodovat s počtem výstupních oddílů. Výstup úložiště objektů blob může podporovat oddíly a schéma rozdělení dědí z vstupního dotazu. Když zadáte klíč oddílu pro úložiště objektů blob, data se rozdělí podle každého vstupního oddílu, takže výsledek zůstává plně paralelní. Tady jsou příklady hodnot oddílů, které umožňují plně paralelní úlohu:

  • Osm vstupních oddílů centra událostí a osm výstupních oddílů centra událostí
  • Osm vstupních oddílů centra událostí a výstup úložiště Blob
  • Osm vstupních oddílů Event Hubu a rozdělení výstupu Blob Storage podle vlastního pole s libovolnou kardinalitou
  • Osm vstupních oddílů úložiště objektů blob a výstup úložiště objektů blob
  • Osm vstupních oddílů Blob Storage a osm výstupních oddílů Event Hub

V následujících částech najdete některé ukázkové scénáře, které jsou trapně paralelní.

Jednoduchý dotaz

• Vstup: Centrum událostí s osmi oddíly
• Výstup: Centrum událostí s osmi particemi (sloupec Klíč partice musí být nastavený na použití PartitionId)

Dotaz:

    --Using compatibility level 1.2 or above
    SELECT TollBoothId
    FROM Input1
    WHERE TollBoothId > 100
    
    --Using compatibility level 1.0 or 1.1
    SELECT TollBoothId
    FROM Input1 PARTITION BY PartitionId
    WHERE TollBoothId > 100

Tento dotaz je jednoduchý filtr. Proto se nemusíme starat o rozdělení vstupu, který se odesílá do centra událostí. Všimněte si, že úlohy s úrovní kompatibility před 1.2 musí obsahovat klauzuli PARTITION BY PartitionId , takže splňuje požadavek č. 2 z dřívější verze. Pro výstup musíme nakonfigurovat výstup centra událostí v úloze tak, aby měl klíč oddílu nastavený na PartitionId. Poslední kontrolou je zajistit, aby se počet vstupních oddílů rovnal počtu výstupních oddílů.

Dotazování pomocí klíče seskupení

• Vstup: Centrum událostí s osmi oddíly
• Výstup: Blob Storage

Dotaz:

    --Using compatibility level 1.2 or above
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId
    
    --Using compatibility level 1.0 or 1.1
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

Tento dotaz má klíč seskupení. Proto se seskupené události musí odeslat do stejného oddílu Event Hubs. Vzhledem k tomu, že v tomto příkladu seskupujeme podle TollBoothID, je třeba zajistit, aby byl TollBoothID při odesílání událostí do služby Event Hubs používán jako klíč oddílu. Potom v Azure Stream Analytics můžete pomocí PARTITION BY PartitionId dědit z tohoto schématu oddílů a povolit úplné paralelizace. Vzhledem k tomu, že výstupem je úložiště objektů blob, nemusíme se starat o konfiguraci hodnoty klíče oddílu podle požadavku č. 4.

Příklad scénářů, které nejsou * trapně paralelní

V předchozí části jsme se v článku zabývali některými trapně paralelními scénáři. V této části se dozvíte o scénářích, které nesplňují všechny požadavky, aby byly triviálně paralelní.

Neshodný počet oddílů

• Vstup: Centrum událostí s osmi oddíly
• Výstup: Centrum událostí s 32 oddíly

Pokud se počet vstupních oddílů neshoduje s počtem výstupních oddílů, topologie není snadno paralelizovatelná bez ohledu na dotaz. Stále ale můžeme získat určitou úroveň paralelizace.

Dotaz s využitím výstupu, který není rozdělený na oddíly

• Vstup: Centrum událostí s osmi oddíly
• Výstup: Power BI

Výstup Power BI v současné době nepodporuje dělení. Proto tento scénář není triviálně paralelní.

Vícekrokový dotaz s různými hodnotami PARTITION BY

• Vstup: Centrum událostí s osmi oddíly
• Výstup: Centrum událostí s osmi oddíly
• Úroveň kompatibility: 1.0 nebo 1.1

Dotaz:

    WITH Step1 AS (
    SELECT COUNT(*) AS Count, TollBoothId, PartitionId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1 Partition By TollBoothId
    GROUP BY TumblingWindow(minute, 3), TollBoothId

Jak vidíte, druhý krok jako klíč dělení používá TollBoothId . Tento krok není stejný jako první krok, a proto je nutné, abychom provedli přeskupení.

Vícekrokový dotaz s různými hodnotami PARTITION BY

• Vstup: Centrum událostí s osmi oddíly (sloupec "Partition key" není nastaven, výchozí hodnota "PartitionId")
• Výstup: Centrum událostí s osmi oddíly (sloupec klíč oddílu musí být nastavený na použití TollBoothId).
• Úroveň kompatibility – 1.2 nebo vyšší

Dotaz:

    WITH Step1 AS (
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute, 3), TollBoothId

Úroveň kompatibility 1.2 nebo vyšší umožňuje spouštění paralelních dotazů ve výchozím nastavení. Například dotaz z předchozí části bude rozdělen, pokud je sloupec "TollBoothId" nastavený jako vstupní klíč oddílu. Klauzule PARTITION BY PartitionId není nutná.

Výpočet maximálních jednotek streamování úlohy

Celkový počet jednotek streamování, které může úloha Stream Analytics použít, závisí na počtu kroků v dotazu definovaném pro úlohu a počtu oddílů pro každý krok.

Kroky v dotazu

Dotaz může mít jeden nebo mnoho kroků. Každý krok je poddotaz definovaný klíčovým slovem WITH . Dotaz, který je mimo klíčové slovo WITH (pouze jeden dotaz), se také počítá jako krok, například příkaz SELECT v následujícím dotazu:

Dotaz:

    WITH Step1 AS (
        SELECT COUNT(*) AS Count, TollBoothId
        FROM Input1 Partition By PartitionId
        GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )
    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute,3), TollBoothId

Tento dotaz má dva kroky.

Poznámka:

Tento dotaz je podrobněji popsán dále v článku.

Rozdělení kroku

Dělení kroku vyžaduje následující podmínky:

• Vstupní zdroj musí být rozdělený na oddíly.
• Příkaz SELECT dotazu musí číst z děleného vstupního zdroje.
• Dotaz v rámci kroku musí mít klíčové slovo PARTITION BY .

Při dělení dotazu se vstupní události zpracovávají a agregují v samostatných skupinách oddílů a výstupní události se generují pro každou ze skupin. Pokud chcete kombinovanou agregaci, musíte vytvořit druhý nepartitionovaný krok pro agregaci.

Výpočet maximálních jednotek streamování pro úlohu

Všechny nepartitionované kroky mohou škálovat až na jednu jednotku streamování (SU V2s) pro úlohu Stream Analytics. Kromě toho můžete přidat jeden SU V2 pro každý oddíl v rozděleném kroku. Některé příklady najdete v následující tabulce.

Dotaz	Maximální počet jednotek SU pro úlohu
Dotaz obsahuje jeden krok. Tento krok není rozdělený na oddíly.	1 SU V2
Vstupní datový proud je rozdělený na 16 částí. Dotaz obsahuje jeden krok. Krok je rozdělený na oddíly.	16 SU V2 (1 * 16 oddílů)
Dotaz obsahuje dva kroky. Žádný z kroků není rozdělený na oddíly.	1 SU V2
Vstupní datový proud je rozdělen na 3 části. Dotaz obsahuje dva kroky. Vstupní krok je rozdělený a druhý krok není rozdělený. Příkaz SELECT načítá z particionovaného vstupu.	4 SU V2s (3 pro rozdělené kroky + 1 pro nekročené kroky

Příklady škálování

Následující dotaz vypočítá počet aut, které projedou mýtnou stanicí s třemi mýtnými bránami během tříminutového okna. Tento dotaz lze rozšířit na jednu SU V2.

    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

Pokud chcete pro dotaz použít více jednotek SU, musí být rozdělený vstupní datový proud i dotaz. Vzhledem k tomu, že oddíl datového streamu je nastavený na 3, je možné škálovat následující upravený dotaz až na 3 SU V2:

    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

Při dělení dotazu jsou vstupní události zpracovávány a agregovány v samostatných skupinách oddílů. Výstupní události se také generují pro každou skupinu. Dělení může způsobit neočekávané výsledky, pokud pole GROUP BY není klíčem oddílu ve vstupním datovém streamu. Například pole TollBoothId v předchozím dotazu není klíčem oddílu input1. Výsledkem je, že data z "TollBooth č. 1" je možné rozdělit do několika oddílů.

Stream Analytics zpracuje každý oddíl Input1 samostatně. V důsledku toho se vytvoří více záznamů o počtu aut pro stejnou mýtnou bránu ve stejném klouzavém okně. Pokud není možné změnit vstupní klíč oddílu, můžete tento problém vyřešit přidáním nepartitionové kroku, který agreguje hodnoty mezi oddíly, jak je znázorněno v následujícím příkladu.

    WITH Step1 AS (
        SELECT COUNT(*) AS Count, TollBoothId
        FROM Input1 Partition By PartitionId
        GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute, 3), TollBoothId

Tento dotaz je možné škálovat na 4 SU V2.

Poznámka:

Pokud spojujete dva datové proudy, ujistěte se, že jsou datové proudy rozdělené podle klíče oddílu sloupce, který používáte k vytvoření spojení. Také se ujistěte, že v obou datových proudech máte stejný počet oddílů.

Dosažení vyšší propustnosti ve velkém měřítku

Trapně paralelní úloha je nezbytná, ale nestačí k zajištění vyšší propustnosti ve velkém měřítku. Každý systém úložiště a odpovídající výstup Stream Analytics má různé varianty, jak dosáhnout nejlepší možné propustnosti zápisu. Stejně jako u jakéhokoli scénáře ve velkém měřítku existuje několik problémů, které je možné vyřešit pomocí správných konfigurací. Tato část popisuje konfigurace pro několik běžných výstupů a poskytuje ukázky pro udržování rychlosti příjmu 1 000, 5 000 a 10 000 událostí za sekundu.

Následující pozorování používají úlohu Stream Analytics s bezstavovým (předávacím) dotazem, jednoduchou uživatelem definovanou funkcí JavaScriptu (UDF), která zapisuje do služby Event Hubs, Azure SQL nebo Azure Cosmos DB.

Centra událostí

Rychlost příjmu dat (události za sekundu)	Jednotky streamování	Výstupní prostředky
1 K	1/3	2 TU
5 K	1	6 TU
10 tis.	2	10 TU

Řešení Event Hubs se škáluje lineárně z hlediska jednotek streamování (SU) a propustnosti, což je nejúčinnější a nejvýkonnější způsob, jak analyzovat a streamovat data z Stream Analytics. Úlohy je možné škálovat až na 66 jednotek SU V2, což zhruba znamená zpracování až 400 MB/s nebo 38 biliónů událostí za den.

Azure SQL

Rychlost příjmu dat (události za sekundu)	Jednotky streamování	Výstupní prostředky
1 K	2/3	S3
5 K	3	P4
10 tis.	6	P6

Azure SQL podporuje paralelní zápis známý jako Inherit Partitioning, ale ve výchozím nastavení není povolen. Povolení dědičného dělení v kombinaci s plně paralelním dotazem ale nemusí být dostatečné k dosažení vyšší propustnosti. Propustnost zápisu SQL závisí výrazně na konfiguraci databáze a schématu tabulek. Článek výkonu výstupu SQL obsahuje podrobnější informace o parametrech, které mohou maximalizovat propustnost zápisu. Jak je uvedeno ve výstupu Azure Stream Analytics do služby Azure SQL Database, toto řešení neškálí lineárně jako plně paralelní kanál nad rámec 8 oddílů a může vyžadovat opětovné rozdělení před výstupem SQL (viz INTO). Prémiové jednotky SKU jsou potřeba k udržení vysoké frekvence I/O operací spolu s náklady režie při zálohování logů, ke kterým dochází každých několik minut.

Azure Cosmos DB

Rychlost příjmu dat (události za sekundu)	Jednotky streamování	Výstupní prostředky
1 K	2/3	20 K RU
5 K	4	60 K RU
10 tis.	8	120 K RU

Výstup služby Azure Cosmos DB ze stream Analytics byl aktualizován tak, aby používal nativní integraci na úrovni kompatibility 1.2. Úroveň kompatibility 1.2 umožňuje výrazně vyšší propustnost a snižuje spotřebu RU oproti verzi 1.1, což je výchozí úroveň kompatibility pro nové úlohy. Řešení používá kontejnery Azure Cosmos DB rozdělené na oddíly /deviceId a zbytek řešení se konfiguruje stejně.

Všechny příklady Azure Streaming ve velkém měřítku využívají Event Hubs jako vstup, do kterého je data přiváděna simulací zatížení testovacích klientů. Každá vstupní událost je dokument JSON o velikosti 1 kB, který snadno překládá nakonfigurované rychlosti příjmu dat na rychlost propustnosti (1 MB/s, 5 MB/s a 10 MB/s). Události simulují zařízení IoT odesílající následující data JSON (ve zkrácené podobě) až pro 1 000 zařízení:

{
    "eventId": "b81d241f-5187-40b0-ab2a-940faf9757c0",
    "complexData": {
        "moreData0": 51.3068118685458,
        "moreData22": 45.34076957651598
    },
    "value": 49.02278128887753,
    "deviceId": "contoso://device-id-1554",
    "type": "CO2",
    "createdAt": "2019-05-16T17:16:40.000003Z"
}

Poznámka:

Konfigurace se můžou změnit kvůli různým komponentám používaným v řešení. Pokud chcete přesnější odhad, přizpůsobte si vzorky tak, aby vyhovovaly vašemu scénáři.

Identifikace kritických bodů

K identifikaci kritických bodů v kanálu použijte podokno Metriky v úloze Azure Stream Analytics. Zkontrolujte vstupní a výstupní události kvůli propustnosti a zpoždění vodoznaku nebo nevyřízeným událostem, aby bylo jasné, jestli úloha udržuje krok se vstupní rychlostí. V případě metrik služby Event Hubs vyhledejte omezené požadavky a odpovídajícím způsobem upravte prahové jednotky. V případě metrik Služby Azure Cosmos DB zkontrolujte maximální spotřebované RU/s na rozsah klíčů oddílu v části Propustnost a ujistěte se, že se rozsahy klíčů oddílů rovnoměrně spotřebovávají. V případě Azure SQL DB monitorujte operace Log IO a CPU.

Získání pomoci

Pokud potřebujete další pomoc, vyzkoušejte naši stránku pro otázky Microsoftu pro Azure Stream Analytics.

Další kroky

• Úvod do Azure Stream Analytics
• Začínáme používat službu Azure 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