Używanie przetwarzania równoległego zapytań w usłudze Azure Stream Analytics

W tym artykule pokazano, jak korzystać z równoległości w usłudze Azure Stream Analytics. Dowiedz się, jak skalować zadania usługi Stream Analytics, konfigurując partycje wejściowe i dostrajając definicję zapytania analizy.

W ramach wymagań wstępnych warto zapoznać się z pojęciem jednostki przesyłania strumieniowego opisanego w artykule Omówienie i dostosowywanie jednostek przesyłania strumieniowego.

Jakie są części zadania usługi Stream Analytics?

Definicja zadania usługi Stream Analytics zawiera co najmniej jedno dane wejściowe przesyłania strumieniowego, zapytanie i dane wyjściowe. Dane wejściowe to miejsce, z którego zadanie odczytuje strumień danych. Zapytanie służy do przekształcania strumienia wejściowego danych, a dane wyjściowe to miejsce, do którego zadanie wysyła wyniki zadania.

Partycje w danych wejściowych i wyjściowych

Partycjonowanie umożliwia podzielenie danych na podzestawy na podstawie klucza partycji. Jeśli dane wejściowe (na przykład Event Hubs) są partycjonowane przez klucz, zalecamy określenie klucza partycji podczas dodawania danych wejściowych do zadania usługi Stream Analytics. Skalowanie zadania usługi Stream Analytics wykorzystuje partycje w danych wejściowych i wyjściowych. Zadanie usługi Stream Analytics może używać i zapisywać różne partycje równolegle, co zwiększa przepływność.

Dane wejściowe

Wszystkie wejścia przesyłania strumieniowego usługi Azure Stream Analytics mogą korzystać z partycjonowania: Event Hubs, IoT Hub, Blob storage, Data Lake Storage Gen2.

Dane wyjściowe

Podczas pracy z usługą Stream Analytics możesz korzystać z partycjonowania w danych wyjściowych:

• Azure Data Lake Storage
• Azure Functions
• Tabela platformy Azure
• Magazyn danych Blob z możliwością jawnego ustawienia klucza partycji
• Azure Cosmos DB (należy jawnie ustawić klucz partycji)
• Event Hubs (należy jawnie ustawić klucz partycji)
• Usługa IoT Hub (należy jawnie ustawić klucz partycji)
• Magistrala usług
• SQL i Azure Synapse Analytics z opcjonalnym partycjonowaniem: zobacz więcej informacji na stronie Dane wyjściowe do Azure SQL Database.

Usługa Power BI nie obsługuje partycjonowania. Można jednak nadal partycjonować dane wejściowe zgodnie z opisem w tej sekcji.

Aby uzyskać więcej informacji na temat partycji, zobacz następujące artykuły:

• Event Hubs features overview (Omówienie funkcji usługi Event Hubs)
• Data partitioning (Partycjonowanie danych)

Zapytanie

Aby zadanie mogło być wykonane równolegle, klucze partycji muszą być zgodne pomiędzy wszystkimi danymi wejściowymi, całą logiką kroków zapytań oraz wszystkimi danymi wyjściowymi. Partycjonowanie logiki zapytań jest określane przez klucze używane do sprzężeń i agregacji (GROUP BY). Ostatnie wymaganie można zignorować, jeśli logika zapytania nie jest oparta na kluczach (projekcja, filtry, sprzężenia odwołań...).

• Jeśli dane wejściowe i wyjściowe są partycjonowane za pomocą WarehouseId, a zapytanie grupuje według ProductId bez WarehouseId, to zadanie nie jest równoległe.
• Jeśli dwa dane wejściowe do sprzężenia są partycjonowane przez różne klucze partycji (WarehouseId i ProductId), zadanie nie jest równoległe.
• Jeśli co najmniej dwa niezależne przepływy danych znajdują się w jednym zadaniu, każde z własnym kluczem partycji, zadanie nie jest równoległe.

Tylko wtedy, gdy wszystkie dane wejściowe, dane wyjściowe i kroki zapytania używają tego samego klucza, zadanie jest równoległe.

Żenujące zadania równoległe

