Teilen über

Erstellen eines Dataflows

  • Artikel

Wichtig

Die von Azure Arc aktivierte Azure IoT Operations Preview befindet sich derzeit in der VORSCHAU. 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 ist. Sie können keine Installation der Vorschauversion upgraden.

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-Portals konfigurieren oder eine benutzerdefinierte Datenflussressource erstellen. Bevor Sie einen Datenfluss erstellen, müssen Sie Datenflussendpunkte für die Datenquellen und -ziele konfigurieren.

Im Folgenden sehen Sie ein Beispiel für eine Datenflusskonfiguration mit einem MQTT-Quellendpunkt, Transformationen und einem Kafka-Zielendpunkt:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
spec:
  profileRef: my-dataflow-profile
  mode: enabled
  operations:
    - operationType: source
      name: my-source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
        serializationFormat: json
    - operationType: builtInTransformation
      name: my-transformation
      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'
        serializationFormat: json
    - operationType: destination
      name: my-destination
      destinationSettings:
        endpointRef: kafka
        dataDestination: factory
Name Beschreibung
profileRef Verweis auf das Datenflussprofil
Modus Modus des Datenflusses, aktiviert oder deaktiviert
operations[] Vom Datenfluss ausgeführte Vorgänge
operationType Typ des Vorgangs, source, destination, oder builtInTransformation

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

Konfigurieren der Quelle

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. Zum Beispiel MQTT- oder Kafka-Themen. Nachfolgend sehen Sie ein Beispiel für eine Datenflusskonfiguration mit einem Quellendpunkt und einer Datenquelle.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: mq-to-kafka
  namespace: azure-iot-operations
spec:
  profileRef: example-dataflow
  operations:
    - operationType: source
      sourceSettings:
        endpointRef: mq-source
        dataSources:
        - azure-iot-operations/data/thermostat
Name Beschreibung
operationType Quelle
sourceSettings Einstellungen für den source-Vorgang
sourceSettings.endpointRef Verweisen auf den source-Endpunkt
sourceSettings.dataSources Datenquellen für den source-Vorgang. Die Platzhalterzeichen (# und +) werden unterstützt.

Konfigurieren der Transformation

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.

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    builtInTransformationSettings:
      datasets:
        # ...
      filter:
        # ...
      map:
        # ...
Name Beschreibung
operationType builtInTransformation
name Name der Transformation
builtInTransformationSettings Einstellungen für den builtInTransformation-Vorgang
builtInTransformationSettings.datasets Hinzufügen weiterer Daten zu den Quelldaten, die einem Dataset und einer Bedingung entsprechen
builtInTransformationSettings.filter Filtern der Daten basierend auf einer Bedingung
builtInTransformationSettings.map Verschieben von Daten aus einem Feld in ein anderes mit optionaler Konvertierung

Anreichern: Hinzufügen von Verweisdaten

Um die Daten anzureichern, können Sie ein 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.

Name Beschreibung
builtInTransformationSettings.datasets.key Dataset zur Anreicherung (Schlüssel in DSS)
builtInTransformationSettings.datasets.expression Bedingung für den Anreicherungsvorgang

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

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    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 dem thermostat1 entspricht, verfügen über die Felder location und manufacturer, die in den Phasen filter und map verfügbar sind.

Sie können Beispieldaten in den verteilten Zustandsspeicher (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.

Name Beschreibung
builtInTransformationSettings.filter.inputs[] Eingaben zum Auswerten einer Filterbedingung
builtInTransformationSettings.filter.expression Bedingung für die Filterauswertung

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

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    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.

Name Beschreibung
builtInTransformationSettings.map[].inputs[] Eingaben für den Zuordnungsvorgang
builtInTransformationSettings.map[].output Ausgabefeld für den Zuordnungsvorgang
builtInTransformationSettings.map[].expression Konvertierungsformel für den Zuordnungsvorgang

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:

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    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.

Konfigurieren des Ziels

Um ein Ziel für den Datenfluss zu konfigurieren, müssen Sie den Endpunkt und einen Pfad (Thema oder Tabelle) für das Ziel angeben.

Name Beschreibung
destinationSettings.endpointRef Verweis auf den destination-Endpunkt (Zielendpunkt)
destinationSettings.dataDestination Ziel für die Daten

Konfigurieren des Zielendpunktverweises

Um den Endpunkt für das Ziel zu konfigurieren, müssen Sie die ID und den Endpunktverweis angeben.

spec:
  operations:
  - operationType: destination
    name: destination1
    destinationSettings:
      endpointRef: eventgrid

Konfigurieren des Zielpfads

Sobald Sie den Endpunkt haben, können Sie den Pfad für das Ziel konfigurieren. Wenn es sich bei dem Ziel um einen MQTT- oder Kafka-Endpunkt handelt, verwenden Sie den Pfad, um das Thema anzugeben.

- operationType: destination
  destinationSettings:
    endpointRef: eventgrid
    dataDestination: factory

Verwenden Sie für Speicherendpunkte wie Microsoft Fabric den Pfad, um den Tabellennamen anzugeben.

- operationType: destination
  destinationSettings:
    endpointRef: adls
    dataDestination: telemetryTable