Usar WebAssembly (WASM) com gráficos de fluxo de dados

Os gráficos de fluxo de dados das Operações IoT do Azure suportam módulos WebAssembly (WASM) para processamento de dados personalizado na borda. Você pode implantar lógica de negócios personalizada e transformações de dados como parte de seus pipelines de fluxo de dados.

Pré-requisitos

• Implante uma instância de Operações IoT do Azure em um cluster Kubernetes habilitado para Arc. Para obter mais informações, consulte Implantar operações do Azure IoT.
• Configure o seu registo de contentores e adicione as definições de grafos de exemplo e módulos WASM seguindo as orientações dos módulos Deploy WebAssembly (WASM) e das definições de grafos.

Visão geral

Os módulos WebAssembly (WASM) nos gráficos de fluxo de dados do Azure IoT Operations permitem processar dados na borda com alto desempenho e segurança. O WASM é executado em um ambiente sandbox e suporta Rust e Python.

Como funcionam os gráficos de fluxo de dados WASM

A implementação do fluxo de dados WASM segue este fluxo de trabalho:

1. Desenvolver módulos WASM: escreva a lógica de processamento personalizada em uma linguagem suportada e compile-a para o formato WebAssembly Component Model. Para saber mais, consulte:
   • Construir módulos WASM para fluxos de dados no VS Code
   • Desenvolver módulos WebAssembly (WASM)
2. Desenvolver definição de gráfico: Defina como os dados se movem pelos módulos usando arquivos de configuração YAML. Para saber mais, consulte Configurar definições de grafos WebAssembly.
3. Armazene artefactos no registo: Envie os módulos WASM compilados e as definições de grafos para um registo de contentores utilizando ferramentas compatíveis com OCI, como o ORAS. Para saber mais, consulte os módulos Deploy WebAssembly (WASM) e as definições de grafos.
4. Configurar pontos de extremidade do Registro: configure os detalhes de autenticação e conexão para que as Operações IoT do Azure possam acessar o registro do contêiner. Para saber mais, consulte Configurar endpoints do registo.
5. Criar grafo de fluxo de dados: Use a interface web da experiência operacional ou ficheiros Bicep para definir um fluxo de dados que use uma definição de grafo.
6. Implementar e executar: O Azure IoT Operations retira definições de grafos e módulos WASM do registo de contentores e executa-os.

Os exemplos seguintes mostram como configurar grafos de fluxo de dados WASM para cenários comuns. Os exemplos usam valores codificados e configurações simplificadas para que você possa começar rapidamente.

Exemplo 1: Implantação básica com um módulo WASM

Este exemplo converte dados de temperatura de Fahrenheit para Celsius usando um módulo WASM. O código-fonte do módulo de temperatura está disponível no GitHub. Se seguiu os passos de exemplo em Implementar módulos WebAssembly (WASM) e definições de grafos, a graph-simple:1.0.0 definição de grafo e o módulo pré-compilado temperature:1.0.0 já estão no seu registo de contentores.

Como funciona

A definição do gráfico cria um pipeline simples de três estágios:

1. Fonte: Recebe dados de temperatura do MQTT
2. Mapa: Processa dados com o módulo WASM de temperatura
3. Coletor: envia dados convertidos de volta para o MQTT

Para saber mais sobre como funciona a definição simples de grafo e a sua estrutura, veja Exemplo 1: Definição simples de grafo.

Formato de entrada:

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

Output format (Formato de saída):

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

A configuração a seguir cria um gráfico de fluxo de dados que usa esse pipeline de conversão de temperatura. O gráfico de fluxo de dados faz referência à graph-simple:1.0.0 definição do grafo YAML e retira o módulo de temperatura do seu registo de contentores.

Configurar o gráfico de fluxo de dados

Essa configuração define três nós que implementam o fluxo de trabalho de conversão de temperatura: um nó de origem que assina dados de temperatura de entrada, um nó de processamento de gráfico que executa o módulo WASM e um nó de destino que publica os resultados convertidos.

O recurso do grafo de fluxo de dados envolve o artefacto de definição do grafo e liga as suas operações abstratas de fonte/sumidouro a pontos finais concretos:

