Uso de WebAssembly (WASM) con gráficos de flujo de datos

Los gráficos de flujo de datos de Operaciones de IoT de Azure admiten módulos de WebAssembly (WASM) para el procesamiento de datos personalizado en el borde. Puede implementar transformaciones de datos y lógica de negocios personalizadas como parte de las canalizaciones de flujo de datos.

Prerrequisitos

• Implemente una instancia de Azure IoT Operations en un clúster de Kubernetes habilitado para Arc. Para más información, consulte Implementación de Operaciones de IoT de Azure.
• Configure el registro de contenedor y agregue las definiciones de grafos de ejemplo y los módulos WASM siguiendo las instrucciones de Implementación de módulos webAssembly (WASM) y definiciones de grafos.

Información general

Los módulos de WebAssembly (WASM) de los gráficos de flujo de datos de Operaciones de IoT de Azure permiten procesar datos en el perímetro con un alto rendimiento y seguridad. WASM se ejecuta en un entorno de espacio aislado y admite Rust y Python.

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 obtener más información, consulte lo siguiente:
   • Compilación de módulos WASM para flujos de datos en VS Code
   • Desarrollo de módulos de WebAssembly (WASM)
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. Implementación y ejecución: Azure IoT Operations extrae definiciones de grafos y módulos WASM del registro de contenedor 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 de temperatura 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.

Cómo funciona

La definición del grafo crea una canalización simple 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.

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 o 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.

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@2025-10-01' existing = {
  name: aioInstanceName
}

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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: '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: 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. 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)"'

Verá los datos de temperatura convertidos de Fahrenheit a Celsius por el módulo de WASM.

{"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 de grafo compleja organiza varios módulos de WASM para realizar análisis avanzados y detección de objetos.

Cómo funciona

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 de Rust.

Configurar grafo de flujo de datos complejo

Esta configuración implementa el flujo de trabajo de procesamiento de varios sensores mediante la definición del graph-complex:1.0.0 grafo YAML. 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.

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@2025-10-01' existing = {
  name: aioInstanceName
}

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

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

resource complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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: '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: 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. Se encuentran en la carpeta images de 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 carpeta ./samples/wasm/images al pod mqtt-client, puede usar 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

Publicar datos de temperatura y humedad simulados y envío de imágenes

Puede combinar los comandos para publicar datos de temperatura, 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 modo determina si el gráfico de flujo de datos está procesando activamente los datos. Puede establecer el modo en 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.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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 usan el perfil predeterminado proporcionado por Operaciones de IoT de Azure.

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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 del disco de solicitud permite a los grafos de flujo de datos mantener el estado entre reinicios. Al habilitar esta característica, el grafo puede recuperar el estado de procesamiento si se reinicia el agente conectado. 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 agente mantiene un rendimiento óptimo porque la persistencia se configura por flujo de datos, por lo que solo los flujos de datos que necesitan persistencia usan esta característica.

El grafo de flujo de datos realiza esta solicitud de persistencia durante la suscripción mediante 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

Esta configuración permite a los clientes MQTT, como grafos de flujo de datos, solicitar persistencia de disco para sus suscripciones mediante 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.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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

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. Hay 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 de Azure Device Registry para la inferencia de esquema

La matriz de orígenes de datos permite suscribirse a varios temas sin modificar la configuración del punto de conexión. Esta flexibilidad permite la reutilización de puntos de conexión en distintos flujos de datos.

{
  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 de WASM de registros de contenedor y los ejecutan con 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

La matriz de configuración permite personalizar el comportamiento del módulo sin volver a generar el artefacto de 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: 'input-unit'
        value: 'fahrenheit'
      }
      {
        key: 'output-unit' 
        value: 'celsius'
      }
      {
        key: 'precision'
        value: '2'
      }
      {
        key: 'enable-filtering'
        value: 'true'
      }
    ]
  }
}

- nodeType: Graph
  name: temperature-processor
  graphSettings:
    registryEndpointRef: my-acr-endpoint
    artifact: temperature-converter:2.1.0
    configuration:
      - key: input-unit
        value: fahrenheit
      - key: output-unit
        value: celsius
      - key: precision
        value: "2"
      - key: enable-filtering
        value: "true"

Los pares clave-valor de configuración se pasan al módulo de WASM en tiempo de ejecución. El módulo puede acceder a estos valores para personalizar su comportamiento. Este enfoque le permite:

• Implementación del mismo módulo de WASM con configuraciones diferentes
• Ajuste de los parámetros de procesamiento sin volver a generar módulos
• Habilitar o deshabilitar características en función de los requisitos de implementación
• Establecer valores específicos del entorno como umbrales o puntos de conexión

Nodos de destino

Los nodos de destino definen dónde se envían los datos procesados. Se conectan a puntos de conexión de flujo de datos que envían datos a corredores MQTT, almacenamiento en la nube 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, ruta de acceso o ubicación específicos para los datos de salida
• Configuración del esquema de salida (opcional) que define el formato de serialización y la validación de esquemas

Para destinos de almacenamiento como Azure Data Lake o Fabric OneLake, puede especificar la configuración del esquema de salida para controlar cómo se serializan y validan los datos.

{
  nodeType: 'Destination'
  name: 'cloud-storage-destination'
  destinationSettings: {
    endpointRef: 'azure-storage-endpoint'
    dataDestination: 'processed-data/temperature'
    outputSchemaSettings: {
      serializationFormat: 'Parquet'
      schemaRef: 'temperature-output-schema:1'
    }
  }
}

- nodeType: Destination
  name: cloud-storage-destination
  destinationSettings:
    endpointRef: azure-storage-endpoint
    dataDestination: processed-data/temperature
    outputSchemaSettings:
      serializationFormat: Parquet
      schemaRef: temperature-output-schema:1

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. Las conexiones pueden incluir opcionalmente el esquema para que el esquema se proporcione al módulo en la inicialización, lo que permite la validación de esquemas como en este ejemplo.

nodeConnections: [
  {
    from: {
      name: 'sensor-data-source'
      schema: {
        schemaRef: 'sensor-input-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'temperature-processor'
    }
  }
  {
    from: {
      name: 'temperature-processor'
    }
    to: {
      name: 'cloud-storage-destination'
    }
  }
]

nodeConnections:
  - from:
      name: sensor-data-source
      schema:
        schemaRef: sensor-input-schema:1
        serializationFormat: Json
    to:
      name: temperature-processor
  - from:
      name: temperature-processor
    to:
      name: cloud-storage-destination

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 se puede usar 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:

• Corredor MQTT local de Operaciones de Azure IoT (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
• Confluent Cloud

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

Puntos de conexión de almacenamiento

Los puntos de conexión de almacenamiento solo pueden servir como destinos. Se conectan a sistemas de almacenamiento en la nube para la retención y el análisis de datos a largo plazo:

• Azure Data Lake Storage
• Microsoft Fabric OneLake
• Almacenamiento local

Los puntos de conexión de almacenamiento suelen requerir la configuración del esquema de salida para definir el formato de serialización de datos.

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.

Contenido relacionado

• Configuración de puntos de conexión de flujo de datos MQTT
• Configuración de puntos de conexión de flujo de datos de Azure Event Hubs y Kafka
• Configuración de puntos de conexión de flujo de datos de Azure Data Lake Storage
• Configuración de puntos de conexión de flujo de datos de OneLake de Microsoft Fabric