Create a data flow graph in Azure IoT Operations

A data flow graph is a composable processing pipeline that transforms data as it moves between sources and destinations. A standard data flow follows a fixed enrich, filter, map sequence. A data flow graph lets you chain transforms in any order, branch into parallel paths, and aggregate data over time windows.

This article walks through creating a data flow graph step by step. For an overview of data flow graphs and the available transforms, see Data flow graphs overview.

Data flow graphs currently support only MQTT, Kafka, and OpenTelemetry endpoints. They don't support other endpoint types such as Data Lake, Microsoft Fabric OneLake, Azure Data Explorer, and Local Storage.

Transforms use an expression language to compute values, test conditions, and reference fields. Expressions refer to inputs by position, not by name: the first input in the inputs list is $1, the second is $2, and so on. Built-in functions such as cToF convert and manipulate those values.

For the complete list of operators, functions, data types, and metadata fields, see the Expressions reference.

Prerequisites

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

The Azure CLI installed on your development machine. Check Available Azure CLI extensions for the minimum required version to use the azure-iot-ops extension. Use az --version to check your version and az upgrade to update if necessary. For more information, see Install the Azure CLI.
The Azure IoT Operations extension for the Azure CLI. Use the following command to add the extension or update it to the latest version:
```
az extension add --upgrade --name azure-iot-ops
```

Azure IoT Operations version 1.2 or later.
A data flow profile. You can use the default profile.
A data flow endpoint for your source and destination. The default MQTT broker endpoint works for getting started.

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
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>"

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.

This article also uses the GRAPH_NAME and PROFILE environment variables for the data flow graph and profile names. Set each one before you run the related commands.

Create a data flow graph

A data flow graph contains three types of elements: sources that bring data in, transforms that process it, and destinations that send it out. Connect them in the order you want data to flow.

In the operations experience, go to your Azure IoT Operations instance.
Select Data flow graph > Create data flow graph.
Enter a name for the data flow graph and select a data flow profile. The default profile is selected by default.
Build your pipeline by adding elements to the canvas:
1. Add a source: Select the source endpoint and configure the topics to subscribe to for incoming messages.
2. Add transforms: Select one or more transforms to process the data. Available transforms include map, filter, branch, concatenate, window, and throttle. For details on each transform type, see Data flow graphs overview.
3. Add a destination: Select the destination endpoint and configure the topic or path to send processed data to.
Connect the elements in the order you want data to flow.
Select Save to deploy the data flow graph.

The Azure CLI uses a data flow graph from a single JSON config file that contains all nodes and connections. Use az iot ops dataflowgraph apply to create or replace the graph. The following example reads temperature data, converts it to Fahrenheit, and sends it to a destination topic.

Create a graph.json file with the data flow graph properties. In the graph.json file, store each transform's rules in the value field as an escaped JSON string. For the readable form of each transform's rules, see the how-to for that transform type.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "convert",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/converted"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "convert"
      }
    },
    {
      "from": {
        "name": "convert"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Apply the config file.

az iot ops dataflowgraph apply \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

The graph uses the default data flow profile. To use a different profile, add --profile <PROFILE_NAME>.

Create a Bicep .bicep file with the following structure. This example creates a data flow graph that reads temperature data, converts it to Fahrenheit, and sends it to a destination topic.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'

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

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: 'temperature-processing'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    profileRef: 'default'
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'telemetry/temperature'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'convert'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/converted'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'sensors' }
        to: { name: 'convert' }
      }
      {
        from: { name: 'convert' }
        to: { name: 'output' }
      }
    ]
  }
}

Deploy the Bicep file:

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep

Important

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

Create a Kubernetes manifest .yaml file with the following structure. This example creates a data flow graph that reads temperature data, converts it to Fahrenheit, and sends it to a destination topic.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: temperature-processing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/temperature

    - nodeType: Graph
      name: convert
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "temperature_f",
                    "expression": "cToF($1)"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/converted

  nodeConnections:
    - from:
        name: sensors
      to:
        name: convert
    - from:
        name: convert
      to:
        name: output

Apply the manifest:

kubectl apply -f main.yaml

Configure the source

The source defines where data enters the pipeline. Specify an endpoint reference and one or more topics.

In the data flow graph editor, select the source element and configure:

Setting	Description
Endpoint	The data flow endpoint to use. Select default for the local MQTT broker.
Topics	One or more topics to subscribe to for incoming messages.

The CLI applies the whole graph at once, so configure the source as a Source node in your graph.json config file, then run az iot ops dataflowgraph apply:

{
  "nodeType": "Source",
  "name": "sensors",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "telemetry/temperature",
      "telemetry/humidity"
    ]
  }
}

{
  nodeType: 'Source'
  name: 'sensors'
  sourceSettings: {
    endpointRef: 'default'
    dataSources: [
      'telemetry/temperature'
      'telemetry/humidity'
    ]
  }
}

Important

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