• A operação da source definição do gráfico se conecta ao nó de origem do fluxo de dados (tópico MQTT)
• A operação da sink definição do gráfico se conecta ao nó de destino do fluxo de dados (tópico MQTT)
• As operações de processamento da definição do gráfico são executadas dentro do nó de processamento do gráfico

Essa separação permite implantar a mesma definição de gráfico com diferentes pontos de extremidade entre ambientes, mantendo a lógica de processamento inalterada.

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

Testar o fluxo de dados

Para testar o fluxo de dados, envie mensagens MQTT de dentro do cluster. Primeiro, implante o pod do cliente MQTT seguindo as instruções em Testar conectividade com o broker MQTT com clientes MQTT. O cliente MQTT fornece os tokens de autenticação e certificados para se conectar ao broker. Para implantar o cliente MQTT, execute o seguinte comando:

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

Enviar mensagens de temperatura

Na primeira sessão de terminal, crie e execute um script para enviar dados de temperatura em 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'

O script publica dados aleatórios de temperatura para o sensor/temperature/raw tópico a cada segundo. Deve ter a seguinte aparência:

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

Deixe o script em execução para continuar publicando dados de temperatura.

Subscrever mensagens processadas

Na segunda sessão do terminal (também conectado ao pod do cliente MQTT), assine o tópico de saída para ver os 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)"'

Você vê os dados de temperatura convertidos de Fahrenheit para Celsius pelo módulo 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}}

Exemplo 2: Implantar um gráfico complexo

Este exemplo demonstra um fluxo de trabalho de processamento de dados sofisticado que lida com vários tipos de dados, incluindo dados de temperatura, umidade e imagem. A definição de gráfico complexo orquestra vários módulos WASM para executar análises avançadas e deteção de objetos.

Como funciona

O gráfico complexo processa três fluxos de dados e os combina em análises de sensores enriquecidas:

• Processamento de temperatura: converte Fahrenheit em Celsius, filtra leituras inválidas e calcula estatísticas
• Processamento de humidade: Acumula medições de humidade ao longo de intervalos de tempo
• Processamento de imagem: Executa a deteção de objetos em instantâneos da câmera e formata resultados

Para saber mais sobre como funciona a definição de grafo complexo, a sua estrutura e o fluxo de dados através de múltiplas etapas de processamento, consulte o Exemplo 2: Definição de grafos complexos.

O grafo utiliza módulos especializados da coleção de operadores Rust.

Configurar o gráfico de fluxo de dados complexo

Esta configuração implementa o fluxo de trabalho de processamento multi-sensor usando a graph-complex:1.0.0 definição do grafo YAML. Repare como a implementação do grafo de fluxo de dados é semelhante ao Exemplo 1 – ambos usam o mesmo padrão de três nós (fonte, processador do grafo, destino), mesmo que a lógica de processamento seja diferente.

Essa semelhança ocorre porque o recurso de gráfico de fluxo de dados atua como um ambiente de host que carrega e executa definições de gráfico. A lógica de processamento real reside na definição do grafo (graph-simple:1.0.0 ou graph-complex:1.0.0), que contém a especificação YAML de operações e ligações entre módulos WASM. O recurso do grafo de fluxo de dados fornece a infraestrutura de execução para extrair a definição do grafo, instanciar os módulos e encaminhar os dados através do fluxo de trabalho 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

Testar o fluxo de dados complexo

Antes de veres qualquer saída, configura os dados de origem.

Carregue arquivos de imagem RAW para o pod mqtt-client

Os arquivos de imagem são para o snapshot módulo para detetar objetos nas imagens. Eles estão localizados na pasta de imagens no GitHub.

Primeiro, clone o repositório para ter acesso aos arquivos de imagem:

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

Para carregar arquivos de imagem RAW da ./samples/wasm/images pasta para o mqtt-client pod, você pode usar o seguinte comando:

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

Verifique se os ficheiros foram carregados:

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

Você deve ver a /tmp/images lista de arquivos na pasta.

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

Publique dados simulados de temperatura, umidade e envie imagens

Você pode combinar os comandos para publicar dados de temperatura, umidade e enviar imagens em um único script. Use o seguinte comando:

# 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'

Verifique a saída

Em um novo terminal, inscreva-se no tópico de saída:

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

A saída se parece com o exemplo a seguir:

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

Aqui, a saída contém os dados de temperatura e umidade, bem como os objetos detetados nas imagens.