Żenujące zadanie równoległe to najbardziej skalowalny scenariusz w usłudze Azure Stream Analytics. Łączy jedną partycję danych wejściowych z jednym wystąpieniem zapytania z jedną partycją danych wyjściowych. Ta równoległość ma następujące wymagania:

• Jeśli logika zapytania zależy od tego samego klucza przetwarzanego przez to samo wystąpienie zapytania, upewnij się, że zdarzenia przechodzą do tej samej partycji danych wejściowych. W przypadku usługi Event Hubs lub usługi IoT Hub oznacza to, że dane zdarzenia muszą mieć ustawioną wartość PartitionKey . Alternatywnie można użyć nadawców partycjonowanych. W przypadku przechowywania blobów oznacza to, że zdarzenia są wysyłane do tego samego folderu partycji. Przykładem może być wystąpienie zapytania, które agreguje dane per userID, gdzie centrum zdarzeń wejściowych jest partycjonowane przy użyciu userID jako klucza partycji. Jeśli jednak logika zapytania nie wymaga przetworzenia tego samego klucza przez to samo wystąpienie zapytania, możesz zignorować to wymaganie. Przykładem tej logiki jest proste zapytanie select-project-filter.

• Następnym krokiem jest partycjonowanie zapytania. W przypadku zadań z poziomem zgodności 1.2 lub wyższym (zalecane) można określić kolumnę niestandardową jako klucz partycji w ustawieniach wejściowych, a zadanie będzie przetwarzane równolegle automatycznie. Zadania z poziomem zgodności 1.0 lub 1.1 wymagają użycia identyfikatora PARTITION BY PartitionId we wszystkich krokach zapytania. Dozwolone są wiele kroków, ale wszystkie muszą być podzielone na partycje przy użyciu tego samego klucza.

• Większość danych wyjściowych obsługiwanych w usłudze Stream Analytics może korzystać z partycjonowania. Jeśli używasz typu danych wyjściowych, który nie obsługuje partycjonowania zadania, nie będzie kłopotliwie równoległy. W przypadku danych wyjściowych usługi Event Hubs upewnij się, że kolumna Klucz partycji jest ustawiona na ten sam klucz partycji używany w zapytaniu. Aby uzyskać więcej informacji, zobacz sekcję danych wyjściowych.

• Liczba partycji wejściowych musi być równa liczbie partycji wyjściowych. Dane wyjściowe usługi Blob Storage mogą obsługiwać partycje i dziedziczyć schemat partycjonowania zapytania nadrzędnego. Po określeniu klucza partycji dla usługi Blob Storage dane są partycjonowane na partycję wejściową, co powoduje, że wynik jest nadal w pełni równoległy. Oto przykłady wartości partycji, które zezwalają na w pełni równoległe zadanie:

  • Osiem partycji wejściowych centrum zdarzeń i osiem partycji wyjściowych centrum zdarzeń
  • Osiem partycji wejściowych w centrum zdarzeń i dane wyjściowe w magazynie danych blob
  • Osiem partycji wejściowych Event Hub i danych wyjściowych Blob Storage podzielonych według pola niestandardowego z dowolną kardynalnością
  • Osiem partycji wejściowych magazynu obiektów blob i danych wyjściowych magazynu obiektów blob
  • Osiem partycji wejściowych do magazynu blob i osiem partycji wyjściowych do centrum zdarzeń

W poniższych sekcjach omówiono kilka przykładowych scenariuszy, które są trywialnie równoległe.

Proste zapytanie

• Dane wejściowe: centrum zdarzeń z ośmioma partycjami
• Dane wyjściowe: centrum zdarzeń z ośmioma partycjami (kolumna "Klucz partycji" musi być ustawiona na wartość PartitionId)

Zapytanie:

    --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

To zapytanie jest prostym filtrem. W związku z tym nie musimy martwić się o partycjonowanie danych wejściowych wysyłanych do centrum zdarzeń. Zwróć uwagę, że zadania z poziomem zgodności przed 1.2 muszą zawierać klauzulę PARTITION BY PartitionId , więc spełnia wymagania nr 2 z wcześniejszej wersji. W przypadku danych wyjściowych musimy skonfigurować wyjściowe Event Hub w zadaniu tak, aby klucz partycji był ustawiony na PartitionId. Ostatnim sprawdzeniem jest upewnienie się, że liczba partycji wejściowych jest równa liczbie partycji wyjściowych.

