Voor sommige scenario's moeten berichten binnenkomen op verschillende MQTT-onderwerpen, afhankelijk van hun inhoud. Sensormetingen boven een kritieke drempelwaarde moeten bijvoorbeeld naar een alerts onderwerp gaan, terwijl normale metingen naar een historian onderwerp gaan. Met gegevensstroomgrafieken kunt u het uitvoeronderwerp dynamisch instellen, ook al heeft de gegevensstroom één bestemming.
Dynamische topicrouting is een techniek gebaseerd op de maptransformatie: een mapregel schrijft het doelonderwerp naar de metadata van het bericht, en de bestemming publiceert naar dat onderwerp. Om berichten via verschillende verwerkingspaden binnen de grafiek te routeren, zie Filter, branch en merge data.
Zie Het overzicht van gegevensstroomgrafieken voor een overzicht van gegevensstroomgrafieken en hoe transformaties in een pijplijn worden samengesteld.
Prerequisites
- Een standaardregistereindpunt dat naar
defaultmcr.microsoft.com verwijst wordt automatisch gemaakt tijdens de implementatie. De ingebouwde transformaties gebruiken dit eindpunt.
De Azure CLI voorbeelden in dit artikel gebruiken omgevingsvariabelen zodat je elke waarde één keer kunt instellen en vervolgens de commando's kunt kopiëren en plakken as-is. Als je de Azure IoT-bewerkingen Codespaces-omgeving vanuit de quickstart gebruikt, zijn deze variabelen al voor je ingesteld en kun je deze stap overslaan. Anders stel je de volgende omgevingsvariabelen in je shell voordat je de commando's uitvoert.
De volgende scripts stellen de meest gebruikte omgevingsvariabelen in:
| Omgevingsvariabele |
Description |
SUBSCRIPTION_ID |
De ID van het abonnement dat je Azure IoT-bewerkingen-instantie bevat. |
RESOURCE_GROUP |
De naam van de resourcegroep die je Azure IoT-bewerkingen-instantie bevat. |
AIO_INSTANCE_NAME |
De naam van je Azure IoT-bewerkingen instance. Om je instanties op te sommen, voer az iot ops list -o tableje . |
CLUSTER_NAME |
De naam van de Azure Arc-enabled Kubernetes-cluster die jouw instantie host. |
LOCATION |
De Azure-regio om te gebruiken voor nieuwe bronnen, bijvoorbeeld eastus. |
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>"
Je hoeft alleen de variabelen in te stellen die dit artikel gebruikt. Dit artikel kan extra omgevingsvariabelen gebruiken voor de bronnamen die je kiest. Het artikel legt uit hoe je ze op de plek kunt plaatsen waar ze worden geïntroduceerd.
Hoe dynamische topic routing werkt
Een kaarttransformatie kan schrijven naar metagegevens van berichten, inclusief het MQTT-onderwerp, met behulp van het $metadata.topic uitvoerpad. De bestemming gebruikt dan de ${outputTopic} variabele om naar welk onderwerp de transformatieset ook maar ingesteld is, te publiceren.
Twee onderdelen werken samen:
-
Binnen de transformatie: Een toewijzingsmapregel schrijft een stringwaarde naar
$metadata.topic.
-
In de bestemming: het
dataDestination veld verwijst naar ${outputTopic}, wat resulteert in de waarde die de transformatie heeft geschreven.
Transformaties gebruiken een expressietaal om waarden, testvoorwaarden en referentievelden te berekenen. Expressies verwijzen naar invoer op positie, niet op naam: de eerste invoer in de inputs lijst is $1, de tweede is $2, enzovoort. Ingebouwde functies zoals cToF converteren en manipuleren deze waarden.
Voor de volledige lijst van operatoren, functies, datatypes en metadatavelden, zie de referentie Expressions.
Dit artikel schrijft naar de metadata van berichten. Voor de metadatapaden die je kunt lezen en schrijven, zie Metadatavelden.
De eenvoudigste benadering maakt gebruik van één kaarttransformatie met een if expressie waarmee het onderwerp wordt gekozen.
Maak in de bewerkingservaring een gegevensstroomgrafiek:
- Voeg een bron toe waaruit
sensors/temperaturewordt gelezen.
- Voeg een kaarttransformatie toe met twee regels:
- Een passthrough-regel met jokertekens (invoer
*, uitvoer *).
- Een rekenregel met invoer
temperature, uitvoer $metadata.topicen expressie if($1 > 1000, "alerts", "historian").
- Voeg een bestemming toe met onderwerp
factory/${outputTopic}.
Wanneer de kaarttransformatie "alerts" naar $metadata.topic schrijft, wordt factory/${outputTopic} naar de bestemming omgezet in factory/alerts.
De Azure CLI gebruikt een dataflowgraaf uit één JSON-configuratiebestand. Maak een graph.json bestand met de grafiekeigenschappen. In het graph.json bestand slaat elke transformatie zijn regels in het value veld op als een ontsnapte JSON-string. Voor de leesbare vorm van de regels van elke transformatie, zie het how-to-artikel voor dat transformatietype.
{
"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
Om de ge-escapete tekenreeks te genereren, sla je de regels op in een bestand zoals rules.json, voer je jq -c . rules.json uit en plak je de eenregelige uitvoer in het veld value.
Pas het configuratiebestand toe.
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' } }
]
}
}
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
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 }
Optie 2: Route met een aftakking, per-pad kaarten en een samenvoeging
Als u verschillende transformaties voor elk pad nodig hebt (niet alleen een ander onderwerp), gebruikt u een vertakkingstransformatie om de stroom te splitsen, een kaarttransformatie op elke arm om het onderwerp in te stellen en padspecifieke regels toe te passen en een samenvoegingstransformatie om de paden samen te voegen.
In de operatie-ervaring:
- Voeg een bron toe waaruit
sensors/temperaturewordt gelezen.
- Voeg een vertakkingstransformatie met voorwaarde
$1 > 1000 toe aan het temperature veld.
- Voeg op het werkelijke pad een kaarttransformatie toe met een jokerteken-passthrough en een regel die wordt ingesteld
$metadata.topic op "alerts".
- Op het valse pad voeg een map-transformatie toe met een jokerteken-passthrough en een regel die
$metadata.topic op "historian" instelt.
- Voeg een samenvoegingstransformatie toe om beide paden samen te voegen.
- Voeg een bestemming toe met onderwerp
factory/${outputTopic}.
De Azure CLI gebruikt een dataflowgraaf uit één JSON-configuratiebestand. Maak een graph.json bestand met de grafiekeigenschappen. In het graph.json bestand slaat elke transformatie zijn regels in het value veld op als een ontsnapte JSON-string. Voor de leesbare vorm van de regels van elke transformatie, zie het how-to-artikel voor dat transformatietype.
{
"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"
}
}
]
}
Pas het configuratiebestand toe.
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' } }
]
}
}
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
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 }
| Consideratie |
Optie 1 (één kaart) |
Optie 2 (branch + kaarten) |
| Eenvoud |
Minder knooppunten, eenvoudiger te lezen |
Meer knooppunten, explicieter |
| Routering met alleen onderwerp |
Ideal |
Werkt, maar meer setup dan nodig is |
| Verschillende transformaties per pad |
Mogelijk met geneste if(), maar het wordt wel ingewikkeld. |
Natuurlijke: iedere vertakking heeft zijn eigen kaartregels |
| Meer paden toevoegen |
Ketenoproepen if() |
Vereist geneste vertakkingen |
Voor eenvoudige onderwerproutering op basis van één voorwaarde is optie 1 eenvoudiger. Gebruik optie 2 wanneer elk pad een andere verwerking nodig heeft dan de onderwerpnaam.
Hoe de outputTopic-variabele het bestemmingsonderwerp oplost
De ${outputTopic} variabele in dataDestination wordt omgezet in de volledige waarde van $metadata.topic zoals ingesteld door de laatste transformatie in de pijplijn. U kunt segmenten ook gebruiken met ${outputTopic.N} (1 geïndexeerd). Als de transformatie bijvoorbeeld $metadata.topic naar "region/west" transformeert:
dataDestination |
Opgelost onderwerp |
factory/${outputTopic} |
factory/region/west |
factory/${outputTopic.1} |
factory/region |
factory/${outputTopic.2} |
factory/west |
Als de datastroom de topicvariabele niet kan oplossen (bijvoorbeeld $metadata.topic nooit is ingesteld), laat het het bericht vallen en registreert een foutmelding.
Verwante onderwerpen