Utiliza transformaciones de WebAssembly (WASM) en grafos de flujo de datos

Operaciones de IoT de Azure diagramas de flujo de datos incluyen transformaciones integradas para tareas de procesamiento comunes, como asignación, filtrado y agregación. Cuando necesite lógica personalizada más allá de lo que proporcionan las transformaciones integradas, puede implementar módulos WebAssembly (WASM) como transformaciones personalizadas en las canalizaciones de grafos de flujo de datos.

Prerrequisitos

• Instancia de Operaciones de IoT de Azure implementada en un clúster de Kubernetes. Para obtener más información, consulte Deploy Operaciones de IoT de Azure.

• Configure un punto de conexión del Registro para acceder a los módulos WASM y las definiciones de grafos. Tiene dos opciones:
  • Inicio rápido con el registro público: Creación de un punto de conexión del Registro para ghcr.io con autenticación anónima. Para obtener instrucciones, consulte Uso de módulos creados previamente desde un registro público.
  • Registro privado: configure su propio registro de contenedor e inserte los módulos de ejemplo siguiendo las instrucciones de Implementación de módulos y definiciones de grafos de WebAssembly (WASM).

Los CLI de Azure ejemplos de este artículo usan variables de entorno para que puedas establecer cada valor una vez y luego copiar y pegar los comandos as-is. Si usas el entorno Operaciones de IoT de Azure Codespaces del quickstart, estas variables ya están configuradas para ti y puedes saltarte este paso. De lo contrario, configura las siguientes variables de entorno en tu shell antes de ejecutar los comandos.

Los siguientes scripts establecen las variables de entorno más usadas:

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

Solo necesitas establecer las variables que utiliza este artículo. Este artículo podría utilizar variables de entorno adicionales para los nombres de recursos que elijas. El artículo explica cómo situarlos donde se introducen.

Este artículo también utiliza las siguientes variables de entorno para los valores que elijas: REGISTRY_ENDPOINT (el nombre del punto final del registro), GRAPH_NAME (el nombre del grafo de flujo de datos), PROFILE (el nombre del perfil de flujo de datos) y REGISTRY_HOST (nombre de host del registro contenedor). Configura cada uno antes de ejecutar los comandos relacionados.

Información general

Mediante el uso de módulos WebAssembly (WASM) en los gráficos de flujo de datos de Operaciones de IoT de Azure, puede procesar datos en el borde con alto rendimiento y seguridad. WASM se ejecuta en un entorno de espacio aislado y admite Rust y Python.

Un flujo de datos es una tubería que mueve y transforma datos entre extremos mediante transformaciones integradas. Un gráfico de flujo de datos amplía los flujos de datos con pasos de procesamiento que se pueden componer. Operaciones de IoT de Azure proporciona gráficos de flujo de datos integrados para operaciones comunes como asignación, filtrado, bifurcación y agregación. Para lógica de procesamiento personalizada, implementa módulos WebAssembly como describe este artículo. Los gráficos de flujo de datos usan definiciones de grafos YAML que especifican cómo se conectan los operadores. El recurso del grafo de flujo de datos envuelve esta definición y asigna sus operaciones abstractas de origen y sumidero a puntos de conexión concretos, como temas de MQTT y de Kafka.

Funcionamiento de los gráficos de flujo de datos de WASM

La implementación del flujo de datos WASM sigue este flujo de trabajo:

1. Desarrollo de módulos de WASM: escriba lógica de procesamiento personalizada en un lenguaje compatible y compílela en el formato de modelo de componentes de WebAssembly. Para saber más, consulta Construir módulos WASM para flujos de datos.
2. Desarrollar definición de grafo: defina cómo se mueven los datos a través de los módulos mediante archivos de configuración de YAML. Para obtener más información, consulte Configuración de definiciones de grafos de WebAssembly.
3. Almacenar artefactos en el registro: envíe los módulos de WASM compilados y las definiciones de grafos a un repositorio de contenedores mediante herramientas compatibles con OCI, como ORAS. Para más información, consulte Implementación de módulos y definiciones de grafos de WebAssembly (WASM).
4. Configurar puntos de conexión del Registro: configure los detalles de autenticación y conexión para que Operaciones de IoT de Azure pueda acceder al registro de contenedor. Para más información, consulte Configuración de puntos de conexión del Registro.
5. Crear grafo de flujo de datos: utilice la interfaz de usuario web del entorno de operaciones o los archivos Bicep para definir un flujo de datos que use una definición de grafo.
6. Despliegue y ejecute: Operaciones de IoT de Azure extrae definiciones de grafos y módulos WASM del registro de contenedores y los ejecuta.

