App braucht Event-Streaming mit hohem Volumen? Event Hub erstellen, Ereignisse senden und lesen

#event-hubs #streaming #partitions #consumer-groups

Nutze Event Hubs für einen kontinuierlichen Ereignisstrom, etwa Sensormesswerte, Anwendungstelemetrie oder Klickaktivitäten. Service Bus eignet sich für zuverlässige Aufgabenverarbeitung, Event Grid verteilt Benachrichtigungen und Event Hubs hält einen Stream bereit, den unabhängige Leser konsumieren und erneut lesen können.

Dieses Lab sendet zehn Sensormesswerte, um die Übung klein zu halten. Es zeigt den Streaming-Ablauf, keinen Lasttest mit hohem Volumen. Du erstellst Ressourcen im Portal und führst eine vorbereitete Datei in Cloud Shell aus. Eine VM oder lokale Installation ist nicht nötig. Die bisherigen Service-Bus- und Event-Grid-Ressourcen werden nicht benötigt.

Namespace und Event Hub erstellen

Suche im Azure-Portal nach Event Hubs → Create:

Subscription: CloudTrips TEST
Resource group: rg-cloudtrips-eventhubs-test-weu
Namespace name: evhns-ctstreamweu
Region: West Europe
Pricing tier: Standard
Throughput units: 1
Auto-inflate: Disabled

Verwende einen global eindeutigen Namespace-Namen; ergänze bei Bedarf ein Suffix. Lass öffentlichen Netzwerkzugriff für das Lab aktiviert. Wähle Standard, damit der nächste Trip Capture nutzen kann. Prüfe die Kostenschätzung: Bereitgestellte Durchsatzkapazität wird auch ohne gesendete Ereignisse berechnet.

Öffne nach der Bereitstellung den Namespace → Entities → Event Hubs → + Event Hub:

Name: telemetry
Partition count: 2
Cleanup policy: Delete
Retention time: 1 day
Capture: Off

Erstelle den Event Hub und prüfe unter telemetry → Consumer groups, dass die integrierte Gruppe $Default vorhanden ist. Wir verwenden diese Gruppe im Lab; eine benutzerdefinierte Consumer Group ist nicht nötig.

Begriff	Bedeutung in diesem Lab
Namespace	Enthält Event Hubs und die gemeinsame Durchsatzkapazität.
Event Hub	Der Telemetriestream.
Partition	Ein geordneter Abschnitt des Streams. Partitionen ermöglichen die Verteilung der Verarbeitung auf mehrere Leser.
Consumer Group	Eine unabhängige Lesesicht für eine Anwendung, etwa ein Dashboard oder ein Archivierungsprogramm.
Retention	Zeitraum, in dem Ereignisse zum Lesen und erneuten Lesen verfügbar bleiben. Lesen löscht sie nicht.

Eine Consumer Group speichert den Lesefortschritt nicht automatisch dauerhaft. Ein Checkpoint hält die zuletzt verarbeitete Position fest, normalerweise in einem separaten Speicher. Für einfaches erneutes Lesen verzichtet diese kurze Demo auf Checkpoints. Über Partitionsgrenzen hinweg gibt es keine globale Reihenfolgegarantie.

Event Hub telemetry mit zwei Partitionen und seiner Aufbewahrungskonfiguration.

Datenzugriff zuweisen

Weise am Namespace → Access control (IAM) → Add role assignment dem Benutzer für Cloud Shell die Rolle Azure Event Hubs Data Owner zu. Deine Owner-Rolle der Subscription erlaubt die Ressourcenverwaltung, ersetzt aber diese Datenzugriffsrolle nicht. Warte einige Minuten auf die Übernahme einer neuen Zuweisung.

Die Rolle erlaubt dem Lab-Benutzer das Senden und Empfangen. Produktive Anwendungen können die engeren Rollen Azure Event Hubs Data Sender oder Azure Event Hubs Data Receiver verwenden.

Zehn Ereignisse senden und lesen

Lade das Event-Hubs-Beispiel herunter. Öffne im Azure-Portal Cloud Shell → Bash. Eine Sitzung ohne Storage-Konto reicht aus. Lade event-hubs-demo.py über Upload/Download files → Upload hoch.

Führe den folgenden Block aus. Ersetze evhns-ctstreamweu, falls dein Namespace anders heißt. Fordert die Kontoprüfung eine Anmeldung, führe az login --use-device-code aus, melde dich mit dem oben berechtigten Benutzer an und starte den Block erneut.

cd "$HOME"
az account show --output table
az account set --subscription 'CloudTrips TEST'

