Route readings by severity

Send high-severity readings through an alert path, then merge them with normal readings. This example first drops readings with temperature > 1000. It adds alert: true when severity > 5 and renames fields on both paths.

Prerequisites

This standalone example uses the existing default MQTT endpoint, the default profile and registry endpoint, telemetry/sensors as input, and telemetry/processed as output. For deployment prerequisites, Bicep parent resources, and Azure CLI environment variables (AIO_INSTANCE_NAME and RESOURCE_GROUP), see Create a data flow graph.

• An instance of Azure IoT Operations deployed in a Kubernetes cluster. For more information, see Deploy Azure IoT Operations.

The Azure CLI examples in this article use environment variables so that you can set each value once and then copy and paste the commands as-is. If you're using the Azure IoT Operations Codespaces environment from the quickstart, these variables are already set for you and you can skip this step. Otherwise, set the following environment variables in your shell before you run the commands.

The following scripts set the most commonly used environment variables:

Environment variable	Description
SUBSCRIPTION_ID	The ID of the subscription that contains your Azure IoT Operations instance.
RESOURCE_GROUP	The name of the resource group that contains your Azure IoT Operations instance.
AIO_INSTANCE_NAME	The name of your Azure IoT Operations instance. To list your instances, run az iot ops list -o table.
CLUSTER_NAME	The name of the Azure Arc-enabled Kubernetes cluster that hosts your instance.
LOCATION	The Azure region to use for new resources, for example eastus.

Bash

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

PowerShell

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

You only need to set the variables that this article uses. This article might use additional environment variables for resource names that you choose. The article explains how to set them where they're introduced.

The example assumes numeric temperature and severity fields and a deviceId; it doesn't perform schema validation.

Configure the severity routes

Concatenate forwards each processed message independently; it doesn't join records or guarantee ordering across paths. The example renames fields but doesn't convert temperature units.

Synthetic input on telemetry/sensors	Output on telemetry/processed
{"deviceId":"sensor-01","temperature":20,"severity":2}	{"id":"sensor-01","temp":20}
{"deviceId":"sensor-01","temperature":20,"severity":8}	{"id":"sensor-01","temp":20,"alert":true}
{"deviceId":"sensor-01","temperature":1001,"severity":8}	Dropped

Operations experience

To build this pipeline in the operations experience:

1. Create a data flow graph and add a source that reads from telemetry/sensors.
2. Add a filter transform. Configure a rule that drops messages where temperature > 1000.
3. Add a branch transform. Configure the condition severity > 5 to route high-severity messages to the true path.
4. Add a map transform on the true path. Configure rules to rename deviceId to id, temperature to temp, and add a field alert set to true.
5. Add a map transform on the false path. Configure rules to rename deviceId to id and temperature to temp.
6. Add a concatenate transform to merge both paths.
7. Add a destination that sends to telemetry/processed.
8. Connect the elements: source → filter → branch → (true path: alert map, false path: normal map) → concatenate → destination.

Azure CLI

Save the following graph properties as graph.json. Transform rules are escaped JSON strings. The Kubernetes tab shows the same rules in a readable form.

{
  "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.1.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"
      }
    }
  ]
}

Apply the config file through Azure Resource Manager. This approach avoids an azure-iot-ops extension validation issue with named outputs such as severity-check.output.true.

INSTANCE_ID=$(az iot ops show --name "$AIO_INSTANCE_NAME" --resource-group "$RESOURCE_GROUP" --query id --output tsv)
EXTENDED_LOCATION=$(az iot ops show --name "$AIO_INSTANCE_NAME" --resource-group "$RESOURCE_GROUP" --query extendedLocation --output json)
GRAPH_ID="$INSTANCE_ID/dataflowProfiles/default/dataflowGraphs/alert-routing"

az rest --method put \
  --url "$GRAPH_ID?api-version=2026-07-01" \
  --body "{\"extendedLocation\":$EXTENDED_LOCATION,\"properties\":$(cat graph.json)}"

az resource wait --created --ids "$GRAPH_ID" --api-version 2026-07-01

Bicep

Save this template as main.bicep. It references the existing instance, custom location, and default profile.

param aioInstanceName string
param customLocationName string

resource aioInstance 'Microsoft.IoTOperations/instances@2026-07-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-07-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-07-01' = {
  name: 'alert-routing'
  parent: defaultDataflowProfile
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    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.1.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' } }
    ]
  }
}

Deploy the template to the instance's resource group. Set CUSTOM_LOCATION_NAME to the existing custom location name.

az deployment group create \
  --resource-group $RESOURCE_GROUP \
  --template-file main.bicep \
  --parameters aioInstanceName=$AIO_INSTANCE_NAME customLocationName=$CUSTOM_LOCATION_NAME

Kubernetes (debug only)

Important

The use of Kubernetes deployment manifests isn't supported in production environments and should only be used for debugging and testing.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: alert-routing
  namespace: azure-iot-operations
spec:
  profileRef: default
  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.1.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 }

Related content

• Filter, branch, and merge messages
• Transform data with map
• Route more than two message kinds
• Common use cases and patterns