Zpracování streamů s využitím Azure Stream Analytics

Tato referenční architektura ukazuje kompletní kanál zpracování streamování . Kanál ingestuje data ze dvou zdrojů, koreluje záznamy ve dvou datových proudech a vypočte klouzavý průměr v časovém intervalu. Výsledky jsou uloženy pro další analýzu.

A referenční implementace této architektury jsou k dispozici v GitHub.

Scénář: společnost taxislužby shromažďuje data o každé taxislužbyové cestě. V tomto scénáři předpokládáme, že existují dvě samostatná zařízení odesílající data. Taxislužby má měřič, který odesílá informace o každé jízdní — době, vzdálenosti a vyzvednutí a umístění dropoff. Samostatné zařízení přijímá platby od zákazníků a odesílá data o tarifech. Společnost taxislužby chce vypočítat průměrnou špičku na ujeté km v reálném čase, aby bylo možné sledovat trendy.

Architektura

Tato architektura se skládá z následujících součástí.

Zdroje dat. V této architektuře existují dva zdroje dat, které generují datové proudy v reálném čase. První datový proud obsahuje jízdní informace a druhý obsahuje informace o tarifech. Referenční architektura zahrnuje simulované generátory dat, který čte ze sady statických souborů a vkládá data do Event Hubs. V reálné aplikaci by zdroje dat byly zařízení nainstalovaná v taxislužby kabinách.

Event Hubs Azure. Event Hubs je služba pro přijímání událostí. Tato architektura používá dvě instance centra událostí, jednu pro každý zdroj dat. Každý zdroj dat pošle datový proud dat do přidruženého centra událostí.

Azure Stream Analytics. Stream Analytics je modul pro zpracování událostí. Úloha Stream Analytics načte datové proudy ze dvou Center událostí a provede zpracování streamu.

Cosmos DB. výstupem z Stream Analytics úlohy je řada záznamů, které jsou zapsány jako dokumenty JSON do databáze Cosmos DB dokumentů.

Microsoft Power BI. Power BI je sada nástrojů pro obchodní analýzy, která slouží k analýze dat pro obchodní přehledy. v této architektuře načte data z Cosmos DB. To umožňuje uživatelům analyzovat kompletní sadu historických dat, která se shromažďují. výsledky můžete také streamovat přímo z Stream Analytics do Power BI pro zobrazení dat v reálném čase. Další informace najdete v článku o streamování v reálném čase v Power BI.

Azure monitor. Azure monitor shromažďuje metriky výkonu o službách Azure nasazených v řešení. Díky vizualizaci na řídicím panelu můžete získat přehled o stavu řešení.

Přijímání dat

Pro simulaci zdroje dat tato referenční architektura používá datovou sadu dat New York City taxislužby data ^[1]. Tato datová sada obsahuje data o cestách taxislužby v New Yorku po čtyřech letech (2010 – 2013). Obsahuje dva typy záznamů: jízdní data a data tarifů. Jízdní data zahrnují dobu trvání cesty, vzdálenost cest a místo vyzvednutí a dropoff. Data tarifů zahrnují výši tarifů, daní a tipů. Mezi společná pole v obou typech záznamů patří číslo Medallion, licence pro napadení a ID dodavatele. Společně tato tři pole jednoznačně identifikují taxislužby a ovladač. Data jsou uložena ve formátu CSV.

[1] Donovan, Brian; Work, Dan (2016): New Yorku City taxislužby Trip data (2010-2013). Univerzita Illinois na Urbana – Champaign. https://doi.org/10.13012/J8PN93H8

Generátor dat je aplikace .NET Core, která načte záznamy a odesílá je do Azure Event Hubs. Generátor odesílá jízdní data ve formátu JSON a data tarifů ve formátu CSV.

Event Hubs používá k segmentaci dat oddíly . Oddíly umožňují příjemci číst jednotlivé oddíly paralelně. Když odesíláte data do Event Hubs, můžete klíč oddílu zadat explicitně. V opačném případě se záznamy přiřazují oddílům v kruhovém dotazování.

