Configure dataflows in Azure IoT Operations

Important

Azure IoT Operations Preview – enabled by Azure Arc is currently in preview. You shouldn't use this preview software in production environments.

You'll need to deploy a new Azure IoT Operations installation when a generally available release is made available. You won't be able to upgrade a preview installation.

See the Supplemental Terms of Use for Microsoft Azure Previews for legal terms that apply to Azure features that are in beta, preview, or otherwise not yet released into general availability.

A dataflow is the path that data takes from the source to the destination with optional transformations. You can configure the dataflow by creating a Dataflow custom resource or using the Azure IoT Operations Studio portal. A dataflow is made up of three parts: the source, the transformation, and the destination.

To define the source and destination, you need to configure the dataflow endpoints. The transformation is optional and can include operations like enriching the data, filtering the data, and mapping the data to another field.

This article shows you how to create a dataflow with an example, including the source, transformation, and destination.

Prerequisites

• An instance of Azure IoT Operations Preview
• A configured dataflow profile
• Dataflow endpoints. For example, create a dataflow endpoint for the local MQTT broker. You can use this endpoint for both the source and destination. Or, you can try other endpoints like Kafka, Event Hubs, or Azure Data Lake Storage. To learn how to configure each type of dataflow endpoint, see Configure dataflow endpoints.

Create dataflow

Once you have dataflow endpoints, you can use them to create a dataflow. Recall that a dataflow is made up of three parts: the source, the transformation, and the destination.

Portal

To create a dataflow in the operations experience portal, select Dataflow > Create dataflow.

Kubernetes

The overall structure of a dataflow configuration is as follows:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

Review the following sections to learn how to configure the operation types of the dataflow.

Configure a source with a dataflow endpoint to get data

To configure a source for the dataflow, specify the endpoint reference and data source. You can specify a list of data sources for the endpoint.

Portal

Use Asset as a source

You can use an asset as the source for the dataflow. This is only available in the operations experience portal.

1. Under Source details, select Asset.

2. Select the asset you want to use as the source endpoint.

3. Select Proceed.

   A list of datapoints for the selected asset is displayed.

   

4. Select Apply to use the asset as the source endpoint.

Kubernetes

Configuring an asset as a source is only available in the operations experience portal.

Use MQTT as a source

Portal

1. Under Source details, select MQTT.

2. Enter the MQTT Topic that you want to listen to for incoming messages.

3. Choose a Message schema from the dropdown list or upload a new schema. If the source data has optional fields or fields with different types, specify a deserialization schema to ensure consistency. For example, the data might have fields that aren't present in all messages. Without the schema, the transformation can't handle these fields as they would have empty values. With the schema, you can specify default values or ignore the fields.

   

4. Select Apply.

Kubernetes

For example, to configure a source using an MQTT endpoint and two MQTT topic filters, use the following configuration:

sourceSettings:
  endpointRef: mq
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/humidity/#

Because dataSources allows you to specify MQTT or Kafka topics without modifying the endpoint configuration, you can reuse the endpoint for multiple dataflows even if the topics are different. To learn more, see Reuse dataflow endpoints.

Specify schema to deserialize data

If the source data has optional fields or fields with different types, specify a deserialization schema to ensure consistency. For example, the data might have fields that aren't present in all messages. Without the schema, the transformation can't handle these fields as they would have empty values. With the schema, you can specify default values or ignore the fields.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

To specify the schema, create the file and store it in the schema registry.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Note

The only supported serialization format is JSON. The schema is optional.

For more information about schema registry, see Understand message schemas.

Shared subscriptions

To use shared subscriptions with MQTT sources, you can specify the shared subscription topic in the form of $shared/<subscription-group>/<topic>.

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Note

If the instance count in the dataflow profile is greater than 1, then the shared subscription topic must be used.

Configure transformation to process data

The transformation operation is where you can transform the data from the source before you send it to the destination. Transformations are optional. If you don't need to make changes to the data, don't include the transformation operation in the dataflow configuration. Multiple transformations are chained together in stages regardless of the order in which they're specified in the configuration. The order of the stages is always

1. Enrich: Add additional data to the source data given a dataset and condition to match.
2. Filter: Filter the data based on a condition.
3. Map: Move data from one field to another with an optional conversion.

Portal

In the operations experience portal, select Dataflow > Add transform (optional).

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

