Verwenden der Abfrageparallelisierung in Azure Stream Analytics

Dieser Artikel veranschaulicht das Nutzen der Parallelisierung in Azure Stream Analytics. Erfahren Sie, wie Sie Stream Analytics-Aufträge durch Konfigurieren der Eingabe in Partitionen und Optimieren der Analysenabfragedefinition skalieren.

Als Voraussetzung sollten Sie mit dem Konzept der Streamingeinheiten vertraut sein, die unter Verstehen und Anpassen von Streamingeinheiten beschrieben werden.

Welche Teile hat ein Stream Analytics-Auftrag?

Eine Stream Analytics-Auftragsdefinition umfasst mindestens eine Streamingeingabe, eine Abfrage und eine Ausgabe. Eingaben sind die Punkte, von denen der Job den Datenstrom liest. Bei der Abfrage wird die Datenstromeingabe transformiert, und der Auftrag sendet die Auftragsergebnisse an die Ausgabe.

Partitionen in Eingaben und Ausgaben

Durch die Partitionierung können Daten basierend auf einem Partitionsschlüssel in Teilmengen unterteilt werden. Wenn Ihre Eingabe (z. B. Event Hubs) durch einen Schlüssel partitioniert wird, wird empfohlen, den Partitionsschlüssel anzugeben, wenn Sie ihrem Stream Analytics-Auftrag eine Eingabe hinzufügen. Bei der Skalierung eines Stream Analytics-Auftrags werden die Partitionen in der Eingabe und Ausgabe genutzt. Mit einem Stream Analytics-Auftrag können unterschiedliche Partitionen parallel genutzt und beschrieben werden, wodurch sich der Durchsatz erhöht.

Eingaben

Alle Azure Stream Analytics-Streamingeingaben können die Partitionierung nutzen: Event Hubs, IoT Hub, Blob Storage, Data Lake Storage Gen2.

Ausgaben

Wenn Sie mit Stream Analytics arbeiten, nutzen Sie die Partitionierung in den folgenden Ausgaben:

• Azure Data Lake Storage
• Azure-Funktionen
• Azure Tabelle
• Blob-Speicher (explizit festlegen des Partitionsschlüssels)
• Azure Cosmos DB (explizit festlegen des Partitionsschlüssels)
• Event Hubs (explizit festlegen des Partitionsschlüssels)
• IoT Hub (explizit festlegen des Partitionsschlüssels)
• Service Bus
• SQL und Azure Synapse Analytics mit optionaler Partitionierung: Weitere Informationen finden Sie auf der Seite Ausgabe an Azure SQL-Datenbank.

Power BI unterstützt keine Partitionierung. Sie können die Eingabe jedoch weiterhin partitionieren, wie in diesem Abschnitt beschrieben.

Weitere Informationen zu den Partitionen finden Sie in den folgenden Artikeln:

• Event Hubs-Features im Überblick
• Datenpartitionierung

Abfrage

Damit ein Auftrag parallel ist, müssen Partitionsschlüssel zwischen allen Eingaben, allen Abfragelogikschritten und allen Ausgaben abgeglichen werden. Die Partitionierung der Abfragelogik wird durch die Schlüssel bestimmt, die für Verknüpfungen und Aggregationen (GROUP BY) verwendet werden. Die letzte Anforderung kann ignoriert werden, wenn die Abfragelogik nicht schlüsselgebunden ist (Projektion, Filter, referentielle Verknüpfungen...).

• Wenn eine Eingabe und eine Ausgabe nach WarehouseId partitioniert werden und die Abfragegruppen nach ProductId ohne WarehouseId, ist der Auftrag nicht parallel.
• Wenn zwei zu verknüpfte Eingaben durch unterschiedliche Partitionsschlüssel (WarehouseId und ProductId) partitioniert werden, ist der Auftrag nicht parallel.
• Wenn ein einzelner Auftrag zwei oder mehr unabhängige Datenflüsse enthält, jeweils mit eigenem Partitionsschlüssel, ist der Auftrag nicht parallel.

Der Auftrag ist nur parallel, wenn alle Eingaben, Ausgaben und Abfrageschritte denselben Schlüssel verwenden.

Hochgradig parallele Aufträge

