Certains scénarios nécessitent que les messages arrivent sur différentes rubriques MQTT en fonction de leur contenu. Par exemple, les lectures de capteur au-dessus d’un seuil critique peuvent avoir besoin d’accéder à une alerts rubrique, tandis que les lectures normales vont à une historian rubrique. Avec les graphiques de flux de données, vous pouvez définir la rubrique de sortie de manière dynamique, même si le flux de données a une destination unique.
Le routage dynamique des sujets est une technique construite sur la transformation de carte : une règle de carte écrit le sujet cible en métadonnées de message, et la destination publie sur ce sujet. Pour acheminer les messages sur différents chemins de traitement au sein du graphe, voir Filtrer, ramifier et fusionner les données.
Pour obtenir une vue d’ensemble des graphiques de flux de données et la façon dont les transformations composent dans un pipeline, consultez vue d’ensemble des graphiques de flux de données.
Prerequisites
- Un point de terminaison de registre par défaut nommé
default qui pointe vers mcr.microsoft.com est créé automatiquement pendant le déploiement. Les transformations intégrées utilisent ce point de terminaison.
Les exemples Azure CLI de cet article utilisent des variables d’environnement afin de pouvoir définir chaque valeur une fois puis copier-coller les commandes as-is. Si vous utilisez l'environnement Opérations Azure IoT Codespaces du quickstart, ces variables sont déjà définies pour vous et vous pouvez sauter cette étape. Sinon, définissez les variables d’environnement suivantes dans votre shell avant d’exécuter les commandes.
Les scripts suivants définissent les variables d’environnement les plus couramment utilisées :
| Variable d'environnement |
Description |
SUBSCRIPTION_ID |
L’identifiant de l’abonnement contenant votre instance Opérations Azure IoT. |
RESOURCE_GROUP |
Le nom du groupe de ressources contenant votre instance Opérations Azure IoT. |
AIO_INSTANCE_NAME |
Le nom de votre instance Opérations Azure IoT. Pour lister vos instances, exécutez az iot ops list -o table. |
CLUSTER_NAME |
Le nom du cluster Kubernetes compatible Azure Arc qui héberge votre instance. |
LOCATION |
La région Azure à utiliser pour de nouvelles ressources, par exemple 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>"
Vous n’avez qu’à définir les variables utilisées dans cet article. Cet article peut utiliser des variables d’environnement supplémentaires pour les noms de ressources que vous choisissez. L’article explique comment les placer là où ils sont introduits.
Fonctionnement du routage dynamique des sujets
Une transformation de type map peut écrire dans les métadonnées du message, y compris le topic MQTT, en utilisant le chemin de sortie $metadata.topic. La destination utilise ensuite la ${outputTopic} variable pour publier sur n’importe quelle rubrique du jeu de transformations.
Deux morceaux fonctionnent ensemble :
-
À l’intérieur de la transformation : une règle de carte écrit une valeur de chaîne dans
$metadata.topic.
-
Dans la destination : Le champ
dataDestination référence ${outputTopic}, ce qui se résout en la valeur que la transformation a écrite.
Les transformations utilisent un langage d’expressions pour calculer les valeurs, les conditions de test et les champs de référence. Les expressions désignent les entrées par position, et non par le nom : la première entrée de la inputs liste est $1, la seconde est $2, et ainsi de suite. Des fonctions intégrées telles que cToF convertissent et manipulent ces valeurs.
Pour la liste complète des opérateurs, fonctions, types de données et champs de métadonnées, voir la référence Expressions.
Cet article s’écrit sur les métadonnées des messages. Pour les chemins de métadonnées que vous pouvez lire et écrire, voir champs de métadonnées.
L’approche la plus simple utilise une transformation de carte avec une if expression qui sélectionne la rubrique.
Dans l’expérience Opérations, créez un graphe de flux de données :
- Ajoutez une source qui lit à partir de
sensors/temperature.
- Ajoutez une transformation de carte avec deux règles :
- Règle de passthrough générique (entrée
*, sortie *).
- Règle de calcul avec entrée
temperature, sortie $metadata.topicet expression if($1 > 1000, "alerts", "historian").
- Ajoutez une destination avec la rubrique
factory/${outputTopic}.
When the map transform writes "alerts" to $metadata.topic, la destination résout factory/${outputTopic} en factory/alerts.
L’interface Azure CLI utilise un graphe de flux de données provenant d’un seul fichier de configuration JSON. Créez un graph.json fichier avec les propriétés du graphique. Dans le graph.json fichier, chaque transformation stocke ses règles dans le value champ sous forme de chaîne JSON échappée. Pour la forme lisible des règles de chaque transformation, voir l’article pratique pour ce type de transformation.
{
"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
Pour générer la chaîne échappée, sauvegardez les règles dans un fichier comme rules.json, exécutez jq -c . rules.json, puis collez la sortie ligne unique dans le value champ.
Appliquez le fichier de configuration.
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' } }
]
}
}
Important
L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.
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 : Itinéraire avec une branche, des cartes par chemin et une fusion
Si vous avez besoin de transformations différentes sur chaque chemin d’accès (pas seulement une rubrique différente), utilisez une transformation de branche pour fractionner le flux, une transformation de carte sur chaque bras pour définir la rubrique et appliquer des règles spécifiques au chemin d’accès et une transformation concatène pour fusionner les chemins.
Dans l'expérience des opérations :
- Ajoutez une source qui lit à partir de
sensors/temperature.
- Ajoutez une transformation de branche avec condition
$1 > 1000 sur le temperature champ.
- On the truechemin, ajoutez une transformationmap avec un passage de caractères génériques et une règle qui définit
$metadata.topic à"alerts".
- On the faux chemin, ajoutez une transformation map avec un passage des caractères génériques et une règle qui définit
$metadata.topic à "historian".
- Ajoutez une transformation concatène pour fusionner les deux chemins.
- Ajoutez une destination avec la rubrique
factory/${outputTopic}.
L’interface Azure CLI utilise un graphe de flux de données provenant d’un seul fichier de configuration JSON. Créez un graph.json fichier avec les propriétés du graphique. Dans le graph.json fichier, chaque transformation stocke ses règles dans le value champ sous forme de chaîne JSON échappée. Pour la forme lisible des règles de chaque transformation, voir l’article pratique pour ce type de transformation.
{
"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"
}
}
]
}
Appliquez le fichier de configuration.
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' } }
]
}
}
Important
L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.
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 }
| Considération |
Option 1 (carte unique) |
Option 2 (branche + cartes) |
| Simplicité |
Moins de nœuds, plus simple à lire |
Plus de nœuds, plus explicites |
| Routage uniquement par sujet |
Idéal |
Fonctionne, mais plus d’installation que nécessaire |
| Différentes transformations par chemin |
Possible avec des if(), imbriqués, cela devient complexe. |
Naturel : chaque branche a ses propres règles cartographiques |
| Ajout de chemins d’accès supplémentaires |
Appels en chaîne if() |
Nécessite des branches imbriquées |
Pour le routage simple de rubriques en fonction d’une condition unique, l’option 1 est plus simple. Utilisez l’option 2 lorsque chaque chemin a besoin d’un traitement différent au-delà du nom de la rubrique.
Comment la variable outputTopic résout le sujet de destination
La variable ${outputTopic} dansdataDestination se résout en la valeur complète de $metadata.topic telle que définie par la dernière transformation du pipeline. Vous pouvez également utiliser des segments avec ${outputTopic.N} (1-indexed). Par exemple, si la transformation définit $metadata.topic sur "region/west" :
dataDestination |
Rubrique résolue |
factory/${outputTopic} |
factory/region/west |
factory/${outputTopic.1} |
factory/region |
factory/${outputTopic.2} |
factory/west |
Si le flux de données ne peut pas résoudre la variable sujet (par exemple, n’a $metadata.topic jamais été défini), il laisse tomber le message et enregistre une erreur.
Contenu connexe