Bazı senaryolar, iletilerin içeriğine bağlı olarak farklı MQTT konularına ulaşmasını gerektirir. Örneğin, kritik eşiğin üzerindeki algılayıcı okumalarının bir alerts konuya, normal okumaların ise bir historian konuya gitmesi gerekebilir. Veri akışı graflarıyla, veri akışının tek bir hedefi olsa bile çıkış konusunu dinamik olarak ayarlayabilirsiniz.
Nasıl çalışır?
Harita dönüşümü, c0 çıkış yolunu kullanarak MQTT konusu da dahil olmak üzere ileti meta verilerine yazabilir. Hedef daha sonra ${outputTopic} değişkenini kullanarak dönüşüm tarafından belirlenen herhangi bir konuya yayın yapar.
İki parça birlikte çalışır:
-
Dönüşümün içinde: Eşleme kuralı bir dize değeri
$metadata.topic öğesine yazar.
-
Hedefte:
dataDestination alanı, dönüştürmenin yazdığı değere çözümlenen ${outputTopic} öğesine başvurur.
En basit yaklaşım, konuyu seçen bir ifadeyle bir if eşleme dönüşümü kullanır.
İşlemler deneyiminde bir veri akışı grafı oluşturun:
- kaynağından okuyan bir
sensors/temperature ekleyin.
- İki kuralla harita dönüşümü ekleme:
- Joker karakter geçiş kuralı (giriş
*, çıkış *).
- Giriş
temperature, çıkış $metadata.topicve ifade if($1 > 1000, "alerts", "historian")içeren bir işlem kuralı.
- konusuyla bir
factory/${outputTopic} ekleyin.
Harita dönüştürmesi "alerts" öğesine yazdığında, hedef $metadata.topic'yi factory/${outputTopic} olarak çözümler factory/alerts.
Azure CLI tek bir JSON yapılandırma dosyasından veri akışı grafiği uygular. Grafik özelliklerine sahip bir graph.json dosya oluşturun.
graph.json dosyasında, her dönüşümün kuralları value alanında kaçış karakterleriyle işlenmiş bir JSON dizesi olarak depolanır. Her dönüşümün kurallarının okunabilir biçimi için bu dönüştürme türünün nasıl yapılır bölümüne bakın.
{
"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"
}
}
]
}
İpucu
Kaçışlı dizeyi oluşturmak için kuralları rules.json gibi bir dosyaya kaydedin, ardından jq -c . rules.json komutunu çalıştırın ve tek satırlık çıktıyı value alanına yapıştırın.
Yapılandırma dosyasını uygulayın.
extendedLocation örneği ve kaynak grubundan otomatik olarak eklenir, bu nedenle dosyaya eklemeyin.
az iot ops dataflowgraph apply \
--name dynamic-topic-routing \
--instance <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' } }
]
}
}
Önemli
Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.
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 }
Seçenek 2: Dallan, her yolu haritalandır ve birleştir
Her yolda farklı dönüştürmelere ihtiyacınız varsa (yalnızca farklı bir konu başlığına değil), akışı bölmek için bir dal dönüşümü, konuyu ayarlamak ve yola özgü kurallar uygulamak için her kolda bir eşleme dönüşümü ve yolları birleştirmek için bir birleştirme dönüşümü kullanın.
Operasyon deneyiminde:
- kaynağından okuyan bir
sensors/temperature ekleyin.
- Alana koşul
$1 > 1000 içeren bir temperature dönüşümü ekleyin.
-
Doğru yolda joker karakter geçişine sahip bir harita dönüşümü ekleyin ve
$metadata.topic'yi "alerts" olarak ayarlayan bir kural ekleyin.
-
False yolunda, joker karakter geçişi ve 'yi
$metadata.topic olarak ayarlayan bir "historian" dönüşümü ekleyin.
- Her iki yolu da birleştirmek için bir birleştirme dönüşümü ekleyin.
- konusuyla bir
factory/${outputTopic} ekleyin.
Azure CLI tek bir JSON yapılandırma dosyasından veri akışı grafiği uygular. Grafik özelliklerine sahip bir graph.json dosya oluşturun.
graph.json dosyasında, her dönüşümün kuralları value alanında kaçış karakterleriyle işlenmiş bir JSON dizesi olarak depolanır. Her dönüşümün kurallarının okunabilir biçimi için bu dönüştürme türünün nasıl yapılır bölümüne bakın.
{
"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"
}
}
]
}
Yapılandırma dosyasını uygulayın.
extendedLocation örneği ve kaynak grubundan otomatik olarak eklenir, bu nedenle dosyaya eklemeyin.
az iot ops dataflowgraph apply \
--name dynamic-topic-routing-branched \
--instance <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' } }
]
}
}
Önemli
Kubernetes dağıtım bildirimlerinin kullanımı üretim ortamlarında desteklenmez ve yalnızca hata ayıklama ve test için kullanılmalıdır.
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 }
Hangi seçeneğin seçileceği
| Değerlendirme |
Seçenek 1 (tek harita) |
2. Seçenek (dal + haritalar) |
| Basitlik |
Daha az düğüm, okuması daha kolay |
Daha fazla düğüm, daha açık |
| Yalnızca konu başlığı yönlendirme |
Ideal |
Çalışır, ancak gerektiğinden daha fazla kurulum gerektirir |
| Yol başına farklı dönüşümler |
if() ile iç içeyken mümkündür, karmaşıklaşır. |
Doğal: Her dal kendi harita kurallarına sahiptir |
| Daha fazla yol ekleme |
Zincirleme if() çağrılar |
İç içe dallar gerektirir |
Tek bir koşula dayalı basit konu yönlendirmesi için 1. seçenek daha basittir. Her yol, konu adının ötesinde farklı işlemeye ihtiyaç duyduğunda 2. seçeneği kullanın.
Konu çevirisi ayrıntıları
${outputTopic} içindeki dataDestination değişkeni, işlem hattındaki son dönüşüm tarafından ayarlanan $metadata.topic'nin tam değerine dönüştürülür. "${outputTopic.N} etiketiyle birlikte, 1'den başlayan dizinli segmentleri de kullanabilirsiniz." Örneğin, $metadata.topic'ın "region/west" olarak ayarlandığı durumlarda:
dataDestination |
Çözümlenen konu |
factory/${outputTopic} |
factory/region/west |
factory/${outputTopic.1} |
factory/region |
factory/${outputTopic.2} |
factory/west |
Konu değişkeni çözümlenemiyorsa (örneğin, $metadata.topic hiç ayarlanmadı), ileti bırakılır ve bir hata günlüğe kaydedilir.
Sonraki Adımlar