Ein peinlich paralleler Auftrag stellt das am stärksten skalierbare Szenario in Azure Stream Analytics dar. Er verbindet eine Partition der Eingabe mit einer Instanz der Abfrage und einer Partition der Ausgabe. Für eine solche Parallelität gelten folgende Anforderungen:

• Wenn Ihre Abfragelogik davon abhängig ist, dass derselbe Schlüssel durch dieselbe Abfrageinstanz verarbeitet wird, müssen Sie sicherstellen, dass die Ereignisse in derselben Partition Ihrer Eingabe aufgenommen werden. Bei Event Hubs oder IoT Hub bedeutet dies, dass für die Ereignisdaten der Wert PartitionKey festgelegt sein muss. Alternativ können Sie partitionierte Absender verwenden. Bei Blob Storage, was bedeutet, dass die Ereignisse an denselben Partitionsordner gesendet werden. Beispiel: Von einer Abfrageinstanz werden Daten anhand von „userID“ aggregiert, wobei die Event Hub-Eingabe mit „userID“ als Partitionsschlüssel partitioniert wurde. Wenn es für Ihre Abfragelogik jedoch nicht erforderlich ist, dass derselbe Schlüssel von derselben Abfrageinstanz verarbeitet wird, können Sie diese Anforderung ignorieren. Ein Beispiel für diese Logik wäre eine einfache Auswahl-, Projekt- oder Filterabfrage.

• Als Nächstes müssen Sie die Abfrage partitionieren. Geben Sie für Aufträge mit Kompatibilitätsebene 1.2 oder höher (empfohlen) eine benutzerdefinierte Spalte als Partitionsschlüssel in den Eingabeeinstellungen an, und der Auftrag ist automatisch parallel. Verwenden Sie für Aufträge mit Kompatibilitätsebene 1.0 oder 1.1 PARTITION BY PartitionId in allen Schritten Ihrer Abfrage. Sie können mehrere Schritte ausführen, aber alle müssen durch denselben Schlüssel partitioniert werden.

• Bei den meisten in Stream Analytics unterstützten Ausgaben können Sie die Vorteile der Partitionierung nutzen. Wenn Sie einen Ausgabetyp verwenden, der keine Partitionierung unterstützt, weist Ihr Auftrag keine hochgradige Parallelität auf. Stellen Sie bei Event Hubs-Ausgaben sicher, dass die Partitionsschlüsselspalte auf denselben Partitionsschlüssel wie den in der Abfrage verwendeten Schlüssel festgelegt ist. Weitere Informationen finden Sie unter Ausgabebereich.

• Die Anzahl von Eingabepartitionen muss mit der Anzahl von Ausgabepartitionen identisch sein. Die Blob Storage-Ausgabe kann Partitionen unterstützen und erbt das Partitionierungsschema der Upstream-Abfrage. Wenn Sie einen Partitionsschlüssel für BLOB-Speicher angeben, werden Daten pro Eingabepartition partitioniert, damit das Ergebnis weiterhin vollständig parallel ist. Im Folgenden werden Beispiele für Partitionswerte vorgestellt, die eine vollständig parallele Aufgabe ermöglichen.

  • Acht Event Hub-Eingabepartitionen und acht Event Hub-Ausgabepartitionen
  • Acht Event Hub-Eingabepartitionen und Blob Storage-Ausgabe
  • Acht Event Hub-Eingabepartitionen und Blob Storage-Ausgabe, partitioniert nach einem benutzerdefinierten Feld mit beliebiger Kardinalität
  • Acht Blob Storage-Eingabepartitionen und Blob Storage-Ausgabe
  • Acht Blob Storage-Eingabepartitionen und acht Event Hub-Ausgabepartitionen

In den folgenden Abschnitten werden einige Beispielszenarien für hochgradige Parallelität behandelt.

Einfache Abfrage

• Eingabe: Ein Event Hub mit acht Partitionen
• Ausgabe: Ein Event Hub mit acht Partitionen („Partitionsschlüsselspalte“ muss auf PartitionId gesetzt werden)

Abfrage:

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

Diese Abfrage stellt einen einfachen Filter dar. Daher müssen Sie sich keine Gedanken über die Partitionierung der Eingabe machen, die an den Event Hub gesendet wird. Beachten Sie, dass Aufträge mit einem niedrigeren Kompatibilitätsgrad als 1.2 die PARTITION BY PartitionId-Klausel enthalten müssen, damit die zweite, weiter oben genannte Anforderung erfüllt ist. Für die Ausgabe müssen Sie die Event Hub-Ausgabe in der Aufgabe so konfigurieren, dass der Partitionsschlüssel auf PartitionId festgelegt ist. Eine letzte Prüfung besteht darin, sicherzustellen, dass die Anzahl der Eingabepartitionen mit der Anzahl der Ausgabepartitionen identisch ist.

Abfrage mit einem Gruppierungsschlüssel

• Eingabe: Event Hub mit acht Partitionen
• Ausgabe: Blob Storage

Abfrage:

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

Diese Abfrage enthält einen Gruppierungsschlüssel. Aus diesem Grund müssen die gruppierten Ereignisse an dieselbe Event Hubs-Partition gesendet werden. Da Sie in diesem Beispiel nach TollBoothID gruppieren, sollten Sie sicherstellen, dass TollBoothID sie als Partitionsschlüssel verwendet wird, wenn die Ereignisse an Event Hubs gesendet werden. Anschließend können wir in Azure Stream Analytics PARTITION NACH PartitionId verwenden, um von diesem Partitionsschema zu erben und vollständige Parallelisierung zu aktivieren. Da es sich bei der Ausgabe um Blob Storage handelt, muss laut der vierten Anforderung kein Wert für den Partitionsschlüssel konfiguriert werden.

Beispiel für Szenarien, die nicht* hochgradig parallel sind

Im vorherigen Abschnitt behandelte der Artikel einige hochgradig parallele Szenarien. In diesem Abschnitt werden Szenarien erläutert, in denen nicht alle Anforderungen hinsichtlich hochgradiger Parallelität erfüllt werden.

Nicht übereinstimmende Partitionsanzahl

• Eingabe: Ein Event Hub mit acht Partitionen
• Ausgabe: Ein Event Hub mit 32 Partitionen

Wenn die Anzahl der Eingabepartitionen nicht der Anzahl der Ausgabepartitionen entspricht, weist die Topologie unabhängig von der Abfrage keine hochgradige Parallelität auf. Sie können jedoch trotzdem einen gewissen Grad der Parallelisierung erreichen.

Abfragen mit nicht partitionierter Ausgabe

• Eingabe: Ein Event Hub mit acht Partitionen
• Ausgabe: Power BI

Die Power BI-Ausgabe unterstützt derzeit keine Partitionierung. Daher besteht in diesem Szenario keine hochgradige Parallelität.

Mehrstufige Abfrage mit unterschiedlichen Werten für „PARTITION BY“

• Eingabe: Event Hub mit acht Partitionen
• Ausgabe: Event Hub mit acht Partitionen
• Kompatibilitätsgrad: 1,0 oder 1,1

Abfrage:

    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

Wie Sie sehen, wird im zweiten Schritt TollBoothId als Partitionierungsschlüssel verwendet. Dieser Schritt entspricht nicht dem ersten Schritt, und deshalb muss eine Umschichtung durchgeführt werden.

Mehrstufige Abfrage mit unterschiedlichen Werten für „PARTITION BY“

• Eingabe: Event Hub mit acht Partitionen („Partitionsschlüsselspalte“ nicht festgelegt, Standardwert: „PartitionId“)
• Ausgabe: Event Hub mit acht Partitionen („Partitionsschlüsselspalte“ muss auf „TollBoothId“ gesetzt werden)
• Kompatibilitätsstufe – 1.2 oder höher

Abfrage:

    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

Kompatibilitätsgrad 1,2 oder höher ermöglicht standardmäßig die parallele Abfrageausführung. Beispielsweise wird die Abfrage aus dem vorherigen Abschnitt partitioniert, solange die Spalte "TollBoothId" als Eingabepartitionsschlüssel festgelegt ist. Die PARTITION BY PartitionId-Klausel ist nicht erforderlich.

Berechnen der maximal möglichen Streaming-Einheiten für einen Auftrag

Die Gesamtanzahl der Streamingeinheiten, die ein Stream Analytics-Auftrag verwenden kann, hängt von der Anzahl der Schritte in der Abfrage ab, die für den Auftrag definiert ist, und der Anzahl der Partitionen für jeden Schritt.

Schritte einer Abfrage

