Konfigurieren von Datenflüssen in Azure IoT Einsatz

Wichtig

Die von Azure Arc unterstützte Vorschauversion von „Azure IoT Einsatz“ befindet sich derzeit in der Vorschauphase. Sie sollten diese Vorschausoftware nicht in Produktionsumgebungen verwenden.

Sie müssen eine neue Installation von „Azure IoT Einsatz“ bereitstellen, wenn ein allgemein verfügbares Release verfügbar wird. Sie werden kein Upgrade für eine Preview-Installation durchführen können.

Die zusätzlichen Nutzungsbestimmungen für Microsoft Azure-Vorschauen enthalten rechtliche Bedingungen. Sie gelten für diejenigen Azure-Features, die sich in der Beta- oder Vorschauversion befinden oder aber anderweitig noch nicht zur allgemeinen Verfügbarkeit freigegeben sind.

Ein Datenfluss ist der Pfad, den Daten von der Quelle zum Ziel mit optionalen Transformationen nehmen. Sie können den Datenfluss mithilfe des Azure IoT Einsatz Studio-Portals konfigurieren oder eine benutzerdefinierte Datenfluss-Ressource erstellen. Ein Datenfluss besteht aus drei Teilen: der Quelle, der Transformation und dem Ziel.

Um die Quelle und das Ziel zu definieren, müssen Sie die Datenflussendpunkte konfigurieren. Die Transformation ist optional und kann Vorgänge wie das Anreichern der Daten, das Filtern der Daten und das Zuordnen der Daten zu einem anderen Feld umfassen.

In diesem Artikel zeigt Ihnen anhand eines Beispiels, wie Sie einen Datenfluss erstellen, einschließlich Quelle, Transformation und Ziel.

Voraussetzungen

• Eine Instanz der Vorschauversion von Azure IoT Einsatz
• Ein konfiguriertes Datenflussprofil
• Datenflussendpunkte. Erstellen Sie beispielsweise einen Datenflussendpunkt für den lokalen MQTT-Broker. Sie können diesen Endpunkt sowohl als Quelle als auch als Ziel verwenden. Sie können auch andere Endpunkte wie Kafka, Event Hubs oder Azure Data Lake Storage ausprobieren. Informationen dazu, wie Sie die einzelnen Typen von Datenflussendpunkten konfigurieren, finden Sie unter Konfigurieren von Datenflussendpunkten.

Dataflow erstellen

Sobald Sie über Datenflussendpunkte verfügen, können Sie diese zum Erstellen eines Datenflusses verwenden. Denken Sie daran, dass ein Datenfluss aus drei Teilen besteht: der Quelle, der Transformation und dem Ziel.

Portal

Um einen Datenfluss im Azure IoT Einsatz-Portal zu erstellen, wählen Sie Datenfluss>Datenfluss erstellen aus.

Kubernetes

Die Gesamtstruktur einer Datenflusskonfiguration sieht wie folgt aus:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

In den folgenden Abschnitten erfahren Sie, wie Sie die Vorgangstypen des Datenflusses konfigurieren.

Konfigurieren einer Quelle mit einem Datenflussendpunkt zum Abrufen von Daten

Um eine Quelle für den Datenfluss zu konfigurieren, geben Sie den Endpunktverweis und die Datenquelle an. Sie können eine Liste der Datenquellen für den Endpunkt angeben.

Portal

Verwenden einer Ressourcen als Quelle

Sie können eine Ressource als Quelle für den Datenfluss verwenden. Dies ist nur im Azure IoT Einsatz-Portal verfügbar.

1. Wählen Sie unter Quelldetails den Eintrag Ressource aus.

2. Wählen Sie die Ressource aus, die Sie als Quellendpunkt verwenden möchten.

3. Wählen Sie Proceed (Fortfahren) aus.

   Es wird eine Liste der Datenpunkte für die ausgewählte Ressource angezeigt.

   

4. Wählen Sie Übernehmen aus, um die Ressource als Quellendpunkt zu verwenden.

Kubernetes

Die Konfiguration einer Ressource als Quelle ist nur im Azure IoT Einsatz-Portal verfügbar.

Verwenden von MQTT als Quelle

Portal

1. Wählen Sie unter Quelldetails die Option MQTT aus.

2. Geben Sie das MQTT-Thema ein, das Sie auf eingehende Nachrichten überwachen möchten.

3. Wählen Sie ein Nachrichtenschema aus der Dropdownliste aus, oder laden Sie ein neues Schema hoch. Wenn die Quelldaten optionale Felder oder Felder mit unterschiedlichen Typen aufweisen, geben Sie ein Deserialisierungsschema an, um die Konsistenz sicherzustellen. Beispielsweise können die Daten Felder enthalten, die nicht in allen Nachrichten vorhanden sind. Ohne das Schema kann die Transformation diese Felder nicht verarbeiten, weil sie Leerwerte enthalten. Mit dem Schema können Sie Standardwerte angeben oder die Felder ignorieren.

   

4. Wählen Sie Übernehmen.

Kubernetes

