Daten im Vorübergehen verarbeiten

 Schwerpunkt Mengendaten verarbeiten
Azure Stream Analytics
Daten im Vorübergehen
verarbeiten
Die Datenbank als Problemfall? Das Nadelöhr bei großen Datenmengen umgehen.
K
lotzen, nicht kleckern: Bei der
Menge an Daten lassen sich industrielle Internet-of-Things-Lösungen nicht lumpen. Sie erzeugen richtig große Datenmengen. Eine Beispielanwendung im ersten Teil des
Artikels [1] zeigte, wie schnell sich
die Datenmengen aufsummieren. Da
stellt sich unweigerlich die Frage, wie
solche Datenmengen ausgewertet
werden können. Insbesondere wenn
es darum geht, einen einzelnen Tele- Werden die Daten von den Event-Quellen (Data in Motion) zur späteren Verarbeitung
metrie-Wert nahezu in Echtzeit zu in einer Datenbank abgelegt (Data at Rest), kann das aufgrund der Datenmenge pro Zeit zu
berechnen. Sehr häufig stößt man auf Problemen führen (Bild 1)
Architekturen, bei denen die Annahme der Telemetrie-Daten und die
Auswertung getrennt sind. Letztere wird häufig in nachgelaRest) abgelegt wird (Bild 1). Der Aufwand, der betrieben wergerten, batchorientierten Prozessen durchgeführt.
den muss, mehrere parallele Instanzen der Analysedienste zu
All diese Lösungen gehen davon aus, dass die Daten von
betreiben, erscheint anfangs gering, steigt aber exponentiell
Event Publishern, sprich von Sensoren oder Software, welche
mit den anfallenden Datenmengen an.
die Daten erhebt, an einen zentralen Sammelpunkt gesendet
Azure Stream Analytics
werden. Mit Azure Event Hubs [2] stellt Microsoft hierfür eiMit Azure Stream Analytics [9] stellt Microsoft einen cloudnen cloudbasierten Service zur Verfügung, der es sehr einbasierten Dienst zur Verfügung, der eingehende Datenströfach macht, Daten zentral zu sammeln und über einen defime direkt als Stream skalierbar verarbeiten kann. Das heißt,
nierbaren Zeitraum zu speichern.
die eingehenden Telemetrie-Daten müssen nicht vorab in eiIn der weiteren Verarbeitung werden die gesammelten Tener Datenbank abgelegt werden. Vielmehr kann der Strom
lemetrie-Daten aus dem Event Hub ausgelesen und physikaan Daten direkt innerhalb von Azure Stream Analytics mitlisch in einem Datensystem abgelegt. Ob dies nun in einem
tels SQL-ähnlicher Befehle verarbeitet werden (Bild 2). Die
relationalen Datenbanksystem wie zum Beispiel Azure SQL
Verarbeitung beschränkt sich hierbei nicht nur auf einen einDatabase [3] oder einer dokumentenorientierten Datenbank
zigen eingehenden Datenstrom.
wie etwa Azure Document DB [4], MongoDB [5] oder einfach
Es können mehrere Datenströme und zusätzlich auch stain Dateien auf einem Filesystem erfolgt, spielt in der Regel
tische Daten in einem sogenannten Stream Analytics Job als
bei der Komplexität der Entwicklung keine Rolle.
Input-Quellen definiert und gemeinsam verarbeitet werden.
Nachgelagerte, meist zeit- oder batchgesteuerte Analysen
Die Ergebnisse dieser Auswertungen können in einem defiverarbeiten und analysieren die gesammelten Daten, um die
nierten Ausgabemedium gespeichert werden.
Grundlage für weitere Entscheidungen zu liefern.
Hier stehen als Ausgabemedien Azure Event Hub, Azure
Dies kann zum Beispiel mit einem Microsoft Parallel Data
SQL Database und Azure Blob Storage zur Verfügung. Als
Warehouse [6], Lösungen wie Hadoop [7] oder HDInsight [8]
Eingabequellen können zurzeit Azure Service Bus Event
durchgeführt werden.
Hubs und Azure Blob Storage definiert werden.
Dieser klassische Ansatz stößt jedoch schnell an seine
Grenzen, wenn es darum geht, die anfallenden DatenmenEinen Azure Stream Analytics Job anlegen
gen in Near-Real-Time zu verarbeiten. Insbesondere wenn
Ein Azure Stream Analytics Job wird über das Azure Mader permanente Datenstrom der Event Publisher (Data in Monagement Portal angelegt. Es werden der Jobname, die Retion) zur nachgelagerten Analyse in Datenbanken (Data at
30
3.2015 www.dotnetpro.de
Schwerpunkt Mengendaten verarbeiten gion, in der er ausgeführt werden soll, und ein
Azure Storage Account benötigt. Letzterer wird
von Stream Analytics intern verwendet, um das
Monitoring und Troubleshooting laufender
Jobs zu ermöglichen. Nach erfolgreicher Anlage können dem Job Inputs und ein dedizierter
Output hinzugefügt werden. Wie der Name
schon vermuten lässt, handelt es sich bei Inputs
um die eingehenden und zu analysierenden
Datenströme beziehungsweise statischen Eingabequellen für Referenzdaten; bei Output um
das Ausgabemedium, in das die Ergebnisse des
Jobs geschrieben werden.
Im folgenden Beispiel wird als Eingabequelle ein bereits existierender Event Hub verwendet. Dieser kann während der Anlage der In- So fließen die Daten, wenn Azure Stream Analytics zum Einsatz kommt (Bild 2)
put-Quelle ausgewählt werden. Zu beachten
ist, dass ein sogenannter „Input Alias“ während der Anlage vergeben werden muss (Bild 3). Dieser Alias
gewählt werden. Es öffnet sich ein Query-Editor, in den Sie
wird im späteren Verlauf verwendet, um bei der Definition
die Query-Logik eingeben.
der Auswertungslogik des eingehenden Datenstroms auf dieDie Stream Analytics Query Language
sen zu verweisen.
Applikationen, die mit Datenströmen arbeiten beziehungsIm weiteren Verlauf des Anlegens wird das Format, in dem
weise diese analysieren, führen ihre Abfragen häufig auf dem
die Telemetrie-Daten von den Event Producern gesendet
eingehenden Datenstrom über einen bestimmten Zeitraum
werden, definiert. Zum jetzigen Zeitpunkt stehen dort Avaus. So sollen etwa alle eingehenden Daten analysiert werro [10], CSV und JSON als Ingest-Format und UTF8 als Enden, die in einem bestimmten Zeitfenster angeliefert wurden
coding zur Verfügung (Bild 4).
(zum Beispiel fünf Sekunden). Die Stream Analytics Query
Nach der Definition eines eingehenden Datenstroms muss
Language ist eine Untermenge von Standard-T-SQL, erweiein Output definiert werden. In diesem werden die Ergebnistert um die Möglichkeit, auf solchen Zeitfenstern basierend
se der Analyse- beziehungsweise Auswertungslogik übertraAbfragen durchzuführen.
gen. Der Output kann ebenfalls komfortabel über das MaStream Analytics stellt aktuell drei unterschiedliche Zeitnagement Portal angelegt werden.
fenstertypen für Abfragen zur Verfügung: Tumbling Window,
Im nächsten Schritt definieren Sie die eigentliche VerarbeiHopping Window und Sliding Window. Eines dieser Zeitfenstungslogik beziehungsweise Analyselogik eines Stream Anater kann in der Abfrage in einem GROUP-BY-Statement verlytics Jobs. Hierzu kann der Menüpunkt Query im Portal auswendet werden. So wird zum Beispiel im folgenden Beispiel
der Durchschnitt des Feldes Humidity alle fünf Sekunden ermittelt und in die Ausgabe geschrieben:
SELECT AVG(Humidity) FROM TelemetryIngest
GROUP BY TumblingWindow(second, 5)
Anlegen eines Azure Stream Analytics Jobs über das Azure
­Management Portal (Bild 3)
www.dotnetpro.de 3.2015
Dabei ist Humidity ein Feld des eingehenden Datenstroms
und TelemetryIngest der Alias, der bei der Anlage der InputQuelle vergeben wurde.
Ein Tumbling Window ist hierbei eine Serie von Zeitfenstern mit fester Größe, die nicht überlappen und zusammenhängend sind. So wird im obigen Beispiel der HumidityDurchschnitt alle fünf Sekunden ermittelt.
Sind fünf Sekunden abgelaufen, wird der Durchschnitt mit
den in den nächsten fünf Sekunden ankommenden Teleme­
trie-Daten neu berechnet.
Ein Hopping Window überschneidet sich im Gegensatz zu
einem Tumbling Window. Das heißt, wird ein Hopping Window mit einer Größe von zehn Sekunden und einer Hop-Time
von fünf Sekunden erzeugt, werden im ersten Fenster alle Telemetrie-Daten beginnend von Timestamp null bis Time- ▶
31
Schwerpunkt Mengendaten verarbeiten
stamp zehn verarbeitet. Im zweiten Fenster werden alle Daten von Timestamp fünf bis Timestamp 15 berücksichtigt. Alle Ingests zwischen Sekunde fünf und Sekunde zehn werden
somit in zwei Berechnungen einbezogen. Bei einem Sliding
Window werden alle möglichen Zeitfenster einer bestimmten
Länge berücksichtigt [11].
Datentypen/Query
Für die Verarbeitung von eingehenden Telemetrie-Daten ist
zum Teil der Datentyp entscheidend – so zum Beispiel bei numerischen Berechnungen [12]. Die Datentypen der eingehenden Telemetrie-Daten werden automatisch von Stream
Analytics auf Basis der eingehenden Daten abgeleitet. Dies
liefert nicht in jedem Fall das gewünschte Ergebnis. Es ist jedoch möglich, Datentypen für den eingehenden Datenstrom
zu definieren. Dies kann mit dem Statement CREATE TABLE
innerhalb der Job-Definition durchgeführt werden.
CREATE TABLE TelemetryIngest (
Auswahl des Datenformats für die ankommenden Daten (Bild 4)
DeviceId nvarchar(max),
Temperature float,
Pollution float,
Humidity float,
);
SELECT AVG(Humidity) FROM TelemetryIngest
der Testdialog nur Eingabedateien im JSON-Format, unabhängig davon, welches Format für Eingabequellen gewählt
wurde. Das heißt, auch wenn als Input-Format für den eingehenden Datenstrom CSV oder Avro gewählt wurde, muss das
Format der Testdatei JSON sein.
GROUP BY TumblingWindow(second,5);
[
{ "DeviceId": "Device-50", "Temperature": 18,
Der Query-Editor im Management Portal bietet zum jetzigen
Zeitpunkt einen Syntax-Check und rudimentäre Hilfe bei der
Suche nach Fehlern in einer Abfrage.
Mit einfachen bekannten SQL-Statements ist es nun möglich festzustellen, wenn zum Beispiel ein Sensor innerhalb einer bestimmten Zeitspanne (im folgenden Beispiel fünf Sekunden) mindestens zweimal die Überschreitung eines
Schwellenwertes mit einer identischen Messung meldet:
"Pollution": 78, "Humidity": 50 },
{ "DeviceId": "Device-50", "Temperature": 17,
"Pollution": 78, "Humidity": 50 },
{ "DeviceId": "Device-50", "Temperature": 17,
"Pollution": 78, "Humidity": 50 },
{ "DeviceId": "Device-50", "Temperature": 17,
"Pollution": 78, "Humidity": 50 },
{ "DeviceId": "Device-40", "Temperature": 16,
"Pollution": 90.5, "Humidity": 50 },
CREATE TABLE TelemetryIngest (
{ "DeviceId": "Device-40", "Temperature": 17,
DeviceId nvarchar(max),
"Pollution": 92.8, "Humidity": 50 },
Temperature float,
{ "DeviceId": "Device-40", "Temperature": 17,
Pollution float,
"Pollution": 95, "Humidity": 50 },
Humidity float,
{ "DeviceId": "Device-40", "Temperature": 17,
);
"Pollution": 97, "Humidity": 50 }
SELECT Pollution, Count(*), DeviceId FROM
]
TelemetryIngest
WHERE Pollution > 50
GROUP BY TumblingWindow(second,5), Pollution,
DeviceId
HAVING COUNT(*) > 1
Weitere, komplexere Abfragen sind ebenfalls möglich. Eine
Übersicht der aktuell unterstützten Statements findet sich unter [13]. Interessant ist auch die Möglichkeit, eine Abfrage ad
hoc mit Demodaten zu testen. Dazu steht nach Klick auf Test
ein File-Upload-Dialog zur Verfügung, mit dessen Hilfe Demodaten hochgeladen werden können. Derzeit unterstützt
32
Beachten Sie, dass bei Verwendung von CREATE TABLE die
Reihenfolge der Felder in der Datei identisch mit der Reihenfolge der Felddefinitionen im CREATE-TABLE-Statement sein
muss. Die Feldnamen innerhalb der JSON-Struktur werden
beim Mapping ignoriert.
Zeitmessung in temporalen Systemen
In temporalen Analysesystemen wie Azure Stream Analytics
ist es wichtig, alle eingehenden Daten mit einem Zeitstempel
zu versehen. Die Herausforderung hierbei ist aber, welche
Zeit herangezogen werden soll. Es kann die Zeit der Anliefe-
3.2015 www.dotnetpro.de
Schwerpunkt Mengendaten verarbeiten rung des Telemetrie-Datums an der eingehenden InputQuelle (zum Beispiel Event Hub) oder ein vom Datenerzeuger (Sensor) generierter Zeitpunkt für die Auswertung wichtig sein. So muss in manchen Systemen die eventuelle Latenz
der Datenübermittlung vom Datenerzeuger an die InputQuelle mit berücksichtigt werden.
In Azure Stream Analytics verfügt jedes eingehende Telemetrie-Datum über einen definierten Zeitstempel. Dieser ist
abhängig von der gewählen Input-Quelle. Bei Verwendung
von Event Hubs ist es zum Beispiel der Zeitpunkt, an dem die
Telemetrie-Daten am Event Hub angeliefert wurden. Dieser
Zeitstempel kann in Abfragen über System.Timestamp ermittelt werden. Soll jedoch für die Analyse ein vom Datenerzeuger erstellter Zeitstempel verwendet werden, kann dies mit
dem Schlüsselwort TIMESTAMP BY erreicht werden.
www.dotnetpro.de/SL1503IoTBigData3
[5] MongoDB, www.dotnetpro.de/SL1503IoTBigData4
[6] MS Parallel Data Warehouse,
www.dotnetpro.de/SL1503IoTBigData5
[7] Hadoop, http://hadoop.apache.org/
[8] HDInsight, www.dotnetpro.de/SL1503IoTBigData6
[9] Azure Stream Analytics,
www.dotnetpro.de/SL1503IoTBigData7
[10] Avro, www.dotnetpro.de/SL1503IoTBigData8
[11] Sliding Window,
www.dotnetpro.de/SL1503IoTBigData9
[12] Datentypen,
www.dotnetpro.de/SL1503IoTBigData10
[13] DML Statements,
www.dotnetpro.de/SL1503IoTBigData1
CREATE TABLE TelemetryIngest (
Robert Eichenseer
entwickelt seit langem Applikationen auf Basis
der Microsoft-Technologien. Aufgrund seiner
Arbeit in internationalen Projekten kann er Anforderungen heutiger Softwareentwicklungen
umsetzen oder betriebswirtschaftliche Prozesse Entwicklungsthemen zuordnen.
DeviceId nvarchar(max),
Temperature float,
Pollution float,
Humidity float,
CreationDate datetime,
);
SELECT Pollution, Count(*), DeviceId
FROM TelemetryIngest TIMESTAMP BY CreationDate
dnpCodeA1503IoTBigData
WHERE Pollution > 50
GROUP BY TumblingWindow(second,5), Pollution,
DeviceId
HAVING COUNT(*) > 1
Bei uns lernen Sie von den Experten!
www.sigs-datacom.de
Archivierungsaufgaben
Ein weiteres mächtiges Anwendungsgebiet für Azure Stream
Analytics ist der Einsatz für Archivierungsaufgaben. In vielen
Systemen werden alle Telemetrie-Daten für spätere Auswertungen und gegebenenfalls auch wegen gesetzlicher Aufbewahrungs- beziehungsweise Nachweispflicht unverändert
abgelegt. Dies kann erreicht werden, indem die Angabe des
Zeitfensters, sozusagen das GROUP-BY-Statement in der
Query, weggelassen wird. Dann werden alle eingehenden
Telemetrie-Daten an das Ausgabemedium weitergeleitet.
Konfiguriert man nun einen Azure Storage Blob als Ausgabe,
werden die Telemetrie-Daten in einer Datei abgelegt.
Die Clean Code Developer Akademie oder
Wie können Teams Software von höherer Qualität entwickeln?
Von und mit den Experten: Ralf Westphal & Stefan Lieser
Die Seminare:
Fazit
Mit Azure Stream Analytics stellt Microsoft einen hochskalierenden und ausfallsicheren Cloud-Dienst zur Verfügung, der
die zeitkritische Analyse von Datenströmen erlaubt. ◾
[1] Robert Eichenseer,
„Mengendaten von Sensoren verarbeiten“,
dotnetpro 2/2015, S. 20 ff.,
www.dotnetpro.de/A1502IoTBigData
[2] Azure Event Hubs,
www.dotnetpro.de/SL1503IoTBigData1
[3] Azure SQL DB, www.dotnetpro.de/SL1503IoTBigData2
[4] Azure Document DB,
www.dotnetpro.de 3.2015
• Clean Architecture
Agiler Softwareentwurf für mehr Effizenz
• Clean Coding
Funktionaler Softwareentwurf für
mehr Nachhaltigkeit
Diese aktuellen Seminare spiegeln die Erfahrungen der Trainer getreu
dem Clean Code Developer Wert „Kontinuierliche Verbesserung“ und
der Praktik „refactoring to deeper insight“ wieder!
Termine, Informationen und mehr unter:
www.sigs-datacom.de/seminare
Kontakt: SIGS DATACOM GmbH, Anja Keß
Lindlaustraße 2c, D-53842 Troisdorf
Tel.: +49 (0) 22 41 / 23 41-201
Fax: +49 (0) 22 41 / 23 41-199, Email: [email protected]