Eine Abfrage kann einen oder mehrere Schritte umfassen. Jeder Schritt besteht aus einer Unterabfrage, die mit dem Schlüsselwort WITH definiert wird. Die Abfrage, die sich außerhalb des Schlüsselworts WITH befindet (nur eine Abfrage), wird ebenfalls als Schritt gezählt, beispielsweise die SELECT-Anweisung in der folgenden Abfrage:

Abfrage:

    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

Die Abfrage umfasst zwei Schritte.

Partitionieren eines Schritts

Für die Partitionierung eines Schrittes gelten die folgenden Voraussetzungen:

• Die Eingabequelle muss partitioniert sein.
• Die SELECT -Anweisung der Abfrage muss aus einer partitionierten Eingabequelle lesen.
• Die Abfrage innerhalb des Schritts muss das Schlüsselwort Partition by aufweisen.

Wenn eine Abfrage partitioniert ist, werden die Eingabeereignisse verarbeitet und in separaten Partitionsgruppen aggregiert, und für jede einzelne Gruppe werden Ausgabeereignisse generiert. Wenn Sie ein kombiniertes Aggregat bevorzugen, müssen Sie einen zweiten, nicht partitionierten Schritt zum Aggregieren erstellen.

Berechnen der maximal möglichen Streaming-Einheiten für einen Auftrag

Alle nicht partitionierten Schritte werden zusammen für einen Stream Analytics-Job auf eine Streamingeinheit (SU V2s) hochskaliert. Darüber hinaus können Sie in einem partitionierten Schritt für jede Partition ein SU V2 hinzufügen. In der folgenden Tabelle finden Sie einige Beispiele.

Beispiele für eine Skalierung

Mit der folgenden Abfrage wird die Anzahl der Fahrzeuge berechnet, die in einem Zeitraum von drei Minuten eine Mautstation mit drei Mautstellen passieren. Sie können diese Abfrage bis zu einem SU V2 skalieren.

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

Um weitere SUs für die Abfrage zu verwenden, partitionieren Sie sowohl den Eingabedatenstrom als auch die Abfrage. Da die Datenstrompartition auf 3 festgelegt ist, kann die folgende modifizierte Abfrage auf bis zu 3 SU V2s skaliert werden:

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

Wenn Sie eine Abfrage partitionieren, werden die Eingabeereignisse in separaten Partitionsgruppen verarbeitet und aggregiert. Die Abfrage generiert Output-Ereignisse für jede der Gruppen. Die Partitionierung kann zu unerwarteten Ergebnissen führen, wenn das Feld GROUP BY nicht den Partitionsschlüssel im Eingabedatenstrom enthält. Beispielsweise enthält das Feld TollBoothId in der vorherigen Abfrage nicht den Partitionsschlüssel von Input1. Das Ergebnis ist, dass die Daten von TollBooth #1 auf mehrere Partitionen verteilt werden können.

Stream Analytics verarbeitet jede der Input1-Partitionen separat. Folglich erstellt die Abfrage im selben rollierenden Fenster mehrere Einträge von der Anzahl der Autos für dieselbe Mautstation. Wenn Sie den Eingabepartitionsschlüssel nicht ändern können, beheben Sie dieses Problem, indem Sie einen Nichtpartitionsschritt hinzufügen, um Werte über Partitionen hinweg zu aggregieren, wie im folgenden Beispiel gezeigt:

    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

Sie können diese Abfrage auf 4 SU V2s skalieren.

Erzielen höherer Durchsätze im großen Maßstab

Ein hochgradig paralleler Auftrag ist notwendig, aber nicht ausreichend, um bei Bedarf einen höheren Durchsatz aufrechtzuerhalten. Jedes Speichersystem und seine entsprechende Stream Analytics-Ausgabe enthält Variationen dabei, wie der beste Schreibdurchsatz zu erzielen ist. Wie bei jedem Szenario im großen Maßstab erfordern einige Herausforderungen die richtigen Konfigurationen, um sie zu lösen. In diesem Abschnitt werden Konfigurationen für einige gängige Ausgaben erläutert und Beispiele für die Aufrechterhaltung der Datenerfassungsraten von 1.000, 5.000 und 10.000 Ereignissen pro Sekunde gezeigt.

Die folgenden Beobachtungen verwenden einen Stream Analytics-Auftrag mit statusfreier (Pass-Through) Abfrage, eine einfache benutzerdefinierte JavaScript-Funktion (User Defined Function, UDF), die in Event Hubs, Azure SQL oder Azure Cosmos DB schreibt.