Um beispielsweise eine Quelle mit einem MQTT-Endpunkt und zwei MQTT-Themenfiltern zu konfigurieren, verwenden Sie die folgende Konfiguration:

sourceSettings:
  endpointRef: mq
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/humidity/#

Weil Sie in dataSources MQTT- oder Kafka-Themen angeben können, ohne die Endpunktkonfiguration zu ändern, können Sie den Endpunkt auch dann für verschiedene Datenflüsse wiederverwenden, wenn die Themen unterschiedlich sind. Weitere Informationen finden Sie unter Wiederverwenden von Datenflussendpunkten.

Angeben des Schemas zum Deserialisieren von Daten

Wenn die Quelldaten optionale Felder oder Felder mit unterschiedlichen Typen aufweisen, geben Sie ein Deserialisierungsschema an, um die Konsistenz sicherzustellen. Beispielsweise können die Daten Felder enthalten, die nicht in allen Nachrichten vorhanden sind. Ohne das Schema kann die Transformation diese Felder nicht verarbeiten, weil sie Leerwerte enthalten. Mit dem Schema können Sie Standardwerte angeben oder die Felder ignorieren.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

Um das Schema anzugeben, erstellen Sie die Datei und speichern sie in der Schemaregistrierung.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Hinweis

Als Serialisierungsformat wird lediglich JSON unterstützt. Das Schema ist optional.

Weitere Informationen zur Schemaregistrierung finden Sie unter Grundlegendes zu Nachrichtenschemata.

Gemeinsame Abonnements

Um freigegebene Abonnements mit MQTT-Quellen zu verwenden, können Sie das Thema für ein freigegebenes Abonnement in Form von $shared/<subscription-group>/<topic> angeben.

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Hinweis

Wenn die Anzahl der Instanzen im Datenflussprofil größer als 1 ist, muss das Thema für das freigegebene Abonnement verwendet werden.

Konfigurieren der Transformation zum Verarbeiten von Daten

Mit dem Transformationsvorgang können Sie die Daten aus der Quelle transformieren, bevor Sie sie an das Ziel senden. Transformationen sind optional. Wenn Sie keine Änderungen an den Daten vornehmen müssen, schließen Sie den Transformationsvorgang nicht in die Datenflusskonfiguration ein. Mehrere Transformationen werden in Phasen verkettet, unabhängig von der Reihenfolge, in der sie in der Konfiguration angegeben sind. Die Reihenfolge der Phasen lautet immer wie folgt

1. Anreichern: Fügen Sie zusätzliche Daten zu den Quelldaten hinzu, die einem Dataset und einer Bedingung zugeordnet werden sollen.
2. Filtern: Filtern Sie die Daten anhand einer Bedingung.
3. Zuordnen: Verschieben Sie die Daten aus einem Feld in ein anderes unter Verwendung einer optionalen Konvertierung.

Portal

Wählen Sie im Azure IoT Einsatz-Portal die Option Datenfluss>Transformation hinzufügen (optional) aus.

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

Anreichern: Hinzufügen von Verweisdaten

Um die Daten anzureichern, können Sie das Verweisdataset im verteilten Zustandsspeicher (Distributed State Store, DSS) von „Azure IoT Einsatz“ verwenden. Das Dataset wird verwendet, um zusätzliche Daten zu den Quelldaten basierend auf einer Bedingung hinzuzufügen. Die Bedingung wird als Feld in den Quelldaten angegeben, die einem Feld im Dataset entsprechen.

Die Schlüsselnamen im verteilten Zustandsspeicher entsprechen einem Dataset in der Datenflusskonfiguration.

Portal

Derzeit ist der Anreicherungsvorgang im Azure IoT Einsatz-Portal nicht verfügbar.

Kubernetes

Sie können z. B. das Feld deviceId in den Quelldaten verwenden, damit es dem Feld asset im Dataset entspricht:

builtInTransformationSettings:
  datasets:
  - key: assetDataset
    inputs:
      - $source.deviceId # ------------- $1
      - $context(assetDataset).asset # - $2
    expression: $1 == $2

Wenn das Dataset über einen Datensatz mit dem Feld asset verfügt, ähnelt es Folgendem:

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

Die Daten aus der Quelle mit dem deviceId-Feld, das thermostat1 entspricht, verfügen über die Felder location und manufacturer in den Phasen filter und map.

Sie können Beispieldaten in den DSS laden, indem Sie das DSS-Toolbeispiel verwenden.

Weitere Informationen zur Bedingungssyntax finden Sie unter Anreichern von Daten mithilfe von Datenflüssen und Konvertieren von Daten mithilfe von Datenflüssen.

Filter: Filtern von Daten basierend auf einer Bedingung

Um die Daten nach einer Bedingung zu filtern, können Sie die filter-Phase verwenden. Die Bedingung wird als Feld in den Quelldaten angegeben, die einem Wert entsprechen.

Portal

1. Wählen Sie unter Transformieren (optional) die Option Filter>Hinzufügen aus.

2. Wählen Sie die Datenpunkte aus, die in das Dataset aufgenommen werden sollen.