En los ejemplos siguientes se muestra cómo configurar gráficos de flujo de datos WASM para escenarios comunes. Los ejemplos usan valores codificados de forma rígida y configuraciones simplificadas para que pueda empezar a trabajar rápidamente.

Ejemplo 1: Implementación básica con un módulo de WASM

En este ejemplo se convierten los datos de temperatura de Fahrenheit a Celsius mediante un módulo de WASM. El código fuente del módulo temperature está disponible en GitHub. Si ha seguido los pasos de ejemplo en Implementación de módulos de WebAssembly (WASM) y definiciones de grafos, la definición de grafo y el módulo precompilado graph-simple:1.0.0 ya están en el temperature:1.0.0 registro de contenedor. La ruta de acceso del artefacto del grafo de ejemplo es azure-samples/explore-iot-operations/graph-simple:1.0.0. Utilice esta ruta de acceso para los ejemplos del GHCR público o cuando copie los artefactos de ejemplo en su propio registro usando la misma ruta de acceso del repositorio.

Funcionamiento del grafo simple

La definición de graph crea una canalización sencilla y de tres fases:

1. Origen: recibe datos de temperatura de MQTT
2. Mapa: procesa datos con el módulo de WASM de temperatura
3. Receptor: devuelve datos convertidos a MQTT

Para obtener más información sobre cómo funciona la definición de grafo simple y su estructura, vea Ejemplo 1: Definición de grafo simple.

Formato de entrada:

{"temperature": {"value": 100.0, "unit": "F"}}

Formato de salida:

{"temperature": {"value": 37.8, "unit": "C"}}

La siguiente configuración crea un grafo de flujo de datos que usa esta canalización de conversión de temperatura. El gráfico de flujo de datos hace referencia a la graph-simple:1.0.0 definición del grafo YAML y extrae el módulo de temperatura del registro de contenedor. El grafo de ejemplo de Azure se encuentra en la ruta del repositorio azure-samples/explore-iot-operations, así que incluya esa ruta en el valor de artifact.

Configuración del grafo de flujo de datos

Esta configuración define tres nodos que implementan el flujo de trabajo de conversión de temperatura: un nodo de origen que se suscribe a los datos de temperatura entrantes, un nodo de procesamiento de grafos que ejecuta el módulo WASM y un nodo de destino que publica los resultados convertidos.

El recurso del grafo de flujo de datos encapsula el artefacto de definición del grafo y conecta sus operaciones abstractas de origen y receptor a puntos de conexión concretos:

• La operación source de la definición del grafo se conecta al nodo de origen del flujo de datos (tema MQTT)
• La operación sink de la definición del grafo se conecta al nodo de destino del flujo de datos (tema MQTT)
• Las operaciones de procesamiento de la definición del grafo se ejecutan dentro del nodo de procesamiento de grafos

Esta separación le permite implementar la misma definición de grafo con distintos puntos de conexión entre entornos, a la vez que mantiene la lógica de procesamiento sin cambios.