V tomto konkrétním scénáři by data dat a tarifů měly končit stejným ID oddílu pro daný taxislužby CAB. To umožňuje Stream Analytics použít stupeň paralelismu při korelaci dvou datových proudů. Záznam v oddílu n jízdních dat bude odpovídat záznamu v oddílu n dat tarifů.

V generátoru dat má model Common data model pro oba typy záznamů PartitionKey vlastnost, která je zřetězením Medallion , HackLicense a VendorId .

public abstract class TaxiData
{
    public TaxiData()
    {
    }

    [JsonProperty]
    public long Medallion { get; set; }

    [JsonProperty]
    public long HackLicense { get; set; }

    [JsonProperty]
    public string VendorId { get; set; }

    [JsonProperty]
    public DateTimeOffset PickupTime { get; set; }

    [JsonIgnore]
    public string PartitionKey
    {
        get => $"{Medallion}_{HackLicense}_{VendorId}";
    }

Tato vlastnost slouží k poskytnutí explicitního klíče oddílu při posílání do Event Hubs:

using (var client = pool.GetObject())
{
    return client.Value.SendAsync(new EventData(Encoding.UTF8.GetBytes(
        t.GetData(dataFormat))), t.PartitionKey);
}

Zpracování streamů

úloha zpracování datového proudu je definována pomocí SQL dotazu s několika různými kroky. První dva kroky jednoduše vybere záznamy ze dvou vstupních proudů.

WITH
Step1 AS (
    SELECT PartitionId,
           TRY_CAST(Medallion AS nvarchar(max)) AS Medallion,
           TRY_CAST(HackLicense AS nvarchar(max)) AS HackLicense,
           VendorId,
           TRY_CAST(PickupTime AS datetime) AS PickupTime,
           TripDistanceInMiles
    FROM [TaxiRide] PARTITION BY PartitionId
),
Step2 AS (
    SELECT PartitionId,
           medallion AS Medallion,
           hack_license AS HackLicense,
           vendor_id AS VendorId,
           TRY_CAST(pickup_datetime AS datetime) AS PickupTime,
           tip_amount AS TipAmount
    FROM [TaxiFare] PARTITION BY PartitionId
),

Další krok spojí dva vstupní proudy, abyste vybrali vyhovující záznamy z každého streamu.

Step3 AS (
  SELECT tr.TripDistanceInMiles,
         tf.TipAmount
    FROM [Step1] tr
    PARTITION BY PartitionId
    JOIN [Step2] tf PARTITION BY PartitionId
      ON tr.PartitionId = tf.PartitionId
     AND tr.PickupTime = tf.PickupTime
     AND DATEDIFF(minute, tr, tf) BETWEEN 0 AND 15
)

Tento dotaz spojuje záznamy se sadou polí, které jednoznačně identifikují odpovídající záznamy ( PartitionId a PickupTime ).

V Stream Analytics se spojení dočasná, což znamená, že záznamy jsou propojeny v rámci určitého časového období. V opačném případě může úloha potřebovat počkat neomezenou dobu na shodu. Funkce DateDiff určuje, jak daleko můžou být v čase pro shodu oddělené dva odpovídající záznamy.

Poslední krok v úloze vypočítá průměrnou špičku za kilometr a seskupí se skákající oknem po dobu 5 minut.

SELECT System.Timestamp AS WindowTime,
       SUM(tr.TipAmount) / SUM(tr.TripDistanceInMiles) AS AverageTipPerMile
  INTO [TaxiDrain]
  FROM [Step3] tr
  GROUP BY HoppingWindow(Duration(minute, 5), Hop(minute, 1))

Stream Analytics poskytuje několik funkcí okna. Okno skákající se posune v čase o pevné období, v tomto případě 1 minutu na segment směrování. Výsledkem je vypočítat klouzavý průměr za posledních 5 minut.

v níže uvedené architektuře jsou do Cosmos DB uloženy pouze výsledky Stream Analytics úlohy. V případě scénáře s velkými objemy dat zvažte také použití Event Hubs Capture k uložení dat nezpracované události do úložiště objektů BLOB v Azure. Uchovávání nezpracovaných dat vám umožní v pozdější době spouštět dávkové dotazy v historických datech, aby bylo možné z těchto dat odvodit nové poznatky.

Aspekty zabezpečení

Event Hubs

Kapacita propustnosti Event Hubs se měří v jednotkách propustnosti. Centrum událostí můžete automaticky škálovat tím, že povolíte Automatickéškálování, které automaticky škáluje jednotky propustnosti na základě provozu až po nakonfigurované maximum.

Stream Analytics

V případě Stream Analytics se v jednotkách streamování měří výpočetní prostředky přidělené úloze. Stream Analytics úlohy se nejlépe škálují, pokud je možné úlohu paralelně provést. Tímto způsobem může Stream Analytics distribuovat úlohu mezi několik výpočetních uzlů.

Pro Event Hubs vstup použijte PARTITION BY klíčové slovo pro rozdělení úlohy Stream Analytics. Data budou rozdělena na podmnožiny na základě oddílů Event Hubs.

Funkce oken a dočasná spojení vyžadují další SU. Pokud je to možné, použijte, PARTITION BY aby se každý oddíl zpracoval samostatně. Další informace najdete v tématu pochopení a úprava jednotek streamování.

Pokud není možné paralelizovat celou úlohu Stream Analytics, zkuste úlohu rozdělit do několika kroků, počínaje jedním nebo několika paralelními kroky. V takovém případě je možné první postup spustit paralelně. Například v této referenční architektuře:

• Kroky 1 a 2 jsou jednoduché SELECT příkazy, které v rámci jednoho oddílu vyberou záznamy.
• Krok 3 provádí spojování oddílů mezi dvěma vstupními proudy. Tento krok využívá skutečnost, že odpovídající záznamy sdílejí stejný klíč oddílu, a proto je zaručeno, že mají v každém vstupním datovém proudu stejné ID oddílu.
• Krok 4 agreguje napříč všemi oddíly. Tento krok nejde rozparalelně provést.

Pomocí diagramu Stream Analytics úlohy můžete zjistit, kolik oddílů je přiřazeno k jednotlivým krokům v rámci úlohy. Diagram úloh pro tuto referenční architekturu znázorňuje následující diagram:

Cosmos DB

Kapacita propustnosti pro Cosmos DB se měří v jednotkách žádostí (RU). Pokud chcete škálovat kontejner Cosmos DB za posledních 10 000 RU, musíte při vytváření kontejneru zadat klíč oddílu a do každého dokumentu zahrnout klíč oddílu.

V této referenční architektuře se nové dokumenty vytvářejí jenom jednou za minutu (interval intervalu se segmentem směrování), takže požadavky na propustnost jsou poměrně nízké. Z tohoto důvodu není v tomto scénáři potřeba přiřazovat klíč oddílu.

Aspekty monitorování

U jakéhokoli řešení zpracování datových proudů je důležité monitorovat výkon a stav systému. Azure Monitor shromažďuje metriky a diagnostické protokoly pro služby Azure použité v architektuře. Azure Monitor je součástí platformy Azure a nevyžaduje ve vaší aplikaci žádný další kód.

Kterýkoli z následujících varovných signálů indikuje, že byste měli škálovat na více velikosti příslušného prostředku Azure:

• Event Hubs omezování požadavků nebo se blíží denní kvótě zpráv.
• Úloha Stream Analytics konzistentně využívá více než 80 % přidělených jednotek streamování (SU).
• Cosmos Databáze začne ohrožovat požadavky.

Referenční architektura zahrnuje vlastní řídicí panel, který se nasazovat do Azure Portal. Po nasazení architektury můžete řídicí panel zobrazit tak, že otevřete Azure Portal a vyberete TaxiRidesDashboard ze seznamu řídicích panelů. Další informace o vytváření a nasazování vlastních řídicích panelů v Azure Portal tématu Vytváření řídicích panelů Azure prostřednictvím kódu programu.

Následující obrázek znázorňuje řídicí panel po Stream Analytics úloha běžela přibližně hodinu.

Panel v levém dolním rohu ukazuje, že spotřeba SU pro Stream Analytics se během prvních 15 minut snižuje a pak se snižuje. Tento vzor je typický, protože úloha dosáhne stabilního stavu.

Všimněte si Event Hubs je omezení požadavků, které se zobrazují v pravém horním panelu. Občasný problém s omezením není problém, protože klientská sada SDK Event Hubs automaticky znovu spustí pokus, když obdrží chybu omezení. Pokud se však zobrazí konzistentní chyby omezování, znamená to, že centrum událostí potřebuje více jednotek propustnosti. Následující graf ukazuje testovací běh pomocí Event Hubs automatické nafouknutí, která podle potřeby automaticky škáluje jednotky propustnosti.

Automatické nafouknutí bylo povoleno přibližně ve 18:35. Můžete vidět pokles požadavků s omezením, protože Event Hubs automaticky škáloval až na 3 jednotky propustnosti.

Zajímavé je, že to mělo vedlejší efekt zvýšení využití SU v Stream Analytics úlohu. Omezováním se Event Hubs uměle snížit rychlost příjmu dat pro Stream Analytics úlohy. Ve skutečnosti je běžné, že řešení jednoho kritického místa výkonu odhalí další. V tomto případě se problém vyřešil přidělením dalších SU Stream Analytics úlohy.

Důležité informace o nákladech

K odhadu nákladů použijte cenovou kalkulačku Azure. Tady jsou některé aspekty služeb používaných v této referenční architektuře.

Azure Stream Analytics

Azure Stream Analytics cena za počet jednotek streamování (0,11 USD za hodinu) požadovaných ke zpracování dat do služby.

Stream Analytics zpracování dat v reálném čase nebo v malých objemech dat může být nákladné. Pro tyto případy použití zvažte použití Azure Functions Logic Apps k přesunu dat z Azure Event Hubs do úložiště dat.

Azure Event Hubs a Azure Cosmos DB

Informace o nákladech na Azure Event Hubs a Cosmos DB najdete v tématu Aspekty nákladů v referenční architektuře Azure Databricks datových proudů.

Nasazení řešení

Pokud chcete nasadit a spustit referenční implementaci, postupujte podle kroků v GitHub readme.

Důležité informace o DevOps

• Vytvořte samostatné skupiny prostředků pro produkční, vývojová a testovací prostředí. Samostatné skupiny prostředků usnadňují správu nasazení, odstraňování testovacích nasazení a přiřazování přístupových práv.

• Pomocí Azure Resource Manager nasaďte prostředky Azure podle procesu infrastruktury jako kódu (IaC). Díky šablonám je automatizace nasazení pomocí Azure DevOps Servicesnebo jiných řešení CI/CD jednodušší.

• Dejte každou úlohu do samostatné šablony nasazení a uložte prostředky do systémů správy zdrojového kódu. Šablony můžete nasadit společně nebo jednotlivě v rámci procesu CI/CD, což usnadňuje proces automatizace.

  V této architektuře se Azure Event Hubs, Log Analytics a Cosmos DB identifikuje jako jedna úloha. Tyto prostředky jsou součástí jedné šablony ARM.

• Zvažte fázování úloh. Nasazení do různých fází a spuštění ověřovacích kontrol v každé fázi před přechodem do další fáze Tímto způsobem můžete nabízení aktualizací do produkčních prostředí vysoce řízeným způsobem a minimalizovat neočekávané problémy s nasazením.

• Zvažte Azure Monitor k analýze výkonu kanálu zpracování datových proudů. Další informace najdete v tématu Monitorování Azure Databricks.

Další informace najdete v části DevOps v tématu Microsoft Azure Well-Architected Framework.

Související prostředky

Možná si budete chtít prohlédněte následující příklady scénářů Azure, které demonstrují konkrétní řešení s využitím některých stejných technologií:

• IoT a analýzy dat ve stavebnictví
• Zjišťování možných podvodů v reálném čase