Enrich: Add reference data

To enrich the data, you can use the reference dataset in the Azure IoT Operations distributed state store (DSS). The dataset is used to add extra data to the source data based on a condition. The condition is specified as a field in the source data that matches a field in the dataset.

Key names in the distributed state store correspond to a dataset in the dataflow configuration.

Portal

Currently, the enrich operation isn't available in the operations experience portal.

Kubernetes

For example, you could use the deviceId field in the source data to match the asset field in the dataset:

builtInTransformationSettings:
  datasets:
  - key: assetDataset
    inputs:
      - $source.deviceId # ------------- $1
      - $context(assetDataset).asset # - $2
    expression: $1 == $2

If the dataset has a record with the asset field, similar to:

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

The data from the source with the deviceId field matching thermostat1 has the location and manufacturer fields available in filter and map stages.

You can load sample data into the DSS by using the DSS set tool sample.

For more information about condition syntax, see Enrich data by using dataflows and Convert data using dataflows.

Filter: Filter data based on a condition

To filter the data on a condition, you can use the filter stage. The condition is specified as a field in the source data that matches a value.

Portal

1. Under Transform (optional), select Filter > Add.

2. Choose the datapoints to include in the dataset.

3. Add a filter condition and description.

   

4. Select Apply.

Kubernetes

For example, you could use the temperature field in the source data to filter the data:

builtInTransformationSettings:
  filter:
    - inputs:
      - temperature ? $last # - $1
      expression: "$1 > 20"

If the temperature field is greater than 20, the data is passed to the next stage. If the temperature field is less than or equal to 20, the data is filtered.

Map: Move data from one field to another

To map the data to another field with optional conversion, you can use the map operation. The conversion is specified as a formula that uses the fields in the source data.

Portal

In the operations experience portal, mapping is currently supported using Compute transforms.

1. Under Transform (optional), select Compute > Add.

2. Enter the required fields and expressions.

   

3. Select Apply.

Kubernetes

For example, you could use the temperature field in the source data to convert the temperature to Celsius and store it in the temperatureCelsius field. You could also enrich the source data with the location field from the contextualization dataset:

builtInTransformationSettings:
  map:
    - inputs:
      - temperature # - $1
      output: temperatureCelsius
      expression: "($1 - 32) * 5/9"
    - inputs:
      - $context(assetDataset).location  
      output: location

To learn more, see Map data by using dataflows and Convert data by using dataflows.

Serialize data according to a schema

If you want to serialize the data before sending it to the destination, you need to specify a schema and serialization format. Otherwise, the data is serialized in JSON with the types inferred. Remember that storage endpoints like Microsoft Fabric or Azure Data Lake require a schema to ensure data consistency.

Portal

Specify the Output schema when you add the destination dataflow endpoint.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

To specify the schema, you can create a Schema custom resource with the schema definition.

For more information about schema registry, see Understand message schemas.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

Supported serialization formats are JSON, Parquet, and Delta.

Configure destination with a dataflow endpoint to send data

To configure a destination for the dataflow, specify the endpoint reference and data destination. You can specify a list of data destinations for the endpoint which are MQTT or Kafka topics.

Portal

1. Select the dataflow endpoint to use as the destination.

   

2. Select Proceed to configure the destination.

3. Add the mapping details based on the type of destination.

Kubernetes

For example, to configure a destination using the MQTT endpoint created earlier and a static MQTT topic, use the following configuration:

destinationSettings:
  endpointRef: mq
  dataDestination: factory

If you've created storage endpoints like Microsoft Fabric, use the data destination field to specify the table or container name:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

Example

The following example is a dataflow configuration that uses the MQTT endpoint for the source and destination. The source filters the data from the MQTT topics thermostats/+/telemetry/temperature/# and humidifiers/+/telemetry/humidity/#. The transformation converts the temperature to Fahrenheit and filters the data where the temperature is less than 100000. The destination sends the data to the MQTT topic factory.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

Verify a dataflow is working

Follow Tutorial: Bi-directional MQTT bridge to Azure Event Grid to verify the dataflow is working.

Export dataflow configuration

To export the dataflow configuration, you can use the operations experience portal or by exporting the Dataflow custom resource.

Portal

Select the dataflow you want to export and select Export from the toolbar.

Kubernetes

kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml