Routen Sie Nachrichten zu verschiedenen MQTT-Themen in Datenflussgraphen

Einige Szenarien erfordern Nachrichten, die je nach Inhalt zu verschiedenen MQTT-Themen eingehen. Sensorwerte über einem kritischen Schwellenwert müssen möglicherweise zu einem alerts-Topic gesendet werden, während normale Sensorwerte zu einem historian-Topic gesendet werden. Mit Datenflussdiagrammen können Sie das Ausgabethema dynamisch festlegen, obwohl der Datenfluss ein einziges Ziel hat.

Dynamisches Topic-Routing ist eine Technik, die auf der Map-Transformation basiert: Eine Map-Regel schreibt das Zielthema in Nachrichtenmetadaten, und das Ziel veröffentlicht an dieses Thema. Um Nachrichten stattdessen über verschiedene Verarbeitungspfade im Graphen zu routen, siehe Filter-, Verzweigungs- und Zusammenführungsdaten.

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

Prerequisites

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	Description
`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.

Wie dynamisches Topic-Routing funktioniert

Eine Map-Transformation kann über den Ausgabepfad $metadata.topic in die Metadaten einer Nachricht schreiben, einschließlich des MQTT-Themas. Das Ziel verwendet dann die Variable ${outputTopic}, um in dem vom Transformationsset festgelegten Thema zu veröffentlichen.

Zwei Teile arbeiten zusammen:

Innerhalb der Transformation: Eine Zuordnungsregel schreibt einen Zeichenfolgenwert in $metadata.topic.
Im Ziel: Das dataDestination-Feld verweist auf ${outputTopic}, das zu dem Wert aufgelöst wird, den die Transformation geschrieben hat.

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.

Dieser Artikel schreibt in Nachrichtenmetadaten. Für die Metadatenpfade, die Sie lesen und schreiben können, siehe Metadatenfelder.

Option 1: Route mit einer einzelnen Map-Transformation und einem bedingten Ausdruck

Der einfachste Ansatz verwendet eine Kartentransformation mit einem if Ausdruck, der das Thema auswählt.

Erstellen Sie in der Betriebsumgebung ein Datenflussdiagramm:

Fügen Sie eine Quelle hinzu, die von sensors/temperature gelesen wird.
Hinzufügen einer Kartentransformation mit zwei Regeln:
- Eine Passthroughregel für Wildcards (Eingabe *, Ausgabe *).
- Eine Berechnungsregel mit Eingabe temperature, Ausgabe $metadata.topicund Ausdruck if($1 > 1000, "alerts", "historian").
Fügen Sie ein Ziel mit Thema factory/${outputTopic}hinzu.

Wenn die Zuordnungstransformation "alerts" nach $metadata.topic schreibt, wird das Ziel factory/${outputTopic} nach factory/alerts aufgelöst.

Die Azure CLI verwendet einen Datenflussgraphen aus einer einzelnen JSON-Konfigurationsdatei. Erstellen Sie eine graph.json Datei mit den Diagrammeigenschaften. In der graph.json Datei speichert jede Transformation ihre Regeln im Feld value als entkommene JSON-Zeichenkette. Für die lesbare Form der Regeln jeder Transformation siehe den Anleitungsartikel zu diesem Transformationstyp.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensors/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "route-by-temperature",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"description\":\"Set topic based on temperature threshold\",\"inputs\":[\"temperature\"],\"output\":\"$metadata.topic\",\"expression\":\"if($1 > 1000, \\\"alerts\\\", \\\"historian\\\")\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "factory/${outputTopic}"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "route-by-temperature"
      }
    },
    {
      "from": {
        "name": "route-by-temperature"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Tip

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.

Wenden Sie die Konfigurationsdatei an.

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'dynamic-topic-routing'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'sensors/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'route-by-temperature'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"description":"Set topic based on temperature threshold","inputs":["temperature"],"output":"$metadata.topic","expression":"if($1 > 1000, \\"alerts\\", \\"historian\\")"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'factory/${outputTopic}'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'route-by-temperature' } }
      { from: { name: 'route-by-temperature' }, 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: dynamic-topic-routing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensors/temperature

    - nodeType: Graph
      name: route-by-temperature
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "description": "Set topic based on temperature threshold",
                    "inputs": ["temperature"],
                    "output": "$metadata.topic",
                    "expression": "if($1 > 1000, \"alerts\", \"historian\")"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: "factory/${outputTopic}"

  nodeConnections:
    - from: { name: sensors }
      to: { name: route-by-temperature }
    - from: { name: route-by-temperature }
      to: { name: output }

Option 2: Route mit einem Zweig, Pro-Pfad-Karten und einer Zusammenführung

Wenn Sie unterschiedliche Transformationen für jeden Pfad benötigen (nicht nur ein anderes Thema), verwenden Sie eine Verzweigungstransformation, um den Fluss aufzuteilen, eine Kartentransformation auf jedem Arm, um das Thema festzulegen und pfadspezifische Regeln anzuwenden, und eine verkettete Transformation zum Zusammenführen der Pfade.

In der Betriebsumgebung:

Fügen Sie eine Quelle hinzu, die von sensors/temperature gelesen wird.
Fügen Sie eine Verzweigungstransformation mit Bedingung $1 > 1000 für das temperature Feld hinzu.
Auf dem Pfad true fügen Sie eine Zuordnungstransformation mit einem Wildcard-Passthrough und einer Regel hinzu, die $metadata.topic auf "alerts" setzt.
Auf dem Pfad false fügen Sie eine Zuordnungstransformation mit einem Wildcard-Passthrough und einer Regel hinzu, die $metadata.topic auf "historian" setzt.
Fügen Sie eine verkettete Transformation hinzu, um beide Pfade zusammenzuführen.
Fügen Sie ein Ziel mit Thema factory/${outputTopic}hinzu.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensors/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "check-temperature",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-branch:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"branch\":{\"inputs\":[\"temperature\"],\"expression\":\"$1 > 1000\",\"description\":\"Route critical temperatures to alerts\"}}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "set-alerts-topic",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[],\"output\":\"$metadata.topic\",\"expression\":\"\\\"alerts\\\"\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "set-historian-topic",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[],\"output\":\"$metadata.topic\",\"expression\":\"\\\"historian\\\"\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "merge",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-concatenate:1.0.0"
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "factory/${outputTopic}"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "check-temperature"
      }
    },
    {
      "from": {
        "name": "check-temperature.output.true"
      },
      "to": {
        "name": "set-alerts-topic"
      }
    },
    {
      "from": {
        "name": "check-temperature.output.false"
      },
      "to": {
        "name": "set-historian-topic"
      }
    },
    {
      "from": {
        "name": "set-alerts-topic"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "set-historian-topic"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "merge"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Wenden Sie die Konfigurationsdatei an.

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'dynamic-topic-routing-branched'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'sensors/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'check-temperature'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-branch:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"branch":{"inputs":["temperature"],"expression":"$1 > 1000","description":"Route critical temperatures to alerts"}}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'set-alerts-topic'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":[],"output":"$metadata.topic","expression":"\\"alerts\\""}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'set-historian-topic'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":[],"output":"$metadata.topic","expression":"\\"historian\\""}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'merge'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'factory/${outputTopic}'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'check-temperature' } }
      { from: { name: 'check-temperature.output.true' }, to: { name: 'set-alerts-topic' } }
      { from: { name: 'check-temperature.output.false' }, to: { name: 'set-historian-topic' } }
      { from: { name: 'set-alerts-topic' }, to: { name: 'merge' } }
      { from: { name: 'set-historian-topic' }, 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: dynamic-topic-routing-branched
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensors/temperature

    - nodeType: Graph
      name: check-temperature
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-branch:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "branch": {
                  "inputs": ["temperature"],
                  "expression": "$1 > 1000",
                  "description": "Route critical temperatures to alerts"
                }
              }

    - nodeType: Graph
      name: set-alerts-topic
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["*"], "output": "*" },
                  { "inputs": [], "output": "$metadata.topic", "expression": "\"alerts\"" }
                ]
              }

    - nodeType: Graph
      name: set-historian-topic
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["*"], "output": "*" },
                  { "inputs": [], "output": "$metadata.topic", "expression": "\"historian\"" }
                ]
              }

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

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: "factory/${outputTopic}"

  nodeConnections:
    - from: { name: sensors }
      to: { name: check-temperature }
    - from: { name: check-temperature.output.true }
      to: { name: set-alerts-topic }
    - from: { name: check-temperature.output.false }
      to: { name: set-historian-topic }
    - from: { name: set-alerts-topic }
      to: { name: merge }
    - from: { name: set-historian-topic }
      to: { name: merge }
    - from: { name: merge }
      to: { name: output }

Wählen Sie zwischen einer einzelnen Map-Transformation und einem Branch

Überlegung	Option 1 (einzelne Karte)	Option 2 (Verzweigung + Karten)
Einfachheit	Weniger Knoten, einfacher zu lesen	Weitere Knoten, expliziter
Themenbasiertes Routing	Ideal	Funktioniert, erfordert aber mehr Einrichtung als nötig.
Unterschiedliche Transformationen pro Pfad	Möglich mit geschachtelten `if()`, es wird jedoch komplex.	Natürlich: Jeder Zweig hat seine eigenen Kartenregeln
Hinzufügen weiterer Pfade	Ketten `if()` Aufrufe	Erfordert geschachtelte Verzweigungen

Für das einfache Themenrouting basierend auf einer einzigen Bedingung ist Option 1 einfacher. Verwenden Sie Option 2, wenn jeder Pfad über den Themennamen hinaus eine andere Verarbeitung benötigt.

Wie die outputTopic-Variable das Zielthema löst

Die ${outputTopic} Variable in dataDestination wird in den kompletten Wert von $metadata.topic aufgelöst, wie er durch die letzte Transformation festgelegt wurde. Sie können auch Segmente mit ${outputTopic.N} (1-indiziert) verwenden. Beispiel: Wenn die Transformation $metadata.topic auf "region/west" setzt:

`dataDestination`	Thema behoben
`factory/${outputTopic}`	`factory/region/west`
`factory/${outputTopic.1}`	`factory/region`
`factory/${outputTopic.2}`	`factory/west`

Wenn der Datenfluss die Topic-Variable nicht auflösen kann (zum Beispiel $metadata.topic nie gesetzt wurde), verzichtet die Nachricht und meldet einen Fehler.

Throttle-Daten

Feedback

War diese Seite hilfreich?

Last updated on 2026-08-14