Daten filtern, verzweigen und zusammenführen in Datenflussdiagrammen

Datenflussdiagramme bieten zwei Möglichkeiten, um zu steuern, welche Nachrichten durch Ihre Pipeline fließen: Filtertransformationen legen unerwünschte Nachrichten ab, und Verzweigungstransformationen leiten jede Nachricht basierend auf einer Bedingung auf einen von zwei Pfaden weiter. Nach der Verzweigung führt eine verkettete Transformation die Pfade wieder zusammen.

Diese Transformationen routen Nachrichten innerhalb des Graphen. Um Nachrichten stattdessen an verschiedene MQTT-Themen basierend auf deren Inhalt zu routen, siehe Nachrichten zu verschiedenen MQTT-Themen routen.

Eine Übersicht über Datenflussdiagramme und die Art und Weise, wie Transformationen in einer Pipeline erstellt werden, finden Sie in der Übersicht über Datenflussdiagramme.

Transformationen verwenden eine Ausdruckssprache, um Werte, Testbedingungen und Referenzfelder zu berechnen. Ausdrücke beziehen sich auf Eingaben nach Position, nicht nach Namen: Die erste Eingabe in der inputs Liste ist $1, die zweite ist $2, und so weiter. Integrierte Funktionen wie cToF konvertieren und bearbeiten diese Werte.

Für die vollständige Liste der Operatoren, Funktionen, Datentypen und Metadatenfelder siehe die Expressions-Referenz.

Voraussetzungen

Eine Instanz von Azure IoT Einsatz, die in einem Kubernetes-Cluster bereitgestellt wird. Weitere Informationen finden Sie unter Deploy Azure IoT Einsatz.

Ein Standardregistrierungsendpunkt mit dem Namen default , auf den mcr.microsoft.com verweist, wird während der Bereitstellung automatisch erstellt. Die integrierten Transformationen verwenden diesen Endpunkt.

Die Azure CLI Beispiele in diesem Artikel verwenden Umgebungsvariablen, sodass Sie jeden Wert einmal festlegen und dann die Befehle as-iskopieren und einfügen können. Wenn Sie die Azure IoT Einsatz Codespaces-Umgebung aus dem Quickstart verwenden, sind diese Variablen bereits für Sie festgelegt und Sie können diesen Schritt überspringen. Ansonsten setze die folgenden Umgebungsvariablen in deiner Shell, bevor du die Befehle ausführst.

Die folgenden Skripte legen die am häufigsten verwendeten Umweltvariablen fest:

Umgebungsvariable	Beschreibung
`SUBSCRIPTION_ID`	Die ID des Abonnements, das Ihre Azure IoT Einsatz-Instanz enthält.
`RESOURCE_GROUP`	Der Name der Ressourcengruppe, die Ihre Azure IoT Einsatz-Instanz enthält.
`AIO_INSTANCE_NAME`	Der Name Ihrer Azure IoT Einsatz Instanz. Um deine Instanzen aufzulisten, führe `az iot ops list -o table`.
`CLUSTER_NAME`	Der Name des Azure Arc-fähigen Kubernetes-Clusters, der deine Instanz hostet.
`LOCATION`	Die Azure-Region zur Nutzung für neue Ressourcen, zum Beispiel `eastus`.

Bash
PowerShell

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

Du musst nur die Variablen festlegen, die dieser Artikel verwendet. Dieser Artikel könnte zusätzliche Umgebungsvariablen für die von Ihnen gewählten Ressourcennamen verwenden. Der Artikel erklärt, wie man sie dort platziert, wo sie eingeführt werden.

Filter-Transformation

Eine Filtertransformation wertet jede eingehende Nachricht anhand einer oder mehrerer Regeln aus und entscheidet, ob die Nachricht über die Pipeline fortgesetzt wird oder verworfen wird.

Von Bedeutung

Ein Filter-Ausdruck wählt die zu entfernenden Nachrichten aus, nicht die, die bewahrt werden sollen. Wenn der Ausdruck wahr ist, wird die Nachricht verworfen. Dieses Verhalten ist das Gegenteil von eines Zuordnungsausdrucks, bei dem ein Wert berechnet und beibehalten wird.

Um die Nachrichten zu behalten, die einer Bedingung entsprechen, kehren Sie den Ausdruck um. Um zum Beispiel nur Werte über 90 zu halten, filtern Sie auf $1 <= 90.

Funktionsweise von Filterregeln

Jede Filterregel verfügt über die folgenden Eigenschaften:

Eigentum	Erforderlich	Beschreibung
`inputs`	Ja	Liste der Feldpfade, die aus der eingehenden Nachricht gelesen werden sollen.
`expression`	Ja	Formel, die auf die Eingabewerte angewendet wurde. Muss einen booleschen Wert zurückgeben. Wenn sie `true` zurückgibt, wird die Nachricht verworfen.
`description`	No	Lesbare Bezeichnung, die in Fehlermeldungen verwendet wird.

Jede Eingabe wird basierend auf ihrer Reihenfolge auf eine Positionsvariable abgebildet: Die erste Eingabe ist $1, die zweite ist $2, und so weiter.

Wenn Sie mehrere Regeln definieren, verwenden sie OR-Logik: Wenn eine Regel als wahr ausgewertet wird, wird die Nachricht gelöscht. Das Modul erleidet einen Kurzschluss, sobald eine Regel zutrifft.

Wichtige Einschränkungen:

Ausdruck ist erforderlich. Jede Filterregel muss eine expression.
filter nimmt ein Array. Stellen Sie Regeln als JSON-Array bereit, "filter": [ { ... } ], auch für eine einzelne Regel. Die Übergabe eines leeren Objekts verhindert das Laden der Transformation, und der daraus resultierende Fehler verweist auf das Artefakt und die Registry statt auf die Regelnutzlast. Diese Nebenbedingung unterscheidet sich von branch, die ein einzelnes Objekt annimmt.
Keine Wildcardeingaben. Jede Eingabe muss auf einen bestimmten Feldpfad verweisen.
Fehlende Felder verursachen Fehler. Wenn ein Feld, auf das in inputs verwiesen wird, nicht existiert, gibt der Filter einen Fehler zurück, anstatt die Nachricht stillschweigend weiterzugeben.
Nicht boolesche Ergebnisse verursachen Fehler. Wenn ein Ausdruck einen nicht booleschen Wert (z. B. eine Zeichenfolge oder Zahl) zurückgibt, gibt der Filter einen Fehler zurück.

Nachrichten nach Bedingung ablegen

Um Nachrichten zu verwerfen, bei denen die Temperatur 100 Grad überschreitet:

Fügen Sie in der Filtertransformationskonfiguration eine Regel hinzu:

Setting	Wert
Eingabe	`temperature`
Ausdruck	`$1 > 100`

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Füge diesen Ausschnitt an die entsprechende Stelle in deiner graph.json Partie hinzu und wende ihn mit az iot ops dataflowgraph apply.

"filter": [
  {
    "inputs": [
      "temperature"
    ],
    "expression": "$1 > 100"
  }
]

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 > 100'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

- inputs:
    - temperature     # $1
  expression: "$1 > 100"

Nachrichten, bei denen die Temperatur 100 oder geringer ist, werden durchgelassen. Nachrichten über 100 werden gelöscht.

Nachrichten nach Bedingung aufbewahren

Oft möchte man das gegenteilige Ergebnis: Man behält nur die Nachrichten, die zu einer Bedingung passen. Da ein Filter-Ausdruck auswählt, was entfernt werden soll, kehren Sie den Vergleich um.

Um nur die Werte über 90 zu halten, lassen Sie alles auf oder unter 90 fallen:

Fügen Sie in der Filtertransformationskonfiguration eine Regel hinzu:

Setting	Wert
Eingabe	`temperature`
Ausdruck	`$1 <= 90`
Beschreibung	`Drop readings at or below 90`

"filter": [
  {
    "inputs": [
      "temperature"
    ],
    "expression": "$1 <= 90",
    "description": "Drop readings at or below 90"
  }
]

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 <= 90'
    description: 'Drop readings at or below 90'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

- inputs:
    - temperature     # $1
  expression: "$1 <= 90"
  description: "Drop readings at or below 90"

Nur Nachrichten mit Werten über 90 laufen weiter durch die Pipeline. Wenn man hier $1 > 90 schreiben würde, hätte das den gegenteiligen Effekt von dem, was Sie erreichen wollen: Alle Messwerte über 90 würden verworfen und die niedrigeren beibehalten.

Tipp