3. Fügen Sie eine Filterbedingung und eine Beschreibung hinzu.

   

4. Wählen Sie Übernehmen.

Kubernetes

Sie können z. B. das Feld temperature in den Quelldaten verwenden, um die Daten zu filtern:

builtInTransformationSettings:
  filter:
    - inputs:
      - temperature ? $last # - $1
      expression: "$1 > 20"

Wenn das Feld temperature größer als 20 ist, werden die Daten an die nächste Phase übergeben. Wenn das Feld temperature kleiner oder gleich 20 ist, werden die Daten gefiltert.

Zuordnung: Verschieben von Daten aus einem Feld in ein anderes

Um die Daten einem anderen Feld mit optionaler Konvertierung zuzuordnen, können Sie den map-Vorgang verwenden. Die Konvertierung wird als Formel angegeben, die die Felder in den Quelldaten verwendet.

Portal

Im Azure IoT Einsatz-Portal wird die Zuordnung derzeit mithilfe von Compute-Transformationen unterstützt.

1. Wählen Sie unter Transformieren (optional) die Option Compute>Hinzufügen aus.

2. Geben Sie die erforderlichen Felder und Ausdrücke ein.

   

3. Wählen Sie Übernehmen.

Kubernetes

Sie können beispielsweise das Feld temperature in den Quelldaten verwenden, um die Temperatur in Celsius zu konvertieren und im Feld temperatureCelsius zu speichern. Sie können die Quelldaten auch mit dem Feld location aus dem Kontextualisierungsdataset anreichern:

builtInTransformationSettings:
  map:
    - inputs:
      - temperature # - $1
      output: temperatureCelsius
      expression: "($1 - 32) * 5/9"
    - inputs:
      - $context(assetDataset).location  
      output: location

Weitere Informationen finden Sie unter Zuordnen von Daten mithilfe von Datenflüssen und Konvertieren von Daten mithilfe von Datenflüssen.

Serialisieren von Daten nach einem Schema

Wenn Sie die Daten vor dem Senden an das Ziel serialisieren möchten, müssen Sie ein Schema- und Serialisierungsformat angeben. Andernfalls werden die Daten mit den abgeleiteten Typen in JSON serialisiert. Denken Sie daran, dass Speicherendpunkte wie Microsoft Fabric oder Azure Data Lake ein Schema benötigen, um die Datenkonsistenz sicherzustellen.

Portal

Geben Sie das Ausgabeschema an, wenn Sie den Zieldatenflussendpunkt hinzufügen.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

Um das Schema anzugeben, können Sie eine benutzerdefinierte Schemaressource mit der Schemadefinition erstellen.

Weitere Informationen zur Schemaregistrierung finden Sie unter Grundlegendes zu Nachrichtenschemata.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

Unterstützt werden die Serialisierungsformate JSON, Parkett und Delta.

Konfigurieren des Ziels mit einem Datenflussendpunkt zum Senden von Daten

Um ein Ziel für den Datenfluss zu konfigurieren, geben Sie den Endpunktverweis und das Datenziel an. Sie können eine Liste von Datenzielen für den Endpunkt angeben, wobei es sich um MQTT- oder Kafka-Themen handelt.

Portal

1. Wählen Sie den Datenflussendpunkt aus, der als Ziel dienen soll.

   

2. Wählen Sie Fortfahren aus, um das Ziel zu konfigurieren.

3. Fügen Sie die Zuordnungsdetails basierend auf dem Zieltyp hinzu.

Kubernetes

Wenn Sie beispielsweise ein Ziel mithilfe des zuvor erstellten MQTT-Endpunkts und eines statischen MQTT-Themas konfigurieren möchten, verwenden Sie die folgende Konfiguration:

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Wenn Sie Speicherendpunkte wie Microsoft Fabric erstellt haben, verwenden Sie das Datenzielfeld, um den Tabellen- oder Containernamen anzugeben:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

Beispiel

Das folgende Beispiel ist eine Datenflusskonfiguration, die den MQTT-Endpunkt als Quelle und Ziel verwendet. Die Quelle filtert die Daten aus den MQTT-Themen thermostats/+/telemetry/temperature/# und humidifiers/+/telemetry/humidity/#. Die Transformation wandelt die Temperatur in Fahrenheit um und filtert die Daten heraus, bei denen die Temperatur kleiner als 100000 ist. Das Ziel sendet die Daten an das MQTT-Thema factory.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

Überprüfen, ob ein Datenfluss funktioniert

Folgen Sie dem Tutorial: Bidirektionale MQTT-Brücke zu Azure Event Grid, um zu überprüfen, ob der Datenfluss funktioniert.

Exportieren der Datenflusskonfiguration

Zum Exportieren der Datenflusskonfiguration können Sie das Azure IoT Einsatz-Portal verwenden oder die benutzerdefinierte Datenflussressource exportieren.

Portal

Wählen Sie den zu exportierenden Datenfluss aus, und wählen Sie auf der Symbolleiste Exportieren aus.

Kubernetes

kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml