Usar o WASM (WebAssembly) com grafos de fluxo de dados

Os grafos de fluxo de dados de Operações do Azure IoT dão suporte a módulos WASM (WebAssembly) para processamento de dados personalizados na borda. Você pode implantar transformações de dados e lógica de negócios personalizadas como parte de seus pipelines de fluxo de dados.

Pré-requisitos

• Implante uma instância de Operações de IoT do Azure em um cluster Kubernetes habilitado para Arc. Para obter mais informações, confira Implantar Operações do Azure IoT.
• Configure o registro de contêiner e adicione as definições de grafo de exemplo e os módulos WASM seguindo as diretrizes em Implantar módulos WebAssembly (WASM) e definições de grafo.

Visão geral

Os módulos WASM (WebAssembly) nos grafos de fluxo de dados das Operações do Azure IoT permitem processar dados na borda com alto desempenho e segurança. O WASM é executado em um ambiente em área restrita e dá suporte a Rust e Python.

Como funcionam os grafos 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 com suporte e compile-a no formato de modelo de componente WebAssembly. Para saber mais, consulte:
   • Criar módulos WASM para fluxos de dados no VS Code
   • Desenvolver módulos WASM (WebAssembly)
2. Desenvolver definição de grafo: defina como os dados se movem pelos módulos usando arquivos de configuração YAML. Para saber mais, consulte Configurar definições de grafo WebAssembly.
3. Armazene artefatos no registro: envie os módulos WASM compilados e as definições de gráfico para um repositório de contêineres usando ferramentas compatíveis com OCI, como o ORAS. Para saber mais, confira Implantar módulos WebAssembly (WASM) e definições de gráficos.
4. Configurar pontos de extremidade do registro: configure detalhes de autenticação e de conexão para que as Operações do Azure IoT possam acessar o registro de contêiner. Para saber mais, confira Configurar pontos de extremidade do Registro.
5. Criar grafo de fluxo de dados: use a interface web do usuário ou os arquivos Bicep da interface de experiência operacional para definir um fluxo de dados que usa uma definição de grafo.
6. Implantar e executar: o Azure IoT Operations extrai definições de grafo e módulos WASM do registro de contêiner e os executa.

Os exemplos a seguir 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 você seguiu as etapas de exemplo em módulos WASM (Deploy WebAssembly) e definições de grafo, a definição do grafo e o graph-simple:1.0.0 módulo pré-compilado temperature:1.0.0 já estão no registro de contêiner.

Como funciona

A definição do grafo 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 MQTT

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

Formato de entrada:

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

Formato de saída:

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

A configuração a seguir cria um grafo de fluxo de dados que usa esse pipeline de conversão de temperatura. O grafo de fluxo de dados faz referência à definição do graph-simple:1.0.0 grafo YAML e extrai o módulo de temperatura do registro de contêiner.

Configurar o grafo 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 grafo que executa o módulo WASM e um nó de destino que publica os resultados convertidos.

O recurso de grafo de fluxo de dados encapsula o artefato de definição de grafo e conecta suas operações abstratas de origem/coletor a pontos de extremidade concretos:

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

Essa separação permite implantar a mesma definição de grafo com pontos de extremidade diferentes 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 agente MQTT com clientes MQTT. O cliente MQTT fornece certificados e tokens de autenticação para se conectar ao agente. 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 de temperatura aleatórios no tópico sensor/temperature/raw a cada segundo. O resultado deve ser assim:

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.

Assinar mensagens processadas

Na segunda sessão do terminal (também conectada 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ê verá 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 grafo complexo

Este exemplo demonstra um fluxo de trabalho sofisticado de processamento de dados que manipula vários tipos de dados, incluindo temperatura, umidade e dados de imagem. A definição de grafo complexo orquestra vários módulos WASM para executar análise avançada e detecção de objetos.

Como funciona

O grafo 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 umidade: acumula medidas de umidade ao longo de intervalos de tempo
• Processamento de imagem: executa a detecção de objetos em instantâneos de câmera e formata resultados

Para saber mais sobre como funciona a definição de grafo complexo, sua estrutura e o fluxo de dados por meio de vários estágios de processamento, consulte Exemplo 2: Definição de grafo complexa.

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

Configurar o grafo de fluxo de dados complexo

Essa configuração implementa o fluxo de trabalho de processamento de vários sensores usando a definição do graph-complex:1.0.0 grafo YAML. Observe como a implantação do grafo de fluxo de dados é semelhante ao Exemplo 1 – ambos usam o mesmo padrão de três nós (origem, processador de grafo, destino), mesmo que a lógica de processamento seja diferente.

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

Carregar arquivos de imagem RAW no pod mqtt-client

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

Primeiro, clone o repositório para obter 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 pasta ./samples/wasm/images para o pod mqtt-client, você pode usar o seguinte comando:

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

Verifique se os arquivos foram carregados:

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

Você deve ver a lista de arquivos na pasta /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 dados simulados de temperatura, umidade e enviar imagens

Você pode combinar os comandos para publicar dados de temperatura, umidade e enviar imagens para 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, assine o 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 seguinte exemplo:

{"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 detectados nas imagens.

Configuração de grafos de fluxo de dados personalizados

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

Visão geral do grafo de fluxo de dados

Um grafo de fluxo de dados define como os dados fluem por meio de módulos WebAssembly para processamento. Cada grafo consiste em:

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

Configuração do modo

A propriedade de modo determina se o grafo de fluxo de dados está processando ativamente os dados. Você pode definir o modo como Enabled ou Disabled (não diferencia maiúsculas de minúsculas). Quando desabilitado, o grafo interrompe o processamento de 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 grafo 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 de disco de solicitação permite que grafos de fluxo de dados mantenham o estado entre reinicializações. Quando você habilita esse recurso, o grafo pode recuperar o estado de processamento se o agente conectado for reiniciado. Esse recurso é útil para cenários de processamento com estado em que a perda de dados intermediários seria problemática. Quando você habilita a persistência de disco de solicitação, o agente mantém os dados MQTT, como mensagens na fila do assinante, em disco. Essa abordagem garante que a fonte de dados do fluxo de dados não sofra perda de dados durante interrupções de energia ou reinicializações do agente. O agente mantém um desempenho ideal porque a persistência é configurada por fluxo de dados, portanto, somente os fluxos de dados que precisam de persistência usam esse recurso.

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

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

Essa configuração permite que clientes MQTT, como grafos de fluxo de dados, solicitem persistência de disco para suas assinaturas usando propriedades de usuário MQTTv5. Para obter uma configuração de persistência detalhada do agente MQTT, consulte Configurar a persistência do agente MQTT.

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ó

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

Nós de origem

Os nós de origem definem onde os dados entram no grafo. Eles se conectam a pontos de extremidade de fluxo de dados que recebem dados de agentes MQTT ou tópicos do 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 do MQTT ou tópicos do Kafka para assinar
• Referência de ativo (opcional) que 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 do ponto de extremidade 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 grafo

Os nós de processamento de grafo 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ó de grafo requer:

• Referência de ponto de extremidade do registro que aponta para um ponto de extremidade do registro para efetuar pull de artefatos
• Especificação do artefato que define o nome e a versão do módulo a serem extraídos
• 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 recriar 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 no runtime. O módulo pode acessar esses valores para personalizar seu comportamento. Essa abordagem permite que você:

• Implantar o mesmo módulo WASM com configurações diferentes
• Ajustar parâmetros de processamento sem recompilar 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 agentes 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 o 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 Azure Data Lake ou 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 de fluxo de dados entre os nós. Cada conexão especifica um nó de origem e um nó de destino, criando o pipeline de processamento. As conexões podem incluir opcionalmente a validação de esquema para garantir a integridade dos dados entre os estágios de processamento.

Quando você especifica a validação de esquema, o sistema valida o formato de dados e a estrutura à medida que flui entre nós. A validação ajuda a capturar inconsistências de dados antecipadamente e garante que os módulos WASM recebam dados no formato esperado.

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 de extremidade de fluxo de dados

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

Pontos de extremidade do MQTT

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

• Agente MQTT local das Operações do Azure IoT (necessário em cada fluxo de dados)
• MQTT da Grade de Eventos do Azure
• Agentes MQTT personalizados

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

Pontos de extremidade do Kafka

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

• Hubs de Eventos do Azure (compatíveis com Kafka)
• Clusters do Apache Kafka
• Confluent Cloud

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

Pontos de extremidade do 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 do 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 registro

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

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

Conteúdo relacionado

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