Verwenden Sie das Feld description, um den Zweck der Regel hinsichtlich dessen zu erfassen, was sie verwirft. Eine Beschreibung wie Drop readings at or below 90 bleibt präzise, während Keep hot readings den Fehler mit invertierten Ausdrücken begünstigt und dann auch in Fehlermeldungen erscheint, sodass diese sich rückwärts lesen.

Verwenden mehrerer Bedingungen

Wenn Sie mehrere Regeln definieren, legt der Filter die Nachricht ab, wenn eine Regel übereinstimmt:

Fügen Sie zwei Regeln hinzu:

Eingabe	Ausdruck	Beschreibung
`temperature`	`$1 > 100`	Hohe Temperatur senken
`humidity`	`$1 > 95`	Hohe Luftfeuchtigkeit senken

"filter": [
  {
    "inputs": [
      "temperature"
    ],
    "expression": "$1 > 100",
    "description": "Drop high temperature"
  },
  {
    "inputs": [
      "humidity"
    ],
    "expression": "$1 > 95",
    "description": "Drop high humidity"
  }
]

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 > 100'
    description: 'Drop high temperature'
  }
  {
    inputs: [ 'humidity' ]
    expression: '$1 > 95'
    description: 'Drop high humidity'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

- inputs:
    - temperature     # $1
  expression: "$1 > 100"
  description: "Drop high temperature"

- inputs:
    - humidity     # $1
  expression: "$1 > 95"
  description: "Drop high humidity"

Meldung	Temperaturregel	Luftfeuchtigkeitsregel	Ergebnis
`{"temperature": 150, "humidity": 60}`	STIMMT	FALSCH	Dropped
`{"temperature": 80, "humidity": 98}`	FALSCH	STIMMT	Dropped
`{"temperature": 80, "humidity": 60}`	FALSCH	FALSCH	Pässe

Tipp

Verwenden Sie mehrere Eingaben in einer Regel, wenn Sie AND-Logik über Felder hinweg benötigen. Verwenden Sie mehrere Regeln, wenn Sie OR-Logik für unabhängige Bedingungen benötigen.

Verwenden komplexer Ausdrücke

Verweisen Sie in einer einzigen Regel auf mehrere Felder, und kombinieren Sie sie mit logischen Operatoren:

Fügen Sie eine Regel mit den Eingaben temperature und humidity sowie dem Ausdruck $1 > 30 && $2 < 60 hinzu.

"filter": [
  {
    "inputs": [
      "temperature",
      "humidity"
    ],
    "expression": "$1 > 30 && $2 < 60",
    "description": "Drop hot and dry readings"
  }
]

filter: [
  {
    inputs: [ 'temperature', 'humidity' ]
    expression: '$1 > 30 && $2 < 60'
    description: 'Drop hot and dry readings'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

- inputs:
    - temperature     # $1
    - humidity        # $2
  expression: "$1 > 30 && $2 < 60"
  description: "Drop hot and dry readings"

Eine vollständige Liste der Operatoren und Funktionen finden Sie unter Expressions reference.

Filternachrichten gegen ein Schema validieren

Konfigurieren Sie eine Filtertransformation, um eingehende Nachrichten vor der Ausführung der Filterregeln mit einem JSON-Schema zu überprüfen. Der Prozess lässt Nachrichten fallen, die nicht dem Schema entsprechen.

Um die Schemaüberprüfung zu aktivieren, legen Sie sie in der Filterkonfiguration fest validateSchematrue . Wenn diese Option aktiviert ist, ruft der Filter das Schema aus schemaRef in der eingehenden Knotenverbindung ab (die from Seite des nodeConnections Eintrags, der in den Filterknoten einfließt).

Die Filtertransformationskonfiguration enthält ein Kontrollkästchen "Schema überprüfen ". Die Vorgangsanzeige unterstützt derzeit allerdings nicht das Konfigurieren oder Anzeigen von schemaRef auf Knotenverbindungen. Um die Schemasvalidierung zu nutzen, konfigurieren Sie die schemaRef der Knotenverbindung mithilfe von Bicep- oder Kubernetes-Manifesten.

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an, fügen Sie dies also an die entsprechende Stelle in Ihrem graph.json hinzu, und wenden Sie es mit az iot ops dataflowgraph apply. In der graph.json Datei werden die Regeln jeder Transformation im value Feld als ESCAPE-JSON-Zeichenfolge gespeichert. Informationen zur lesbaren Form der Regeln der einzelnen Transformationen finden Sie in der Vorgehensweise für diesen Transformationstyp.

"nodes": [
  {
    "nodeType": "Graph",
    "name": "schema-filter",
    "graphSettings": {
      "registryEndpointRef": "default",
      "artifact": "azureiotoperations/graph-dataflow-filter:1.0.0",
      "configuration": [
        {
          "key": "rules",
          "value": "{\"validateSchema\":true,\"filter\":[]}"
        }
      ]
    }
  }
],
"nodeConnections": [
  {
    "from": {
      "name": "sensors",
      "schema": {
        "schemaRef": "aio-sr://my-namespace/sensor-schema:1",
        "serializationFormat": "Json"
      }
    },
    "to": {
      "name": "schema-filter"
    }
  }
]

Tipp

Um die entkommene Zeichenkette zu erzeugen, speichere die Regeln in einer Datei wie rules.json, führe jq -c . rules.json, und füge die einzeilige Ausgabe ins Feld value ein.

Fügen Sie validateSchema in die JSON-Filterregeln ein und konfigurieren Sie schemaRef in der eingehenden Knotenverbindung.

nodes: [
  {
    nodeType: 'Graph'
    name: 'schema-filter'
    graphSettings: {
      registryEndpointRef: 'default'
      artifact: 'azureiotoperations/graph-dataflow-filter:1.0.0'
      configuration: [
        {
          key: 'rules'
          value: '{"validateSchema":true,"filter":[]}'
        }
      ]
    }
  }
]
nodeConnections: [
  {
    from: {
      name: 'sensors'
      schema: {
        schemaRef: 'aio-sr://my-namespace/sensor-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: { name: 'schema-filter' }
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

nodes:
  - nodeType: Graph
    name: schema-filter
    graphSettings:
      registryEndpointRef: default
      artifact: azureiotoperations/graph-dataflow-filter:1.0.0
      configuration:
        - key: rules
          value: |
            {
              "validateSchema": true,
              "filter": []
            }

nodeConnections:
  - from:
      name: sensors
      schema:
        schemaRef: "aio-sr://my-namespace/sensor-schema:1"
        serializationFormat: Json
    to:
      name: schema-filter

Leitlinien:

Verwenden Sie nur einen Validierungsfilter pro Pipeline.
Platzieren Sie zuerst den Überprüfungsfilter, sodass ungültige Nachrichten vor der anderen Verarbeitung verworfen werden.
Filterregeln gelten nach Ablauf der Schemaüberprüfung weiterhin. Wenn Sie nur eine Schemaüberprüfung benötigen, lassen Sie die Filterregeln leer.
Das schemaRef Muss auf ein Schema in der Schemaregistrierung verweisen. Das serializationFormat gibt das Schemaformat an (zum Beispiel Json).

Informationen zum Konfigurieren von Schemas finden Sie unter "Grundlegendes zu Nachrichtenschemas".

Anreichern von Filterregeln mit externen Daten

Filterregeln unterstützen Datasets, mit denen Sie Werte mit Daten aus einem externen Zustandsspeicher vergleichen können. Ausführliche Informationen zum Konfigurieren von Datasets finden Sie unter "Anreichern mit externen Daten".

Vollständige Filterkonfiguration

Fügen Sie in der Filtertransformationskonfiguration eine oder mehrere Regeln mit Eingaben und booleschen Ausdrücken hinzu. Aktivieren Sie optional die Schemaüberprüfung und konfigurieren Sie Datasets für Anreicherungs-Nachschlagevorgänge.

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Fügen Sie dies also dem Transformationsknoten configuration in Ihrem graph.json hinzu, und wenden Sie es mit az iot ops dataflowgraph applyan.

Die Regeln sind ein JSON-Objekt:

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": ["temperature"],
      "expression": "$1 > 100",
      "description": "Drop high temperature readings"
    },
    {
      "inputs": ["rawValue", "$context(limits).maxValue"],
      "expression": "$1 > $2",
      "description": "Drop readings above device-specific limit"
    }
  ]
}

Diese Regeln werden als Escapezeichenfolge im value-Feld abgelegt:

"configuration": [
  {
    "key": "rules",
    "value": "{\"datasets\":[{\"key\":\"device_limits as limits\",\"inputs\":[\"$source.deviceId\",\"$context.deviceId\"],\"expression\":\"$1 == $2\"}],\"filter\":[{\"inputs\":[\"temperature\"],\"expression\":\"$1 > 100\",\"description\":\"Drop high temperature readings\"},{\"inputs\":[\"rawValue\",\"$context(limits).maxValue\"],\"expression\":\"$1 > $2\",\"description\":\"Drop readings above device-specific limit\"}]}"
  }
]

Die JSON-Filterregeln werden als value für den rules Schlüssel übergeben.

configuration: [
  {
    key: 'rules'
    value: '{"datasets":[{"key":"device_limits as limits","inputs":["$source.deviceId","$context.deviceId"],"expression":"$1 == $2"}],"filter":[{"inputs":["temperature"],"expression":"$1 > 100","description":"Drop high temperature readings"},{"inputs":["rawValue","$context(limits).maxValue"],"expression":"$1 > $2","description":"Drop readings above device-specific limit"}]}'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": ["temperature"],
      "expression": "$1 > 100",
      "description": "Drop high temperature readings"
    },
    {
      "inputs": ["rawValue", "$context(limits).maxValue"],
      "expression": "$1 > $2",
      "description": "Drop readings above device-specific limit"
    }
  ]
}

Schlüssel	Erforderlich	Beschreibung
`filter`	Ja	Reihe von Filterregeln.
`datasets`	No	Array von Datasetdefinitionen für Anreicherungs-Nachschlagevorgänge.
`validateSchema`	No	Wenn `true`, überprüft Nachrichten anhand eines JSON-Schemas, bevor Filterregeln ausgeführt werden. Wird standardmäßig auf `false` festgelegt.

Verzweigungstransformation

Eine Verzweigungstransformation wertet eine Bedingung für jede eingehende Nachricht aus und leitet sie an einen von zwei Ausgabepfaden weiter: true oder false. Im Gegensatz zu einem Filter (der Nachrichten abbricht), behält eine Verzweigung jede Nachricht bei und leitet sie auf den entsprechenden Pfad weiter.

Funktionsweise der Verzweigung

Jede Nachricht geht zu genau einem der beiden Pfade. Nichts geht verloren.

Wichtige Einschränkungen:

Der Verzweigungsausdruck muss einen booleschen Wert zurückgeben. Nicht-boolesche Ergebnisse verursachen einen Fehler.
Keine Wildcardeingaben.
Genau eine Verzweigungsregel. Der branch Schlüssel verwendet ein einzelnes Objekt, kein Array.

Von Bedeutung

Verzweigt teilt Nachrichten in separate Verarbeitungspfade auf, aber alle Pfade müssen sich durch eine Konkatenate-Transformation wieder zusammenfügen, bevor sie das Ziel erreichen. Stellen Sie sich die Verzweigung als Eine Möglichkeit vor, unterschiedliche Transformationen auf verschiedene Nachrichten anzuwenden, nicht als Möglichkeit zum Weiterleiten an mehrere Endpunkte.

Verzweigungsregel definieren

So verzweigen Sie Nachrichten basierend auf einem Schweregradschwellenwert:

Legen Sie in der Verzweigungstransformationskonfiguration Folgendes fest:

Setting	Wert
Eingabe	`severity`
Ausdruck	`$1 > 5`

Die Regeln sind ein JSON-Objekt:

{
  "branch": {
    "inputs": ["severity"],
    "expression": "$1 > 5",
    "description": "Route high-severity messages"
  }
}

Diese Regeln werden als Escapezeichenfolge im value-Feld abgelegt:

"configuration": [
  {
    "key": "rules",
    "value": "{\"branch\":{\"inputs\":[\"severity\"],\"expression\":\"$1 > 5\",\"description\":\"Route high-severity messages\"}}"
  }
]

configuration: [
  {
    key: 'rules'
    value: '{"branch":{"inputs":["severity"],"expression":"$1 > 5","description":"Route high-severity messages"}}'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "branch": {
    "inputs": ["severity"],
    "expression": "$1 > 5",
    "description": "Route high-severity messages"
  }
}

Nachrichten, die severity größer als 5 sind, wechseln zum true Pfad. Alle anderen gehen zum false Pfad.

Überprüfen Sie Verzweigungsnachrichten anhand eines Schemas

Ab Version 1.1.0 können Sie eine Verzweigungstransformation konfigurieren, um eingehende Nachrichten vor der Auswertung des Verzweigungsausdrucks anhand eines JSON-Schemas zu überprüfen.

Um die Schema-Validierung zu aktivieren, setzen Sie validateSchema in der Branch-Konfiguration auf true. Das Feld validateSchema ist optional und steht standardmäßig auf false. Wenn diese Option aktiviert ist, ruft die Verzweigung das Schema aus schemaRef in der eingehenden Knotenverbindung ab (die from Seite des nodeConnections Eintrags, der in den Verzweigungsknoten einfließt).

Nachrichten, die die Schemaüberprüfung bestehen, werden anschließend einer Verzweigungsbewertung unterzogen.
Nachrichten, die die Schema-Validierung nicht bestehen, gehen auf den Pfad false .

Die Konfiguration der Verzweigungstransformation enthält das Kontrollkästchen Schema validieren. Die Vorgangsanzeige unterstützt derzeit allerdings nicht das Konfigurieren oder Anzeigen von schemaRef auf Knotenverbindungen. Um die Schemasvalidierung zu nutzen, konfigurieren Sie die schemaRef der Knotenverbindung mithilfe von Bicep- oder Kubernetes-Manifesten.

"nodes": [
  {
    "nodeType": "Graph",
    "name": "schema-branch",
    "graphSettings": {
      "registryEndpointRef": "default",
      "artifact": "azureiotoperations/graph-dataflow-branch:1.1.0",
      "configuration": [
        {
          "key": "rules",
          "value": "{\"validateSchema\":true,\"branch\":{\"inputs\":[\"severity\"],\"expression\":\"$1 > 5\"}}"
        }
      ]
    }
  }
],
"nodeConnections": [
  {
    "from": {
      "name": "sensors",
      "schema": {
        "schemaRef": "aio-sr://my-namespace/sensor-schema:1",
        "serializationFormat": "Json"
      }
    },
    "to": {
      "name": "schema-branch"
    }
  }
]

Tipp

Um die entkommene Zeichenkette zu erzeugen, speichere die Regeln in einer Datei wie rules.json, führe jq -c . rules.json, und füge die einzeilige Ausgabe ins Feld value ein.

Fügen Sie validateSchema in das JSON für die Verzweigungsregeln ein und konfigurieren Sie schemaRef für die eingehende Knotenverbindung:

nodes: [
  {
    nodeType: 'Graph'
    name: 'schema-branch'
    graphSettings: {
      registryEndpointRef: 'default'
      artifact: 'azureiotoperations/graph-dataflow-branch:1.1.0'
      configuration: [
        {
          key: 'rules'
          value: '{"validateSchema":true,"branch":{"inputs":["severity"],"expression":"$1 > 5"}}'
        }
      ]
    }
  }
]
nodeConnections: [
  {
    from: {
      name: 'sensors'
      schema: {
        schemaRef: 'aio-sr://my-namespace/sensor-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: { name: 'schema-branch' }
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

nodes:
  - nodeType: Graph
    name: schema-branch
    graphSettings:
      registryEndpointRef: default
      artifact: azureiotoperations/graph-dataflow-branch:1.1.0
      configuration:
        - key: rules
          value: |
            {
              "validateSchema": true,
              "branch": {
                "inputs": ["severity"],
                "expression": "$1 > 5"
              }
            }

nodeConnections:
  - from:
      name: sensors
      schema:
        schemaRef: "aio-sr://my-namespace/sensor-schema:1"
        serializationFormat: Json
    to:
      name: schema-branch

Zweigausgänge verbinden

Verwenden Sie in der Pipelinekonfiguration den Knotennamen, gefolgt von .output.true oder .output.false um jeden Pfad mit einer nachgeschalteten Transformation zu verbinden.

Ziehen Sie im Datenflussdiagramm-Editor Verbindungen aus den tatsächlichen und falschen Ausgaben der Verzweigungstransformation in die entsprechenden nachgeschalteten Transformationen.

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "severity-check"
    }
  },
  {
    "from": {
      "name": "severity-check.output.true"
    },
    "to": {
      "name": "alert-transform"
    }
  },
  {
    "from": {
      "name": "severity-check.output.false"
    },
    "to": {
      "name": "normal-transform"
    }
  }
]

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'severity-check' } }
  { from: { name: 'severity-check.output.true' }, to: { name: 'alert-transform' } }
  { from: { name: 'severity-check.output.false' }, to: { name: 'normal-transform' } }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

nodeConnections:
  - from: { name: sensors }
    to: { name: severity-check }
  - from: { name: severity-check.output.true }
    to: { name: alert-transform }
  - from: { name: severity-check.output.false }
    to: { name: normal-transform }

Zusammenführen von Pfaden mit Verketten

Alle Zweigpfade müssen zusammengeführt werden, bevor sie das Ziel erreichen. Eine verkettete Transformation führt sie zusammen. Sie hat keine Konfiguration und keine Regeln. Nachrichten von allen verbundenen Eingaben werden unverändert weitergeleitet.

Fügen Sie eine Verkettungstransformation zum Zeichenbereich hinzu, und verbinden Sie beide Verzweigungspfade damit, und verbinden Sie dann die Verkettung mit dem Ziel.

{
  "nodeType": "Graph",
  "name": "merge",
  "graphSettings": {
    "registryEndpointRef": "default",
    "artifact": "azureiotoperations/graph-dataflow-concatenate:1.0.0"
  }
}

{
  nodeType: 'Graph'
  name: 'merge'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
  }
}

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

- nodeType: Graph
  name: merge
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-concatenate:1.0.0

Beispiel: Filtern, Verzweigen und Zusammenführen

Dieses durchgängige Beispiel filtert fehlerhafte Messwerte heraus, verzweigt nach Schweregrad, wendet auf jeden Pfad unterschiedliche Zuordnungstransformationen an und führt die Ergebnisse zusammen.

Um diese Pipeline in der Betriebserfahrung aufzubauen:

Erstellen Sie ein Datenflussdiagramm und fügen Sie eine Quelle hinzu, die aus telemetry/sensors gelesen wird.
Fügen Sie eine Filtertransformation hinzu. Konfigurieren Sie eine Regel, die Nachrichten temperature > 1000dort abbricht.
Fügen Sie eine Verzweigungstransformation hinzu. Konfigurieren Sie die Bedingung severity > 5 , um Nachrichten mit hohem Schweregrad an den tatsächlichen Pfad weiterzuleiten.
Fügen Sie eine Kartentransformation auf dem wahren Pfad hinzu. Konfigurieren Sie Regeln zum Umbenennen von deviceId in id und von temperature in temp, und fügen Sie ein Feld alert hinzu, das auf true gesetzt wird.
Fügen Sie eine Kartentransformation auf dem falschen Pfad hinzu. Konfigurieren Sie Regeln für die Umbenennung deviceId in id und temperature in temp.
Fügen Sie eine verkettete Transformation hinzu, um beide Pfade zusammenzuführen.
Fügen Sie ein Ziel hinzu, das an telemetry/processed sendet.
Verbinden Sie die Elemente: Quelle → Filter → Verzweigung → (wahrer Pfad: Alarmkarte, falscher Pfad: normale Karte) → Verkettung → Ziel.

Die Azure CLI wendet ein Datenflussdiagramm aus einer einzelnen JSON-Konfigurationsdatei an. Erstellen Sie eine graph.json Datei mit den Diagrammeigenschaften. In der graph.json Datei werden die Regeln jeder Transformation im value Feld als ESCAPE-JSON-Zeichenfolge gespeichert. Informationen zur lesbaren Form der Regeln der einzelnen Transformationen finden Sie in der Vorgehensweise für diesen Transformationstyp.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/sensors"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "remove-bad-data",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-filter:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"filter\":[{\"inputs\":[\"temperature\"],\"expression\":\"$1 > 1000\",\"description\":\"Drop impossible temperature readings\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "severity-check",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-branch:1.1.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"branch\":{\"inputs\":[\"severity\"],\"expression\":\"$1 > 5\",\"description\":\"Route high-severity messages\"}}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "alert-transform",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"deviceId\"],\"output\":\"id\"},{\"inputs\":[\"temperature\"],\"output\":\"temp\"},{\"inputs\":[],\"output\":\"alert\",\"expression\":\"true\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "normal-transform",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"deviceId\"],\"output\":\"id\"},{\"inputs\":[\"temperature\"],\"output\":\"temp\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "merge",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-concatenate:1.0.0"
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/processed"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "remove-bad-data"
      }
    },
    {
      "from": {
        "name": "remove-bad-data"
      },
      "to": {
        "name": "severity-check"
      }
    },
    {
      "from": {
        "name": "severity-check.output.true"
      },
      "to": {
        "name": "alert-transform"
      }
    },
    {
      "from": {
        "name": "severity-check.output.false"
      },
      "to": {
        "name": "normal-transform"
      }
    },
    {
      "from": {
        "name": "alert-transform"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "normal-transform"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "merge"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Wenden Sie die Konfigurationsdatei an.

az iot ops dataflowgraph apply \
  --name alert-routing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'alert-routing'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/sensors' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'remove-bad-data'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-filter:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"filter":[{"inputs":["temperature"],"expression":"$1 > 1000","description":"Drop impossible temperature readings"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'severity-check'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-branch:1.1.0'
          configuration: [
            {
              key: 'rules'
              value: '{"branch":{"inputs":["severity"],"expression":"$1 > 5","description":"Route high-severity messages"}}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'alert-transform'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["deviceId"],"output":"id"},{"inputs":["temperature"],"output":"temp"},{"inputs":[],"output":"alert","expression":"true"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'normal-transform'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["deviceId"],"output":"id"},{"inputs":["temperature"],"output":"temp"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'merge'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/processed'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
      { from: { name: 'remove-bad-data' }, to: { name: 'severity-check' } }
      { from: { name: 'severity-check.output.true' }, to: { name: 'alert-transform' } }
      { from: { name: 'severity-check.output.false' }, to: { name: 'normal-transform' } }
      { from: { name: 'alert-transform' }, to: { name: 'merge' } }
      { from: { name: 'normal-transform' }, to: { name: 'merge' } }
      { from: { name: 'merge' }, to: { name: 'output' } }
    ]
  }
}

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: alert-routing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/sensors

    - nodeType: Graph
      name: remove-bad-data
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-filter:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "filter": [
                  {
                    "inputs": ["temperature"],
                    "expression": "$1 > 1000",
                    "description": "Drop impossible temperature readings"
                  }
                ]
              }

    - nodeType: Graph
      name: severity-check
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-branch:1.1.0
        configuration:
          - key: rules
            value: |
              {
                "branch": {
                  "inputs": ["severity"],
                  "expression": "$1 > 5",
                  "description": "Route high-severity messages"
                }
              }

    - nodeType: Graph
      name: alert-transform
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["deviceId"], "output": "id" },
                  { "inputs": ["temperature"], "output": "temp" },
                  { "inputs": [], "output": "alert", "expression": "true" }
                ]
              }

    - nodeType: Graph
      name: normal-transform
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["deviceId"], "output": "id" },
                  { "inputs": ["temperature"], "output": "temp" }
                ]
              }

    - nodeType: Graph
      name: merge
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-concatenate:1.0.0

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/processed

  nodeConnections:
    - from: { name: sensors }
      to: { name: remove-bad-data }
    - from: { name: remove-bad-data }
      to: { name: severity-check }
    - from: { name: severity-check.output.true }
      to: { name: alert-transform }
    - from: { name: severity-check.output.false }
      to: { name: normal-transform }
    - from: { name: alert-transform }
      to: { name: merge }
    - from: { name: normal-transform }
      to: { name: merge }
    - from: { name: merge }
      to: { name: output }

Feedback

War diese Seite hilfreich?

Last updated on 2026-08-03

Daten filtern, verzweigen und zusammenführen in Datenflussdiagrammen

Voraussetzungen

Filter-Transformation

Funktionsweise von Filterregeln

Nachrichten nach Bedingung ablegen

Nachrichten nach Bedingung aufbewahren

Verwenden mehrerer Bedingungen

Verwenden komplexer Ausdrücke

Filternachrichten gegen ein Schema validieren

Anreichern von Filterregeln mit externen Daten

Vollständige Filterkonfiguration

Verzweigungstransformation

Funktionsweise der Verzweigung

Verzweigungsregel definieren

Überprüfen Sie Verzweigungsnachrichten anhand eines Schemas

Zweigausgänge verbinden

Zusammenführen von Pfaden mit Verketten

Beispiel: Filtern, Verzweigen und Zusammenführen

Verwandte Inhalte

Feedback

Zusätzliche Ressourcen