python3 -m venv /tmp/ct-eventhubs
source /tmp/ct-eventhubs/bin/activate
python -m pip install 'azure-eventhub>=5.12,<6' 'azure-identity>=1.17,<2' 'aiohttp>=3.9,<4'

python event-hubs-demo.py --namespace evhns-ctstreamweu --consumer-group '$Default'

Das Beispiel verwendet deine Azure-CLI-Anmeldung und eine verschlüsselte WebSocket-Verbindung. Das asynchrone SDK nutzt aiohttp, das der Installationsbefehl mitinstalliert. Ein Connection String ist nicht nötig.

Das Skript gibt eine Run ID aus, sendet zehn JSON-Messwerte für sensor-01 und liest anschließend Ereignisse. Dabei prüft es, ob alle zehn Messwerte dieses Laufs angekommen sind. Erwartete Ausgabe:

Run ID: <your-run-id>
SENT 10 events; run ID: <your-run-id>
READ partition=... sequence=... reading=1 temperature=20.1
...
READ partition=... sequence=... reading=10 temperature=21.0
PASS: read all 10 events for run <your-run-id>

Jedes Ereignis enthält runId, deviceId, reading und temperature. Der Partitionsschlüssel sensor-01 hält die Ereignisse dieses Geräts in einer Partition. Dieser kleine Lauf muss deshalb nicht beide Partitionen befüllen. Die ausgegebene Sequenznummer vergibt Event Hubs innerhalb der Partition; sie ist nicht die Messwertnummer.

Cloud Shell mit gelesenen Messwerten, Partitions- und Sequenzwerten sowie abschließender PASS-Zeile.

Erneut lesen, ohne erneut zu senden

Kopiere die Run ID des erfolgreichen Laufs. Ersetze PASTE_RUN_ID und bei Bedarf den Namespace und führe aus:

python event-hubs-demo.py --namespace evhns-ctstreamweu \
  --mode read --run-id PASTE_RUN_ID --consumer-group '$Default'

Diesmal liest das Skript nur. Es beginnt beim frühesten noch aufbewahrten Ereignis (starting_position="-1"), ignoriert andere Run IDs und prüft dieselben zehn Messwerte erneut. Es bestätigt oder löscht keine Ereignisse wie ein Service-Bus-Worker. Starte jeweils nur einen Demo-Empfänger.

Der reine Lesebefehl und ein weiteres PASS-Ergebnis für dieselbe Run ID ohne neue SENT-Zeile.

Eine echte Anwendung setzt normalerweise an dauerhaft gespeicherten Checkpoints fort, statt bei jedem Start den aufbewahrten Stream erneut zu durchsuchen. Unabhängige Anwendungen nutzen getrennte Consumer Groups und eigene Checkpoint-Daten. Dieses Beispiel hält seinen Prüffortschritt nur im Arbeitsspeicher und stoppt nach allen zehn Messwerten oder einem Empfangstimeout.

Prüfe bei verweigertem Zugriff das angemeldete Konto und seine Datenrolle. Meldet der Lauf fehlende Ereignisse, kontrolliere telemetry, $Default, Namespace-Namen und kopierte Run ID. Ereignisse außerhalb der Aufbewahrungsdauer lassen sich nicht erneut lesen. Mit –mode send sendet das Skript einen neuen Batch, ohne einen Empfänger zu starten. Das verwenden wir im Capture-Trip erneut.

Meldet die Demo Read 0/10, kann das Senden trotzdem erfolgreich gewesen sein. Lade das aktuelle Beispiel erneut herunter und hoch, um Partitionsdiagnosen zu erhalten. PARTITION zeigt, ob der Hub Ereignisse enthält, START nennt jeden Leser und WAIT zeigt gelesene und passende Ereignisse. Diese Zeilen allein beweisen noch keinen erfolgreichen Empfang.

Lies mit –mode read und derselben Run ID über den obigen Wiederholungsbefehl erneut. Behalte die einfachen Anführungszeichen um $Default bei, damit Bash es nicht als Variable interpretiert. Nutze bei weiteren Fehlern Partitionsdiagnosen und vollständige Fehlermeldung zur Untersuchung. Ein Timeout allein belegt die Ursache nicht. Sende zur Untersuchung des Lesens keine neuen Ereignisse.

Abschluss

Behalte Namespace und Event Hub für den nächsten Capture-Trip. Schließe Cloud Shell nach dem Test. Falls du nicht weitermachst, lösche rg-cloudtrips-eventhubs-test-weu, um die Kapazitätskosten des Namespace zu beenden. Nur das Skript zu stoppen beendet diese Kosten nicht.