Configuração de gráficos de fluxo de dados personalizados

Esta seção fornece informações detalhadas sobre como configurar gráficos de fluxo de dados com módulos WASM. Ele abrange todas as opções de configuração, pontos de extremidade de fluxo de dados e configurações avançadas.

Visão geral do fluxograma de dados

Um gráfico de fluxo de dados define como os dados fluem através dos módulos WebAssembly para processamento. Cada gráfico é composto por:

• Modo que controla se o gráfico está ativado ou desativado
• Referência de perfil que se vincula a um perfil de fluxo de dados que define configurações de dimensionamento e recursos
• Persistência de disco que, opcionalmente, habilita o armazenamento persistente para o estado do gráfico
• Nós que definem os componentes de origem, processamento e destino
• Conexões de nó que especificam como os dados fluem entre nós

Configuração do modo

A propriedade mode determina se o gráfico de fluxo de dados está processando dados ativamente. Você pode definir o modo como Enabled ou Disabled (sem distinção entre maiúsculas e minúsculas). Quando desativado, o gráfico para de processar dados, mas mantém sua configuração.

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

Referência de perfil

A referência de perfil conecta seu gráfico de fluxo de dados a um perfil de fluxo de dados, que define configurações de dimensionamento, contagens de instâncias e limites de recursos. Se você não especificar uma referência de perfil, deverá usar uma referência de proprietário do Kubernetes. A maioria dos cenários usa o perfil padrão fornecido pelas Operações do Azure IoT.

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

Solicitar persistência de disco

A persistência do disco de solicitação permite que os gráficos de fluxo de dados mantenham o estado nas reinicializações. Quando você habilita esse recurso, o gráfico pode recuperar o estado de processamento se o broker conectado for reiniciado. Esse recurso é útil para cenários de processamento com monitoração de estado em que a perda de dados intermediários seria problemática. Quando você habilita a persistência de disco de solicitação, o broker persiste os dados MQTT, como mensagens na fila de assinantes, no disco. Essa abordagem garante que a fonte de dados do seu fluxo de dados não sofra perda de dados durante quedas de energia ou reinicializações do broker. O broker mantém o desempenho ideal porque a persistência é configurada por fluxo de dados, portanto, apenas os fluxos de dados que precisam de persistência usam esse recurso.

O gráfico de fluxo de dados faz essa solicitação de persistência durante a assinatura usando uma propriedade de usuário MQTTv5. Esta funcionalidade só funciona quando:

• O fluxo de dados usa o broker MQTT como fonte (nó de origem com ponto de extremidade MQTT)
• O broker MQTT tem persistência habilitada com o modo de persistência dinâmica definido para Enabled o tipo de dados, como filas de assinantes

Essa configuração permite que clientes MQTT, como gráficos de fluxo de dados, solicitem persistência de disco para suas assinaturas usando propriedades de usuário MQTTv5. Para obter uma configuração detalhada da persistência do broker MQTT, consulte Configure MQTT broker persistence.

A configuração aceita Enabled ou Disabled, com Disabled como padrão.

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

Configuração do nó

Os nós são os blocos de construção de um gráfico de fluxo de dados. Cada nó tem um nome exclusivo dentro do gráfico e executa uma função específica. Existem três tipos de nós:

Nós de origem

Os nós de origem definem onde os dados entram no gráfico. Eles se conectam a pontos de extremidade de fluxo de dados que recebem dados de corretores MQTT ou tópicos Kafka. Cada nó de origem deve especificar:

• Referência de ponto de extremidade que aponta para um ponto de extremidade de fluxo de dados configurado
• Fontes de dados como uma lista de tópicos MQTT ou tópicos Kafka para assinar
• Referência de ativo (opcional) que se vincula a um ativo do Registro de Dispositivo do Azure para inferência de esquema

A matriz de fontes de dados permite que você assine vários tópicos sem modificar a configuração do ponto de extremidade. Essa flexibilidade permite a reutilização de pontos finais em diferentes fluxos de dados.