- nodeType: Source
  name: sensors
  sourceSettings:
    endpointRef: default
    dataSources:
      - telemetry/temperature
      - telemetry/humidity

Add transforms

Transforms process data between the source and destination. Each transform references a built-in artifact and has configuration rules.

The available built-in transforms are:

Transform	Artifact	Description
Map	`azureiotoperations/graph-dataflow-map:1.0.0`	Rename, restructure, compute, and copy fields.
Filter	`azureiotoperations/graph-dataflow-filter:1.0.0`	Drop messages that match a condition.
Branch	`azureiotoperations/graph-dataflow-branch:1.0.0`	Route each message to a `true` or `false` path based on a condition.
Concatenate	`azureiotoperations/graph-dataflow-concatenate:1.0.0`	Merge two or more paths back into one.
Window	`azureiotoperations/graph-dataflow-window:1.0.0`	Collect messages over a time interval, then aggregate.
Throttle	`azureiotoperations/graph-dataflow-throttle:1.0.0`	Limit the message rate per MQTT topic pattern.

For more information on enriching messages with external data, see Enrich with external data.

In the data flow graph editor, select Add transform and choose the transform type. Configure the rules in the visual editor.

Each transform is a node with nodeType set to Graph in your graph.json config file. The transform's rules are a JSON object, like this map that converts temperature to Fahrenheit:

{
  "map": [
    {
      "inputs": ["temperature"],
      "output": "temperature_f",
      "expression": "cToF($1)"
    }
  ]
}

The configuration property takes these rules as a string, so escape the rules JSON and place it in the value field. Apply the full graph with az iot ops dataflowgraph apply:

{
  "nodeType": "Graph",
  "name": "convert",
  "graphSettings": {
    "registryEndpointRef": "default",
    "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
    "configuration": [
      {
        "key": "rules",
        "value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
      }
    ]
  }
}

Tip

To generate the escaped string, save the rules to a file like rules.json, run jq -c . rules.json, and paste the single-line output into the value field.

Each transform is a node with nodeType: 'Graph'. The configuration property passes rules as a JSON string:

{
  nodeType: 'Graph'
  name: 'convert'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
    configuration: [
      {
        key: 'rules'
        value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
      }
    ]
  }
}

Important

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

Each transform is a node with nodeType: Graph. The configuration property passes rules as a JSON string:

- nodeType: Graph
  name: convert
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-map:1.0.0
    configuration:
      - key: rules
        value: |
          {
            "map": [
              {
                "inputs": ["temperature"],
                "output": "temperature_f",
                "expression": "cToF($1)"
              }
            ]
          }

Chain multiple transforms

Chain any number of transforms. Connect them in the nodeConnections section in the order you want data to flow:

Drag connections between transforms on the canvas to define the processing order.

Define the processing order in the nodeConnections section of your graph.json config file:

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "remove-bad-data"
    }
  },
  {
    "from": {
      "name": "remove-bad-data"
    },
    "to": {
      "name": "convert"
    }
  },
  {
    "from": {
      "name": "convert"
    },
    "to": {
      "name": "output"
    }
  }
]

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
  { from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
  { from: { name: 'convert' }, to: { name: 'output' } }
]

Important

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

nodeConnections:
  - from: { name: sensors }
    to: { name: remove-bad-data }
  - from: { name: remove-bad-data }
    to: { name: convert }
  - from: { name: convert }
    to: { name: output }

Configure the destination

The destination defines where processed data goes. Specify an endpoint reference and a topic or path.

Select the destination element and configure:

Setting	Description
Endpoint	The data flow endpoint to send data to.
Topic	The topic or path to publish processed data to.

Configure the destination as a Destination node in your graph.json config file, then apply the full graph with az iot ops dataflowgraph apply:

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "telemetry/processed"
  }
}

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'telemetry/processed'
  }
}

Important

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

- nodeType: Destination
  name: output
  destinationSettings:
    endpointRef: default
    dataDestination: telemetry/processed

For dynamic topic routing based on message content, see Route messages to different topics.

Verify the data flow graph is working

After you deploy a data flow graph, verify it's running:

In the operations experience, select your data flow graph to view its status. A healthy graph shows a Running state.

Use az iot ops dataflowgraph show to view a graph's details:

az iot ops dataflowgraph show \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

To list all data flow graphs associated with a profile, use az iot ops dataflowgraph list:

az iot ops dataflowgraph list \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Check the status of the DataflowGraph resource:

az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE

Important

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

kubectl get dataflowgraph temperature-processing -n azure-iot-operations

Check the pod logs for any errors:

kubectl logs -l app=dataflow -n azure-iot-operations --tail=50

Feedback

Was this page helpful?

Last updated on 2026-07-31

Create a data flow graph in Azure IoT Operations

Prerequisites

Create a data flow graph

Configure the source

Add transforms

Chain multiple transforms

Configure the destination

Verify the data flow graph is working

Related content

Feedback

Additional resources