Event Hubs

Die Event Hubs-Lösung lässt sich hinsichtlich der Streamingeinheiten (SU) und des Durchsatzes linear skalieren, was sie zur effizientesten und leistungsstärksten Methode zum Analysieren und Streamen von Daten aus Stream Analytics macht. Sie können Aufträge bis zu 66 SU V2s skalieren, was in etwa auf die Verarbeitung von bis zu 400 MB/s oder 38 Billionen Ereignissen pro Tag übersetzt wird.

Azure SQL

Azure SQL unterstützt paralleles Schreiben, auch Partitionierungsvererbung genannt, was aber nicht standardmäßig aktiviert ist. Allerdings ist das Aktivieren der Partitionierungsvererbung zusammen mit einer vollständig parallelen Abfrage möglicherweise nicht ausreichend, um höhere Durchsätze zu erreichen. Die Schreibdurchsätze von SQL sind signifikant von Ihrer Datenbankkonfiguration und dem Tabellenschema abhängig. Der Artikel SQL-Ausgabeleistung enthält weitere Details zu den Parametern, mit denen sich Ihr Schreibdurchsatz maximieren lässt. Wie im Artikel Azure Stream Analytics-Ausgabe in Azure SQL-Datenbank erwähnt, skaliert diese Lösung nicht linear als vollständig parallele Pipeline über 8 Partitionen hinaus und es kann erforderlich sein, vor der SQL-Ausgabe eine Repartitionierung vorzunehmen (siehe INTO). Premium-SKUs sind erforderlich, um hohe E/A-Raten zusammen mit dem Mehraufwand durch Protokollsicherungen, die alle paar Minuten erfolgen, aufrechtzuerhalten.

Azure Cosmos DB

Die Azure Cosmos DB Ausgabe von Stream Analytics wird aktualisiert, um die native Integration unter Kompatibilitätsebene 1.2 zu nutzen. Kompatibilitätsgrad 1.2 ermöglicht einen wesentlich höheren Durchsatz und verringert den RU-Verbrauch im Vergleich zu 1.1, bei dem es sich um den Standardkompatibilitätsgrad für neue Aufträge handelt. Die Lösung verwendet unter „/deviceId“ partitionierte Azure Cosmos DB-Container, und der Rest der Lösung ist identisch konfiguriert.

Alle Azure-Beispiele zum Streaming im großen Stil verwenden eine Event Hubs-Instanz als Eingabe, die von Last simulierenden Testclients Eingaben erhält. Jedes Eingabeereignis ist ein 1-KB-JSON-Dokument, das problemlos konfigurierte Datenerfassungsraten in Durchsatzraten (1 MB/s, 5 MB/s und 10 MB/s) umsetzt. Ereignisse simulieren ein IoT-Gerät, das die folgenden JSON-Daten (in einer Kurzform) für bis zu 1,000 Geräte sendet:

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

Identifizierung von Engpässen

Verwenden Sie den Bereich „Metriken“ in ihrem Azure Stream Analytics-Auftrag, um Engpässe in Ihrer Pipeline zu identifizieren. Überprüfen Sie Eingabe-/Ausgabeereignisse hinsichtlich des Durchsatzes sowie "Wasserzeichenverzögerung" oder Ereignisse im Rückstand, um festzustellen, ob der Auftrag mit der Eingangsrate Schritt halten kann. Suchen Sie bei Event Hubs-Metriken nach Gedrosselte Anforderungen, und passen Sie die Schwellenwerteinheiten entsprechend an. Überprüfen Sie für Azure Cosmos DB-Metriken Max. genutzte RU/Sek. pro Partitionsschlüsselbereich unter „Durchsatz“, um sicherzustellen, dass Ihre Partitionsschlüsselbereiche gleichmäßig genutzt werden. Überwachen Sie für Azure SQL-DB Protokoll-E/A und CPU.

Hier erhalten Sie Hilfe

Um weitere Hilfe zu erhalten, probieren Sie die Microsoft Q&A-Frageseite für Azure Stream Analytics aus.

Nächste Schritte

• Einführung in Azure Stream Analytics
• Erste Schritte mit Azure Stream Analytics
• Azure Stream Analytics Abfragesprachen-Referenz
• Referenz zur Azure Stream Analytics-Verwaltungs-REST-API