{
  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

Nós de processamento de gráficos

Os nós de processamento de gráficos contêm os módulos WebAssembly que transformam dados. Esses nós extraem artefatos WASM de registros de contêiner e os executam com parâmetros de configuração especificados. Cada nó gráfico requer:

• Referência de ponto de extremidade do Registro que aponta para um ponto de extremidade do Registro para extrair artefatos
• Especificação de artefato que define o nome do módulo e a versão a ser extraída
• Parâmetros de configuração como pares chave-valor passados para o módulo WASM

A matriz de configuração permite personalizar o comportamento do módulo sem reconstruir o artefato WASM. As opções de configuração comuns incluem parâmetros de processamento, limites, configurações de conversão e sinalizadores de recursos.

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

Os pares chave-valor de configuração são passados para o módulo WASM em tempo de execução. O módulo pode acessar esses valores para personalizar seu comportamento. Esta abordagem permite-lhe:

• Implante o mesmo módulo WASM com configurações diferentes
• Ajuste os parâmetros de processamento sem reconstruir módulos
• Habilitar ou desabilitar recursos com base nos requisitos de implantação
• Definir valores específicos do ambiente, como limites ou pontos de extremidade

Nós de destino

Os nós de destino definem para onde os dados processados são enviados. Eles se conectam a pontos de extremidade de fluxo de dados que enviam dados para corretores MQTT, armazenamento em nuvem ou outros sistemas. Cada nó de destino especifica:

• Referência de ponto de extremidade que aponta para um ponto de extremidade de fluxo de dados configurado
• Destino de dados como tópico, caminho ou local específico para dados de saída
• Configurações de esquema de saída (opcional) que definem o formato de serialização e a validação de esquema

Para destinos de armazenamento como o Azure Data Lake ou o Fabric OneLake, você pode especificar configurações de esquema de saída para controlar como os dados são serializados e validados.

{
  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

Conexões de nó

As conexões de nó definem o caminho do fluxo de dados entre nós. Cada conexão especifica um nó de origem e um nó de destino, criando o pipeline de processamento. As ligações podem opcionalmente incluir o esquema para que o esquema seja fornecido ao módulo na inicialização, permitindo a validação do esquema como neste exemplo.

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

Pontos finais de fluxo de dados

Os gráficos de fluxo de dados se conectam a sistemas externos por meio de pontos finais de fluxo de dados. O tipo de ponto de extremidade determina se ele pode ser usado como origem, destino ou ambos:

Pontos finais MQTT

Os pontos de extremidade MQTT podem servir como fontes e destinos. Eles se conectam a corretores MQTT, incluindo:

• Agente MQTT local do Azure IoT Operations (necessário em cada fluxo de dados)
• Azure Event Grid MQTT
• Corretores MQTT personalizados

Para obter informações detalhadas sobre a configuração, consulte Configurar pontos de extremidade de fluxo de dados MQTT.

Pontos finais de Kafka

Os pontos finais de Kafka podem servir como fontes e destinos. Eles se conectam a sistemas compatíveis com Kafka, incluindo:

• Hubs de Eventos do Azure (compatível com Kafka)
• Clusters Apache Kafka
• Nuvem confluente

Para obter informações detalhadas sobre configuração, consulte Configurar Hubs de Eventos do Azure e pontos de extremidade de fluxo de dados Kafka.

Pontos finais de armazenamento

Os pontos de extremidade de armazenamento só podem servir como destinos. Eles se conectam a sistemas de armazenamento em nuvem para retenção e análise de dados de longo prazo:

• Armazenamento Azure Data Lake
• Microsoft Fabric OneLake
• Armazenamento local

Os pontos de extremidade de armazenamento normalmente exigem configurações de esquema de saída para definir o formato de serialização de dados.

Pontos de extremidade do Registo

Os pontos de extremidade do Registro fornecem acesso a registros de contêiner para extrair módulos WASM e definições de gráficos. Eles não são usados diretamente no fluxo de dados, mas os nós de processamento de gráficos fazem referência a eles.

Para obter informações detalhadas sobre a configuração, consulte Configurar pontos de extremidade do Registro.

Conteúdo relacionado

• Configurar pontos de extremidade de fluxo de dados MQTT
• Configurar Hubs de Eventos do Azure e pontos de extremidade de fluxo de dados Kafka
• Configurar pontos de extremidade de fluxo de dados do Armazenamento do Azure Data Lake
• Configurar pontos de extremidade de fluxo de dados do Microsoft Fabric OneLake