{
  "mode": "Enabled",
  "requestDiskPersistence": "Disabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "temperature-source",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensor/temperature/raw"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "temperature-processor",
      "graphSettings": {
        "registryEndpointRef": "<REGISTRY_ENDPOINT_NAME>",
        "artifact": "azure-samples/explore-iot-operations/graph-simple:1.0.0",
        "configuration": [
          {
            "key": "key1",
            "value": "example-value"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "temperature-destination",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "sensor/temperature/processed"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "temperature-source"
      },
      "to": {
        "name": "temperature-processor"
      }
    },
    {
      "from": {
        "name": "temperature-processor"
      },
      "to": {
        "name": "temperature-destination"
      }
    }
  ]
}

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

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_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: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'temperature-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'temperature-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'azure-samples/explore-iot-operations/graph-simple:1.0.0'
          configuration: [
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'temperature-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'sensor/temperature/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'temperature-source'
        }
        to: {
          name: 'temperature-processor'
        }
      }
      {
        from: {
          name: 'temperature-processor'
        }
        to: {
          name: 'temperature-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: temperature-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw

    - nodeType: Graph
      name: temperature-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: azure-samples/explore-iot-operations/graph-simple:1.0.0
        configuration:
          - key: key1
            value: example-value

    - nodeType: Destination
      name: temperature-destination
      destinationSettings:
        endpointRef: default
        dataDestination: sensor/temperature/processed

  nodeConnections:
    - from:
        name: temperature-source
      to:
        name: temperature-processor

    - from:
        name: temperature-processor
      to:
        name: temperature-destination

kubectl apply -f dataflow-graph.yaml

Prueba del flujo de datos

Para probar el flujo de datos, envíe mensajes MQTT desde el clúster. El módulo de temperatura espera mensajes en un formato JSON específico con un objeto anidado temperature que contiene los campos value (numérico) y unit (cadena). Por ejemplo: {"temperature":{"value":72,"unit":"F"}}.

En primer lugar, implemente el pod cliente MQTT siguiendo las instrucciones de Probar la conectividad con el corredor MQTT con clientes MQTT. El cliente MQTT proporciona los tokens de autenticación y los certificados para conectarse al agente. Para implementar el cliente MQTT, ejecute el siguiente comando:

kubectl apply -f https://raw.githubusercontent.com/Azure-Samples/explore-iot-operations/main/samples/quickstarts/mqtt-client.yaml

Enviar mensajes de temperatura

En la primera sesión de terminal, cree y ejecute un script para enviar datos de temperatura en Fahrenheit:

# Connect to the MQTT client pod
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
# Create and run temperature.sh from within the MQTT client pod
while true; do
  # Generate a random temperature value between 0 and 6000 Fahrenheit
  random_value=$(shuf -i 0-6000 -n 1)
  payload="{\"temperature\":{\"value\":$random_value,\"unit\":\"F\"}}"

  echo "Publishing temperature: $payload"

  # Publish to the input topic
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$payload" \
    -t "sensor/temperature/raw" \
    -d \
    --cafile /var/run/certs/ca.crt \
    -D PUBLISH user-property __ts $(date +%s)000:0:df \
    -D CONNECT authentication-method 'K8S-SAT' \
    -D CONNECT authentication-data $(cat /var/run/secrets/tokens/broker-sat)

  sleep 1
done'

El script publica datos de temperatura aleatorios en el tema de sensor/temperature/raw cada segundo. Debería ser parecido a este:

Publishing temperature: {"temperature":{"value":1234,"unit":"F"}}
Publishing temperature: {"temperature":{"value":5678,"unit":"F"}}

Deje el script en ejecución para continuar publicando datos de temperatura.

Suscribirse a mensajes procesados

En la segunda sesión de terminal (también conectada al pod de cliente MQTT), suscríbase al tema de salida para ver los valores de temperatura convertidos:

# Connect to the MQTT client pod
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
mosquitto_sub -h aio-broker -p 18883 -t "sensor/temperature/processed" --cafile /var/run/certs/ca.crt \
-D CONNECT authentication-method "K8S-SAT" \
-D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)"'

Ves que el módulo WASM convierte los datos de temperatura de Fahrenheit a Celsius.

{"temperature":{"value":1292.2222222222222,"count":0,"max":0.0,"min":0.0,"average":0.0,"last":0.0,"unit":"C","overtemp":false}}
{"temperature":{"value":203.33333333333334,"count":0,"max":0.0,"min":0.0,"average":0.0,"last":0.0,"unit":"C","overtemp":false}}

Ejemplo 2: Implementar un grafo complejo

En este ejemplo se muestra un flujo de trabajo sofisticado de procesamiento de datos que controla varios tipos de datos, incluidos los datos de temperatura, humedad e imagen. La definición del grafo complex organiza varios módulos WASM para realizar análisis avanzados y detección de objetos.

Cómo funciona el gráfico complejo

El complejo gráfico procesa tres secuencias de datos y las combina en análisis de sensores enriquecidos:

• Procesamiento de temperatura: convierte Fahrenheit a Celsius, filtra lecturas no válidas y calcula las estadísticas.
• Procesamiento de humedad: acumula medidas de humedad a lo largo de intervalos de tiempo
• Procesamiento de imágenes: realiza la detección de objetos en instantáneas de cámara y da formato a los resultados

Para obtener más información sobre cómo funciona la definición de grafo compleja, su estructura y el flujo de datos a través de varias fases de procesamiento, vea Ejemplo 2: Definición de grafo compleja.

El gráfico usa módulos especializados de la colección de operadores Rust.

Configurar grafo de flujo de datos complejo

Esta configuración implementa el flujo de trabajo de procesamiento de varios sensores mediante la graph-complex:1.0.0 definición del grafo YAML. La ruta de acceso del artefacto del grafo de ejemplo es azure-samples/explore-iot-operations/graph-complex:1.0.0. Observe cómo la implementación del grafo de flujo de datos es similar al ejemplo 1 : ambos usan el mismo patrón de tres nodos (origen, procesador de grafos, destino), aunque la lógica de procesamiento sea diferente.

Esta similitud se produce porque el recurso del grafo de flujo de datos actúa como un entorno host que carga y ejecuta definiciones de grafos. La lógica de procesamiento real reside en la definición del grafo (graph-simple:1.0.0 o graph-complex:1.0.0), que contiene la especificación YAML de operaciones y conexiones entre módulos WASM. El recurso de grafo de flujo de datos proporciona la infraestructura en tiempo de ejecución para extraer la definición del grafo, crear instancias de los módulos y enrutar los datos a través del flujo de trabajo definido.

{
  "mode": "Enabled",
  "requestDiskPersistence": "Disabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensor-source",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensor/temperature/raw",
          "sensor/humidity/raw",
          "sensor/images/raw"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "sensor-processor",
      "graphSettings": {
        "registryEndpointRef": "<REGISTRY_ENDPOINT_NAME>",
        "artifact": "azure-samples/explore-iot-operations/graph-complex:1.0.0",
        "configuration": [
          {
            "key": "snapshot_topic",
            "value": "sensor/images/raw"
          },
          {
            "key": "key1",
            "value": "example-value"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "analytics-destination",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "analytics/sensor/processed"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensor-source"
      },
      "to": {
        "name": "sensor-processor"
      }
    },
    {
      "from": {
        "name": "sensor-processor"
      },
      "to": {
        "name": "analytics-destination"
      }
    }
  ]
}

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

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<COMPLEX_GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_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 complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensor-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
            'sensor/humidity/raw'
            'sensor/images/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'sensor-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'azure-samples/explore-iot-operations/graph-complex:1.0.0'
          configuration: [
            {
              key: 'snapshot_topic'
              value: 'sensor/images/raw'
            }
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'analytics-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'analytics/sensor/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'sensor-source'
        }
        to: {
          name: 'sensor-processor'
        }
      }
      {
        from: {
          name: 'sensor-processor'
        }
        to: {
          name: 'analytics-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <COMPLEX_GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: sensor-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw
          - sensor/humidity/raw
          - sensor/images/raw

    - nodeType: Graph
      name: sensor-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: azure-samples/explore-iot-operations/graph-complex:1.0.0
        configuration:
          - key: snapshot_topic
            value: sensor/images/raw
          - key: key1
            value: example-value

    - nodeType: Destination
      name: analytics-destination
      destinationSettings:
        endpointRef: default
        dataDestination: analytics/sensor/processed

  nodeConnections:
    - from:
        name: sensor-source
      to:
        name: sensor-processor

    - from:
        name: sensor-processor
      to:
        name: analytics-destination

Prueba del flujo de datos complejo

Antes de ver cualquier salida, configure los datos de origen.

Carga de archivos de imagen RAW en el pod mqtt-client

Los archivos de imagen son para que el módulo snapshot detecte objetos en las imágenes. Los archivos están en la carpeta de imágenes en GitHub.

En primer lugar, clone el repositorio para obtener acceso a los archivos de imagen:

git clone https://github.com/Azure-Samples/explore-iot-operations.git
cd explore-iot-operations

Para cargar archivos de imagen RAW desde la ./samples/wasm/images carpeta al mqtt-client pod, use el siguiente comando:

kubectl cp ./samples/wasm/images azure-iot-operations/mqtt-client:/tmp

Compruebe que los archivos se cargan:

kubectl exec -it mqtt-client -n azure-iot-operations -- ls /tmp/images

Debería ver la lista de archivos en la carpeta /tmp/images.

beaker.raw          laptop.raw          sunny2.raw
binoculars.raw      lawnmower.raw       sunny4.raw
broom.raw           milkcan.raw         thimble.raw
camera.raw          photocopier.raw     tripod.raw
computer_mouse.raw  radiator.raw        typewriter.raw
daisy3.raw          screwdriver.raw     vacuum_cleaner.raw
digital_clock.raw   sewing_machine.raw
hammer.raw          sliding_door.raw

Publicación de datos de temperatura y humedad simulados y envío de imágenes

Puede combinar los comandos para publicar datos de temperatura y humedad y enviar imágenes a un solo script. Use el comando siguiente:

# Connect to the MQTT client pod and run the script
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
while true; do 
  # Generate a random temperature value between 0 and 6000
  temp_value=$(shuf -i 0-6000 -n 1)
  temp_payload="{\"temperature\":{\"value\":$temp_value,\"unit\":\"F\"}}"
  echo "Publishing temperature: $temp_payload"
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$temp_payload" \
    -t "sensor/temperature/raw" \
    --cafile /var/run/certs/ca.crt \
    -D CONNECT authentication-method "K8S-SAT" \
    -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
    -D PUBLISH user-property __ts $(date +%s)000:0:df

  # Generate a random humidity value between 30 and 90
  humidity_value=$(shuf -i 30-90 -n 1)
  humidity_payload="{\"humidity\":{\"value\":$humidity_value}}"
  echo "Publishing humidity: $humidity_payload"
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$humidity_payload" \
    -t "sensor/humidity/raw" \
    --cafile /var/run/certs/ca.crt \
    -D CONNECT authentication-method "K8S-SAT" \
    -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
    -D PUBLISH user-property __ts $(date +%s)000:0:df

  # Send an image every 2 seconds
  if [ $(( $(date +%s) % 2 )) -eq 0 ]; then
    file=$(ls /tmp/images/*.raw | shuf -n 1)
    echo "Sending file: $file"
    mosquitto_pub -h aio-broker -p 18883 \
      -f $file \
      -t "sensor/images/raw" \
      --cafile /var/run/certs/ca.crt \
      -D CONNECT authentication-method "K8S-SAT" \
      -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
      -D PUBLISH user-property __ts $(date +%s)000:0:df
  fi

  # Wait for 1 second before the next iteration
  sleep 1
done'

Consulta de la salida

En un nuevo terminal, suscríbase al tema de salida:

kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
mosquitto_sub -h aio-broker -p 18883 -t "analytics/sensor/processed" --cafile /var/run/certs/ca.crt \
-D CONNECT authentication-method "K8S-SAT" \
-D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)"'

La salida es similar al siguiente ejemplo:

{"temperature":[{"count":9,"max":2984.4444444444443,"min":248.33333333333337,"average":1849.6296296296296,"last":2612.222222222222,"unit":"C","overtemp":true}],"humidity":[{"count":10,"max":76.0,"min":30.0,"average":49.7,"last":38.0}],"object":[{"result":"milk can; broom; screwdriver; binoculars, field glasses, opera glasses; toy terrier"}]}
{"temperature":[{"count":10,"max":2490.5555555555557,"min":430.55555555555554,"average":1442.6666666666667,"last":1270.5555555555557,"unit":"C","overtemp":true}],"humidity":[{"count":9,"max":87.0,"min":34.0,"average":57.666666666666664,"last":42.0}],"object":[{"result":"broom; Saint Bernard, St Bernard; radiator"}]}

Aquí, la salida contiene los datos de temperatura y humedad, así como los objetos detectados en las imágenes.

Configuración de gráficos de flujo de datos personalizados

En esta sección se proporciona información detallada sobre cómo configurar gráficos de flujo de datos con módulos de WASM. Abarca todas las opciones de configuración, los puntos de conexión de flujo de datos y la configuración avanzada.

Introducción al grafo de flujo de datos

Un grafo de flujo de datos define cómo fluyen los datos a través de módulos de WebAssembly para su procesamiento. Cada grafo consta de:

• Modo que controla si el gráfico está habilitado o deshabilitado
• Referencia de perfil que vincula a un perfil de flujo de datos que define la configuración de escalado y recursos
• Persistencia de disco que habilita opcionalmente el almacenamiento persistente para el estado del grafo
• Nodos que definen los componentes de origen, procesamiento y destino
• Conexiones de nodo que especifican cómo fluyen los datos entre nodos

Configuración del modo

La propiedad mode determina si el gráfico de flujo de datos procesa activamente los datos. Configure el modo a Enabled o Disabled (sin distinción entre mayúsculas y minúsculas). Cuando está deshabilitado, el gráfico deja de procesar los datos, pero conserva su configuración.

{
  "mode": "Enabled"
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  // ... other properties
  properties: {
    mode: 'Enabled'  // or 'Disabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  mode: Enabled  # or Disabled
  # ... other properties

Referencia de perfil

La referencia de perfil conecta el gráfico de flujo de datos a un perfil de flujo de datos, que define la configuración de escalado, los recuentos de instancias y los límites de recursos. Si no especifica una referencia de perfil, debe usar una referencia de propietario de Kubernetes en su lugar. La mayoría de los escenarios utilizan el perfil predeterminado que proporciona Operaciones de IoT de Azure.

az iot ops dataflowgraph apply \
  --name $GRAPH_NAME \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --profile $PROFILE \
  --config-file graph.json

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  // This establishes the profile relationship
  // ... other properties
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  profileRef: default  # Reference to the profile name
  # ... other properties

Solicitud de persistencia de disco

La persistencia en disco de petición ayuda a los grafos de flujo de datos a mantener el estado entre reinicios. Cuando activas esta función, el gráfico puede recuperar el estado de procesamiento si el broker conectado se reinicia. Esta característica es útil para escenarios de procesamiento con estado en los que la pérdida de datos intermedios sería problemática. Al habilitar la persistencia del disco de solicitud, el agente conserva los datos MQTT, como los mensajes de la cola del suscriptor, en el disco. Este enfoque garantiza que el origen de datos del flujo de datos no experimente pérdida de datos durante las interrupciones de energía ni los reinicios del agente. El broker mantiene un rendimiento óptimo porque configuras la persistencia por flujo de datos, así que solo los flujos de datos que necesitan persistencia usan esta función.

El grafo de flujo de datos realiza esta solicitud de persistencia durante la suscripción usando una propiedad de usuario MQTTv5. Esta característica solo funciona cuando:

• El flujo de datos usa el corredor MQTT como origen (nodo de origen con punto de conexión MQTT)
• El corredor MQTT tiene la persistencia habilitada con el modo de persistencia dinámica establecido en Enabled para el tipo de datos, como las colas de suscriptores

Con esta configuración, los clientes MQTT, como los grafos de flujo de datos, pueden solicitar persistencia en disco para sus suscripciones utilizando las propiedades de usuario MQTTv5. Para obtener información detallada sobre la configuración de persistencia del corredor MQTT, consulte Configuración de la persistencia del corredor MQTT.

La configuración acepta Enabled o Disabled, con Disabled como valor predeterminado.

{
  "requestDiskPersistence": "Enabled"
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  // ... other properties
  properties: {
    requestDiskPersistence: 'Enabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Enabled
  # ... other properties

Reglas y límites de nomenclatura

Los recursos del grafo de flujo de datos y sus componentes tienen restricciones de nomenclatura aplicadas en diferentes capas:

El gráfico de flujo de datos también aplica las siguientes reglas estructurales:

• No hay nombres de nodo duplicados: cada nodo del grafo debe tener un nombre único.
• Tipos de conexión válidos: El grafo solo permite los siguientes tipos de conexión de nodo: Fuente a Grafo, Fuente a Destino, Grafo a Grafo y Grafo a Destino.
• Sin ciclos: el gráfico no puede contener conexiones circulares que crearían bucles de procesamiento infinitos.
• Sin bucles propios: un nodo no se puede conectar a sí mismo.
• Sin superposición de temas: si un origen y destino usan el mismo punto de conexión, sus temas MQTT no se pueden superponer, lo que crearía un bucle de mensajes infinito.

Configuración del nodo

Los nodos son los bloques de creación de un gráfico de flujo de datos. Cada nodo tiene un nombre único dentro del grafo y realiza una función específica. El gráfico incluye tres tipos de nodos:

Nodos de origen

Los nodos de origen definen dónde entran los datos en el gráfico. Se conectan a puntos de conexión de flujo de datos que reciben datos de agentes MQTT o temas de Kafka. Cada nodo de origen debe especificar:

• Referencia de punto de conexión que apunta a un punto de conexión de flujo de datos configurado.
• Orígenes de datos como una lista de temas MQTT o temas de Kafka a los que suscribirse
• Referencia de recursos (opcional) que vincula a un recurso del Registro de dispositivos de Azure para la inferencia de esquemas

El array de fuentes de datos permite suscribirse a múltiples temas sin modificar la configuración del endpoint. Esta flexibilidad significa que puedes reutilizar endpoints a través de diferentes flujos de datos.

{
  "nodeType": "Source",
  "name": "sensor-data-source",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "sensor/+/temperature",
      "sensor/+/humidity"
    ],
    "assetRef": "weather-station-asset"
  }
}

{
  nodeType: 'Source'
  name: 'sensor-data-source'
  sourceSettings: {
    endpointRef: 'default'  // References a data flow endpoint
    dataSources: [
      'sensor/+/temperature'  // MQTT topic with wildcard
      'sensor/+/humidity'
    ]
    assetRef: 'weather-station-asset'  // Optional asset reference
  }
}

- nodeType: Source
  name: sensor-data-source
  sourceSettings:
    endpointRef: default  # References a data flow endpoint
    dataSources:
      - sensor/+/temperature  # MQTT topic with wildcard
      - sensor/+/humidity
    assetRef: weather-station-asset  # Optional asset reference

Nodos de procesamiento de grafos

Los nodos de procesamiento de grafos contienen los módulos de WebAssembly que transforman los datos. Estos nodos extraen artefactos WASM de los registros de contenedor y los ejecutan mediante parámetros de configuración especificados. Cada nodo de grafo requiere:

• Referencia del punto de conexión del Registro que apunta a un punto de conexión del Registro para extraer artefactos
• Especificación de artefacto que define el nombre del módulo y la versión que se van a extraer
• Parámetros de configuración como pares clave-valor pasados al módulo de WASM

El array de configuración permite personalizar el comportamiento de los módulos sin reconstruir el artefacto WASM. Entre las opciones de configuración comunes se incluyen parámetros de procesamiento, umbrales, configuración de conversión y marcas de características.

{
  "nodeType": "Graph",
  "name": "temperature-processor",
  "graphSettings": {
    "registryEndpointRef": "my-acr-endpoint",
    "artifact": "temperature-converter:2.1.0",
    "configuration": [
      {
        "key": "temperature_lower_bound",
        "value": "-40"
      },
      {
        "key": "temperature_upper_bound",
        "value": "3422"
      }
    ]
  }
}

{
  nodeType: 'Graph'
  name: 'temperature-processor'
  graphSettings: {
    registryEndpointRef: 'my-acr-endpoint'
    artifact: 'temperature-converter:2.1.0'
    configuration: [
      {
        key: 'temperature_lower_bound'
        value: '-40'
      }
      {
        key: 'temperature_upper_bound'
        value: '3422'
      }
    ]
  }
}

- nodeType: Graph
  name: temperature-processor
  graphSettings:
    registryEndpointRef: my-acr-endpoint
    artifact: temperature-converter:2.1.0
    configuration:
      - key: temperature_lower_bound
        value: "-40"
      - key: temperature_upper_bound
        value: "3422"

Pasar los pares clave-valor de configuración al módulo WASM en tiempo de ejecución. El módulo puede acceder a estos valores para personalizar su comportamiento. Utilizando este enfoque, puedes:

• Implemente el mismo módulo WASM con configuraciones diferentes.
• Ajuste los parámetros de procesamiento sin volver a generar módulos.
• Habilite o deshabilite las características en función de los requisitos de implementación.
• Establezca valores específicos del entorno como umbrales o puntos de conexión.

Nodos de destino

Los nodos de destino definen dónde envía el grafo los datos procesados. Se conectan a extremos de flujo de datos que envían datos a brokers MQTT u otros sistemas. Cada nodo de destino especifica:

• Referencia de punto de conexión que apunta a un punto de conexión de flujo de datos configurado.
• Destino de datos como tema específico, ruta o ubicación para los datos de salida.
• Configuración de esquema de salida (opcional) que definen el formato de serialización y la validación del esquema.

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "processed-data/temperature",
  }
}

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'processed-data/temperature'
  }
}

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

Conexiones de nodo

Las conexiones de nodo definen la ruta de acceso del flujo de datos entre los nodos. Cada conexión especifica un nodo de origen y un nodo de destino, creando la canalización de procesamiento. Opcionalmente, puede incluir un esquema en la conexión. El módulo recibe el esquema en la inicialización, lo que soporta la validación del esquema como en este ejemplo.

"nodeConnections": [
  {
    "from": {
      "name": "sensor-data-source",
      "schema": {
        "schemaRef": "aio-sr://sensor-input-schema:1",
        "serializationFormat": "Json"
      }
    },
    "to": {
      "name": "temperature-processor"
    }
  },
  {
    "from": {
      "name": "temperature-processor"
    },
    "to": {
      "name": "output"
    }
  }
]

nodeConnections: [
  {
    from: {
      name: 'sensor-data-source'
      schema: {
        schemaRef: 'aio-sr://sensor-input-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'temperature-processor'
    }
  }
  {
    from: {
      name: 'temperature-processor'
    }
    to: {
      name: 'output'
    }
  }
]

nodeConnections:
  - from:
      name: sensor-data-source
      schema:
        schemaRef: aio-sr://sensor-input-schema:1
        serializationFormat: Json
    to:
      name: temperature-processor
  - from:
      name: temperature-processor
    to:
      name: output

Puntos de conexión del flujo de datos

Los grafos de flujo de datos se conectan a sistemas externos a través de puntos de conexión de flujo de datos. El tipo de punto de conexión determina si puede usarlo como origen, destino o ambos.

Puntos de conexión MQTT

Los puntos de conexión MQTT pueden servir como orígenes y destinos. Se conectan a corredores MQTT, entre los que se incluyen:

• MQTT broker local de Operaciones de IoT de Azure (necesario en cada flujo de datos)
• Azure Event Grid MQTT
• Corredores MQTT personalizados

Para obtener información detallada sobre la configuración, consulte Configuración de puntos de conexión de flujo de datos de MQTT.

Puntos de conexión de Kafka

Los puntos de conexión de Kafka pueden servir como orígenes y destinos. Se conectan a sistemas compatibles con Kafka, entre los que se incluyen:

• Azure Event Hubs (compatible con Kafka)
• Clústeres de Apache Kafka
• Nube Confluente

Para obtener información detallada sobre la configuración, consulte Configurar Azure Event Hubs y los puntos de conexión del flujo de datos de Kafka.

Puntos de conexión del registro

Los puntos de conexión del Registro proporcionan acceso a los registros de contenedor para extraer módulos de WASM y definiciones de grafos. No se usan directamente en el flujo de datos, pero los nodos de procesamiento de grafos hacen referencia a ellos.

Para obtener información detallada sobre la configuración, consulte Configuración de puntos de conexión del Registro.

Solución de problemas de gráficos de flujo de datos

En esta sección se proporcionan sugerencias para solucionar problemas comunes al trabajar con gráficos de flujo de datos.

No se encontró RegistryEndpoint

Si el gráfico de flujo de datos no se inicia e informa de que no encuentra el punto de conexión del Registro, compruebe lo siguiente:

1. Coincidencias del nombre del punto de conexión del Registro: el valor registryEndpointRef del gráfico de flujo de datos debe coincidir exactamente con el name del recurso RegistryEndpoint. Compruebe si hay errores tipográficos y distinción entre mayúsculas y minúsculas.

   # List all registry endpoints in the namespace
   kubectl get registryendpoints -n azure-iot-operations
   

2. El punto de conexión del Registro está en el espacio de nombres correcto: el punto de conexión del Registro debe estar en el azure-iot-operations espacio de nombres (o en el mismo espacio de nombres que el gráfico de flujo de datos).

3. El punto de conexión del registro está listo: Compruebe el estado de su punto de conexión del registro:

   kubectl describe registryendpoint $REGISTRY_ENDPOINT -n azure-iot-operations
   

4. La autenticación está configurada correctamente: si utiliza una identidad administrada, asegúrese de que la extensión Arc de Operaciones de IoT de Azure tiene permisos de AcrPull en el registro. Si usa la autenticación anónima con un registro público, compruebe que la dirección URL del host es correcta.

5. Los artefactos existen en el Registro: compruebe que la definición del grafo y los módulos WASM a los que se hace referencia en el grafo están disponibles en las etiquetas esperadas del Registro:

   # Check if artifacts exist (example with ORAS)
   oras manifest fetch $REGISTRY_HOST/graph-simple:1.0.0
   

El gráfico de flujo de datos se está ejecutando, pero no procesando datos

Si implementa el gráfico de flujo de datos, pero no procesa los mensajes:

1. Comprobación del estado del gráfico de flujo de datos: busque errores en el estado del recurso del grafo de flujo de datos.

   kubectl get dataflowgraph $GRAPH_NAME -n azure-iot-operations -o yaml
   

2. Compruebe los temas de MQTT: asegúrese de que los temas de origen del gráfico de flujo de datos coinciden con los temas en los que está publicando datos.

3. Comprobar marcas de tiempo: los gráficos de flujo de datos usan marcas de tiempo del reloj lógico híbrido (HLC) para el procesamiento de mensajes. Incluye la __ts propiedad de usuario cuando publiques mensajes MQTT para asegurar un procesamiento oportuno.

Contenido relacionado

• Construcción de módulos WASM para flujos de datos en flujos de trabajo de desarrollo basados en IDE y CLI
• Configura las definiciones de grafos de WebAssembly para la estructura YAML y los parámetros de configuración de grafos
• Implementación de módulos y definiciones de grafos de WASM para la configuración del registro y la administración de artefactos
• Ejecutar inferencia ONNX en WASM para la integración de modelos de aprendizaje automático
• Configuración de puntos de conexión de flujo de datos MQTT
• Configurar los puntos de conexión de flujo de datos de Azure Event Hubs y Kafka
• Configurar puntos de conexión de flujo de datos de Azure Data Lake Storage
• Configurar puntos de conexión de flujo de datos de OneLake en Microsoft Fabric