Wykonywanie zapytań przy użyciu klucza grupowania

• Dane wejściowe: Event Hub z ośmioma partycjami
• Dane wyjściowe: Blob Storage

Zapytanie:

    --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

To zapytanie ma klucz grupowania. W związku z tym zdarzenia zgrupowane razem muszą być wysyłane do tej samej partycji usługi Event Hubs. Ponieważ w tym przykładzie grupujemy według identyfikatora TollBoothID, upewnij się, że TollBoothID jest on używany jako klucz partycji, gdy zdarzenia są wysyłane do usługi Event Hubs. Następnie w usłudze Azure Stream Analytics można użyć PARTITION BY PartitionId, aby skorzystać z tego schematu partycji i włączyć pełną równoległość. Ponieważ dane wyjściowe są magazynem obiektów blob, nie musimy martwić się o skonfigurowanie wartości klucza partycji zgodnie z wymaganiami nr 4.

Przykład scenariuszy, które nie są* żenujące równoległe

W poprzedniej sekcji artykuł obejmował niektóre kłopotliwe scenariusze równoległe. W tej sekcji dowiesz się więcej na temat scenariuszy, które nie spełniają wszystkich wymagań, aby były naturalnie równoległe.

Niezgodna liczba partycji

• Dane wejściowe: centrum zdarzeń z ośmioma partycjami
• Wynik: centrum wydarzeń z 32 partycjami

Jeśli liczba partycji wejściowych nie jest zgodna z liczbą partycji wyjściowych, topologia nie jest kłopotliwie równoległa niezależnie od zapytania. Jednak nadal możemy uzyskać pewien stopień równoległości.

Wykonywanie zapytań przy użyciu danych wyjściowych bez partycjonowania

• Dane wejściowe: centrum zdarzeń z ośmioma partycjami
• Dane wyjściowe: Power BI

Dane wyjściowe usługi Power BI nie obsługują obecnie partycjonowania. W związku z tym ten scenariusz nie jest trywialnie równoległy.

Zapytanie wieloetapowe z różnymi wartościami dla PARTITION BY

• Dane wejściowe: Event Hub z ośmioma partycjami
• Dane wyjściowe: Event Hub z ośmioma partycjami
• Poziom zgodności: 1.0 lub 1.1

Zapytanie:

    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 widać, drugi krok używa TollBoothId jako klucza partycjonowania. Ten krok nie jest taki sam jak pierwszy krok i dlatego wymaga przetasowania.

Zapytanie wieloetapowe z różnymi wartościami PARTITION BY

• Dane wejściowe: Centrum zdarzeń z ośmioma partycjami ("Kolumna klucza partycji" nie jest ustawiona, wartość domyślna to "PartitionId")
• Dane wyjściowe: Centrum zdarzeń z ośmioma partycjami (Kolumna klucza partycji musi być ustawiona na "TollBoothId")
• Poziom zgodności — 1.2 lub nowszy

Zapytanie:

    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

Poziom zgodności 1.2 lub nowszy domyślnie umożliwia równoległe wykonywanie zapytań. Na przykład zapytanie z poprzedniej sekcji zostanie podzielone na partycje, o ile kolumna "TollBoothId" jest ustawiona jako klucz partycji wejściowej. Klauzula PARTITION BY PartitionId nie jest wymagana.

Obliczanie maksymalnej liczby jednostek strumieniowania zadania

Łączna liczba jednostek przesyłania strumieniowego, które mogą być używane przez zadanie usługi Stream Analytics, zależy od liczby kroków w zapytaniu zdefiniowanym dla zadania i liczby partycji dla każdego kroku.

Kroki w zapytaniu

Zapytanie może zawierać jeden lub wiele kroków. Każdy krok jest podzapytaniem zdefiniowanym przez słowo kluczowe WITH . Zapytanie spoza słowa kluczowego WITH (tylko jedno zapytanie) jest również liczone jako krok, na przykład instrukcja SELECT w następującym zapytaniu:

Zapytanie:

    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

To zapytanie ma dwa kroki.

Podzielić krok

Podział kroku wymaga spełnienia następujących warunków:

• Źródło danych wejściowych musi być podzielone na partycje.
• Instrukcja SELECT zapytania musi odczytywać dane z partycjonowanego źródła danych wejściowych.
• Zapytanie w kroku musi mieć słowo kluczowe PARTITION BY .

Gdy zapytanie jest partycjonowane, zdarzenia wejściowe są przetwarzane i agregowane w oddzielnych grupach partycji, a zdarzenia wyjściowe są generowane dla każdej z grup. Jeśli chcesz stworzyć łączony agregat, musisz utworzyć drugi krok bez partycji do agregacji.

Obliczanie maksymalnej liczby jednostek przesyłania strumieniowego dla zadania

Wszystkie kroki niepartycyjne mogą być łącznie skalowane do jednej jednostki przesyłowej (SU V2) dla zadania Stream Analytics. Ponadto można dodać jedną jednostkę SU V2 dla każdej partycji w kroku partycjonowanym. W poniższej tabeli przedstawiono kilka przykładów .

Przykłady skalowania

Poniższe zapytanie oblicza liczbę samochodów w trzyminutowym oknie przejeżdżających przez stację poboru opłat, która ma trzy punkty poboru opłat. To zapytanie można skalować w górę do jednego SU V2.

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

Aby użyć większej liczby jednostek usługowych (SU) dla zapytania, zarówno strumień danych wejściowych, jak i zapytanie muszą być podzielone na części. Ponieważ partycja strumienia danych jest ustawiona na 3, następujące zmodyfikowane zapytanie można skalować w górę do 3 jednostek SU V2:

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

Gdy zapytanie jest partycjonowane, zdarzenia wejściowe są przetwarzane i agregowane w oddzielnych grupach partycji. Zdarzenia wyjściowe są również generowane dla każdej z grup. Partycjonowanie może spowodować nieoczekiwane wyniki, gdy pole GROUP BY nie jest kluczem partycji w strumieniu danych wejściowych. Na przykład pole TollBoothId w poprzednim zapytaniu nie jest kluczem partycji Input1. Wynikiem jest to, że dane z TollBooth #1 mogą być rozłożone w wielu partycjach.

Każda z partycji Input1 będzie przetwarzana oddzielnie przez usługę Stream Analytics. W rezultacie zostanie utworzonych wiele rekordów liczenia samochodów dla tego samego punktu poboru opłat w tym samym oknie przesunięcia. Jeśli nie można zmienić klucza partycji wejściowej, ten problem można rozwiązać, dodając krok niepartycyjny do agregacji wartości między partycjami, jak w poniższym przykładzie:

    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

To zapytanie można skalować do czterech jednostek SU V2.

Osiąganie większej przepływności na dużą skalę

Żenujące zadanie równoległe jest konieczne, ale nie wystarczające do utrzymania wyższej przepływności na dużą skalę. Każdy system magazynowania i odpowiadające mu dane wyjściowe usługi Stream Analytics mają różnice w sposobie osiągnięcia najlepszej możliwej przepływności zapisu. Podobnie jak w przypadku dowolnego scenariusza na dużą skalę, istnieją pewne wyzwania, które można rozwiązać przy użyciu odpowiednich konfiguracji. W tej sekcji omówiono konfiguracje dla kilku typowych danych wyjściowych i przedstawiono przykłady umożliwiające utrzymanie współczynników pozyskiwania 1 K, 5 K i 10 K na sekundę.

Poniższe obserwacje wykorzystują zadanie usługi Stream Analytics z bezstanowym (przepustowym) zapytaniem oraz podstawową funkcję zdefiniowaną przez użytkownika w języku JavaScript (UDF), która zapisuje dane do usługi Event Hubs, Azure SQL lub Azure Cosmos DB.

Centra zdarzeń

Rozwiązanie Event Hubs skaluje liniowo pod względem jednostek przesyłania strumieniowego (SU) i przepływności, dzięki czemu jest to najbardziej wydajny i wydajny sposób analizowania i przesyłania strumieniowego danych z usługi Stream Analytics. Zadania można skalować w górę do 66 SU V2s, co w przybliżeniu przekłada się na przetwarzanie do 400 MB/s lub 38 bilionów zdarzeń dziennie.

Azure SQL

Usługa Azure SQL obsługuje równoległe pisanie, nazywane Dziedziczone partycjonowanie, ale nie jest domyślnie włączone. Jednak włączenie funkcji Dziedziczenie partycjonowania wraz z w pełni równoległym zapytaniem może nie być wystarczające do osiągnięcia większej przepływności. Przepływność zapisu SQL zależy znacznie od konfiguracji bazy danych i schematu tabeli. Artykuł Wydajność danych wyjściowych SQL zawiera więcej szczegółów na temat parametrów, które mogą zmaksymalizować przepływność zapisu. Jak wspomniano w artykule Dotyczącym danych wyjściowych usługi Azure Stream Analytics w usłudze Azure SQL Database , to rozwiązanie nie jest skalowane liniowo jako w pełni równoległy potok poza 8 partycjami i może wymagać ponownej partycjonowania przed danymi wyjściowymi SQL (zobacz INTO). Jednostki SKU w warstwie Premium są potrzebne do utrzymania wysokiego tempa operacji we/wy wraz z obciążeniem związanym z wykonywaniem kopii zapasowych dzienników co kilka minut.

Azure Cosmos DB

Dane wyjściowe usługi Azure Cosmos DB z usługi Stream Analytics zostały zaktualizowane w celu korzystania z natywnej integracji na poziomie zgodności 1.2. Poziom zgodności 1.2 zapewnia znacznie większą przepływność i zmniejsza zużycie jednostek RU w porównaniu z 1,1, co jest domyślnym poziomem zgodności dla nowych zadań. Rozwiązanie korzysta z kontenerów usługi Azure Cosmos DB podzielonych na partycje na /deviceId, a reszta rozwiązania jest identycznie skonfigurowana.

Wszystkie przykłady przesyłania strumieniowego na dużą skalę platformy Azure używają usługi Event Hubs jako danych wejściowych, które są przekazywane przez obciążenie symulujące klientów testowych. Każde zdarzenie wejściowe to dokument JSON o rozmiarze 1 KB, który łatwo tłumaczy skonfigurowane współczynniki pozyskiwania na współczynniki przepływności (1 MB/s, 5 MB/s i 10 MB/s). Zdarzenia symulują urządzenie IoT wysyłające następujące dane JSON (w skróconej postaci) dla maksymalnie 1000 urządzeń:

{
    "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"
}

Identyfikowanie wąskich gardeł

Użyj okienka Metryki w zadaniu Azure Stream Analytics, aby zidentyfikować wąskie gardła w potoku danych. Przejrzyj Zdarzenia Wejścia/Wyjścia pod kątem przepływności oraz "Opóźnienia znaku wodnego" lub Zdarzeń z zaległościami, aby sprawdzić, czy zadanie nadąża za szybkością wprowadzania. W przypadku metryk usługi Event Hubs poszukaj ograniczonych żądań i dostosuj odpowiednio jednostki progowe. W przypadku metryk usługi Azure Cosmos DB sprawdź Maksymalnie zużywane jednostki RU/s na zakres kluczy partycji w obszarze Przepustowość, aby upewnić się, że zakresy kluczy partycji są równomiernie wykorzystywane. W przypadku usługi Azure SQL DB monitoruj Log IO i CPU.

Uzyskaj pomoc

Aby uzyskać dalszą pomoc, wypróbuj stronę pytań i odpowiedzi firmy Microsoft dotyczącą usługi Azure Stream Analytics.

Następne kroki

• Wprowadzenie do usługi Azure Stream Analytics
• Get started using Azure Stream Analytics (Rozpoczynanie pracy z usługą Azure Stream Analytics)
• Azure Stream Analytics Query Language Reference (Dokumentacja dotycząca języka zapytań usługi Azure Stream Analytics)
• Azure Stream Analytics Management REST API Reference (Dokumentacja interfejsu API REST zarządzania usługą Azure Stream Analytics)