Um grafo de fluxo de dados é um pipeline de processamento composável que transforma dados à medida que se move entre fontes e destinos. Um fluxo de dados padrão segue a sequência de enriquecer, filtrar, mapear. Um grafo de fluxo de dados permite encadear transformações em qualquer ordem, ramificar em caminhos paralelos e agregar dados ao longo das janelas de tempo.
Este artigo explica como criar um grafo de fluxo de dados passo a passo. Para obter uma visão geral dos grafos de fluxo de dados e das transformações disponíveis, consulte a visão geral dos grafos de fluxo de dados.
Atualmente, os grafos de fluxo de dados dão suporte apenas a endpoints MQTT, Kafka e OpenTelemetry. Eles não suportam outros tipos de endpoints como Data Lake, Microsoft Fabric OneLake, Azure Data Explorer e Local Storage.
Transformadas usam uma linguagem de expressão para calcular valores, condições de teste e campos de referência. Expressões referem-se a entradas por posição, não pelo nome: a primeira entrada na inputs lista é $1, a segunda é $2, e assim por diante. Funções integradas como cToF convertem e manipulam esses valores.
Para a lista completa de operadores, funções, tipos de dados e campos de metadados, veja a referência Expressions.
Pré-requisitos
- Uma instância de Operações do Azure IoT implantada em um cluster do Kubernetes. Para obter mais informações, consulte Deploy Operações do Azure IoT.
O CLI do Azure instalado em seu computador de desenvolvimento. Verifique as extensões disponíveis do CLI do Azure para a versão mínima exigida para usar a extensão azure-iot-ops. Use az --version para verificar sua versão e az upgrade para atualizar, se necessário. Para obter mais informações, consulte Instale o CLI do Azure.
A extensão de Operações do Azure IoT da CLI do Azure. Use o seguinte comando para adicionar a extensão ou atualizá-la à versão mais recente:
az extension add --upgrade --name azure-iot-ops
Os CLI do Azure exemplos deste artigo usam variáveis de ambiente para que você possa definir cada valor uma vez e então copiar e colar os comandos as-is. Se você está usando o ambiente Operações do Azure IoT Codespaces do quickstart, essas variáveis já estão definidas para você e você pode pular essa etapa. Caso contrário, defina as seguintes variáveis de ambiente no seu shell antes de executar os comandos.
Os seguintes scripts definem as variáveis de ambiente mais comumente usadas:
| Variável de ambiente |
Descrição |
SUBSCRIPTION_ID |
O ID da assinatura que contém sua instância do Operações do Azure IoT. |
RESOURCE_GROUP |
O nome do grupo de recursos que contém sua instância do Operações do Azure IoT. |
AIO_INSTANCE_NAME |
O nome da sua instância do Operações do Azure IoT. Para listar suas instâncias, execute az iot ops list -o table. |
CLUSTER_NAME |
O nome do cluster Kubernetes habilitado para Azure Arc que hospeda sua instância. |
LOCATION |
A região Azure para usar para novos recursos, por exemploeastus, . |
SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>
$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"
Você só precisa definir as variáveis que este artigo utiliza. Este artigo pode usar variáveis adicionais de ambiente para nomes de recursos que você escolher. O artigo explica como posicioná-los onde são apresentados.
Este artigo também utiliza as GRAPH_NAME variáveis de ambiente e PROFILE para o gráfico de fluxo de dados e nomes de perfis. Defina cada um antes de executar os comandos relacionados.
Criar um grafo de fluxo de dados
Um grafo de fluxo de dados contém três tipos de elementos: fontes que trazem dados, transformações que os processam e destinos que os enviam. Conecte-os na ordem em que você deseja que os dados fluam.
Na experiência de operações, vá para sua instância do Operações do Azure IoT.
Selecione Grafo de fluxo de dados>Criar grafo de fluxo de dados.
Insira um nome para o grafo de fluxo de dados e selecione um perfil de fluxo de dados. O perfil padrão é selecionado por padrão.
Crie seu pipeline adicionando elementos ao canvas.
Adicione uma fonte: selecione o ponto de extremidade de origem e configure os tópicos nos quais você deseja se inscrever para receber as mensagens de entrada.
Adicionar transformações: selecione uma ou mais transformações para processar os dados. As transformações disponíveis incluem map, filter, branch, concatenate, window e throttle. Para obter detalhes sobre cada tipo de transformação, consulte a visão geral dos grafos de fluxo de dados.
Adicionar um destino: selecione o ponto de extremidade de destino e configure o tópico ou caminho para o qual enviar dados processados.
Conecte os elementos na ordem em que você deseja que os dados fluam.
Selecione Salvar para implantar o grafo de fluxo de dados.
A CLI do Azure usa um grafo de fluxo de dados a partir de um único arquivo de configuração JSON que contém todos os nós e conexões. Use az iot ops dataflowgraph apply para criar ou substituir o grafo. O exemplo a seguir lê dados de temperatura, converte para Fahrenheit e envia para um tópico de destino.
Crie um graph.json arquivo com as propriedades do grafo de fluxo de dados. No graph.json arquivo, armazene as regras de cada transformação no value campo como uma string JSON escapada. Para obter a forma legível das regras de cada transformação, consulte o instruções para esse tipo de transformação.
{
"mode": "Enabled",
"nodes": [
{
"nodeType": "Source",
"name": "sensors",
"sourceSettings": {
"endpointRef": "default",
"dataSources": [
"telemetry/temperature"
]
}
},
{
"nodeType": "Graph",
"name": "convert",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
}
]
}
},
{
"nodeType": "Destination",
"name": "output",
"destinationSettings": {
"endpointRef": "default",
"dataDestination": "telemetry/converted"
}
}
],
"nodeConnections": [
{
"from": {
"name": "sensors"
},
"to": {
"name": "convert"
}
},
{
"from": {
"name": "convert"
},
"to": {
"name": "output"
}
}
]
}
Aplique o arquivo de configuração.
az iot ops dataflowgraph apply \
--name temperature-processing \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP \
--config-file graph.json
O gráfico utiliza o default perfil de fluxo de dados. Para usar um perfil diferente, adicione --profile <PROFILE_NAME>.
Crie um arquivo Bicep .bicep com a estrutura a seguir. Este exemplo cria um grafo de fluxo de dados que lê dados de temperatura, converte-os em Fahrenheit e os envia para um tópico de destino.
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
name: customLocationName
}
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
parent: aioInstance
name: 'default'
}
resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
parent: defaultDataflowProfile
name: 'temperature-processing'
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
profileRef: 'default'
mode: 'Enabled'
nodes: [
{
nodeType: 'Source'
name: 'sensors'
sourceSettings: {
endpointRef: 'default'
dataSources: [
'telemetry/temperature'
]
}
}
{
nodeType: 'Graph'
name: 'convert'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
}
]
}
}
{
nodeType: 'Destination'
name: 'output'
destinationSettings: {
endpointRef: 'default'
dataDestination: 'telemetry/converted'
}
}
]
nodeConnections: [
{
from: { name: 'sensors' }
to: { name: 'convert' }
}
{
from: { name: 'convert' }
to: { name: 'output' }
}
]
}
}
Implante o arquivo Bicep:
az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep
Importante
O uso de manifestos de implantação do Kubernetes não tem suporte em ambientes de produção e só deve ser usado para depuração e teste.
Crie um arquivo de manifesto .yaml do Kubernetes com a estrutura a seguir. Este exemplo cria um grafo de fluxo de dados que lê dados de temperatura, converte-os em Fahrenheit e os envia para um tópico de destino.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
name: temperature-processing
namespace: azure-iot-operations
spec:
profileRef: default
nodes:
- nodeType: Source
name: sensors
sourceSettings:
endpointRef: default
dataSources:
- telemetry/temperature
- nodeType: Graph
name: convert
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"map": [
{
"inputs": ["*"],
"output": "*"
},
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
- nodeType: Destination
name: output
destinationSettings:
endpointRef: default
dataDestination: telemetry/converted
nodeConnections:
- from:
name: sensors
to:
name: convert
- from:
name: convert
to:
name: output
Aplique o manifesto:
kubectl apply -f main.yaml
A fonte define onde os dados entram no pipeline. Especifique uma referência de ponto de extremidade e um ou mais tópicos.
No editor do grafo de fluxo de dados, selecione o elemento de origem e configure:
| Configurações |
Descrição |
|
Ponto de extremidade |
O ponto de extremidade do fluxo de dados a ser utilizado. Selecione o padrão para o agente MQTT local. |
|
Topics |
Um ou mais tópicos para se inscrever no recebimento de mensagens. |
A CLI aplica o grafo inteiro de uma vez, portanto configure a fonte como um nó Source no seu arquivo de configuração graph.json e depois execute az iot ops dataflowgraph apply:
{
"nodeType": "Source",
"name": "sensors",
"sourceSettings": {
"endpointRef": "default",
"dataSources": [
"telemetry/temperature",
"telemetry/humidity"
]
}
}
{
nodeType: 'Source'
name: 'sensors'
sourceSettings: {
endpointRef: 'default'
dataSources: [
'telemetry/temperature'
'telemetry/humidity'
]
}
}
Importante
O uso de manifestos de implantação do Kubernetes não tem suporte em ambientes de produção e só deve ser usado para depuração e teste.
- nodeType: Source
name: sensors
sourceSettings:
endpointRef: default
dataSources:
- telemetry/temperature
- telemetry/humidity
Transforma dados de processo entre a origem e o destino. Cada transformação faz referência a um artefato embutido e possui regras de configuração.
As transformações integradas disponíveis são:
| Transformar |
Artefato |
Descrição |
|
Map |
azureiotoperations/graph-dataflow-map:1.0.0 |
Renomear, reestruturar, computar e copiar campos. |
|
Filter |
azureiotoperations/graph-dataflow-filter:1.0.0 |
Solte mensagens que correspondam a uma condição. |
|
Branch |
azureiotoperations/graph-dataflow-branch:1.0.0 |
Encaminhe cada mensagem para um true ou false caminho com base em uma condição. |
|
Concatenate |
azureiotoperations/graph-dataflow-concatenate:1.0.0 |
Mescle dois ou mais caminhos em um só. |
|
Window |
azureiotoperations/graph-dataflow-window:1.0.0 |
Colete mensagens em um intervalo de tempo e, em seguida, agregar. |
|
Throttle |
azureiotoperations/graph-dataflow-throttle:1.0.0 |
Limite a taxa de mensagens por padrão de tópico do MQTT. |
Para obter mais informações sobre como enriquecer mensagens com dados externos, consulte Enriquecer com dados externos.
No editor do grafo de fluxo de dados, selecione Adicionar transformação e escolha o tipo de transformação. Configure as regras no editor visual.
Cada transformação é um nó com nodeType definido como Graph no arquivo de configuração graph.json. As regras da transformação são um objeto JSON, como este mapa que converte a temperatura em Fahrenheit:
{
"map": [
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
A configuration propriedade aceita essas regras como uma cadeia, então escape do JSON das regras e coloque no value campo. Aplique o grafo completo com az iot ops dataflowgraph apply:
{
"nodeType": "Graph",
"name": "convert",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
}
]
}
}
Dica
Para gerar a string escapada, salve as regras em um arquivo como rules.json, execute jq -c . rules.json e cole a saída em uma única linha no campo value.
Cada transformação é um nó com nodeType: 'Graph'. A configuration propriedade passa regras como uma cadeia de caracteres JSON:
{
nodeType: 'Graph'
name: 'convert'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
}
]
}
}
Importante
O uso de manifestos de implantação do Kubernetes não tem suporte em ambientes de produção e só deve ser usado para depuração e teste.
Cada transformação é um nó com nodeType: Graph. A configuration propriedade passa regras como uma cadeia de caracteres JSON:
- nodeType: Graph
name: convert
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"map": [
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
Encadeie qualquer número de transformações. Conecte-os na nodeConnections seção na ordem em que você deseja que os dados fluam:
Arraste conexões entre transformações na tela para definir a ordem de processamento.
Defina a ordem de processamento na seção nodeConnections do seu arquivo de configuração graph.json:
"nodeConnections": [
{
"from": {
"name": "sensors"
},
"to": {
"name": "remove-bad-data"
}
},
{
"from": {
"name": "remove-bad-data"
},
"to": {
"name": "convert"
}
},
{
"from": {
"name": "convert"
},
"to": {
"name": "output"
}
}
]
nodeConnections: [
{ from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
{ from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
{ from: { name: 'convert' }, to: { name: 'output' } }
]
Importante
O uso de manifestos de implantação do Kubernetes não tem suporte em ambientes de produção e só deve ser usado para depuração e teste.
nodeConnections:
- from: { name: sensors }
to: { name: remove-bad-data }
- from: { name: remove-bad-data }
to: { name: convert }
- from: { name: convert }
to: { name: output }
O destino define para onde os dados processados vão. Especifique uma referência de ponto de extremidade e um tópico ou caminho.
Selecione o elemento de destino e configure:
| Configurações |
Descrição |
|
Ponto de extremidade |
O ponto de extremidade de fluxo de dados para o qual enviar dados. |
|
Topic |
O tópico ou caminho para o qual publicar dados processados. |
Configure o destino como um nó Destination no seu arquivo de configuração graph.json, depois aplique o grafo completo com az iot ops dataflowgraph apply:
{
"nodeType": "Destination",
"name": "output",
"destinationSettings": {
"endpointRef": "default",
"dataDestination": "telemetry/processed"
}
}
{
nodeType: 'Destination'
name: 'output'
destinationSettings: {
endpointRef: 'default'
dataDestination: 'telemetry/processed'
}
}
Importante
O uso de manifestos de implantação do Kubernetes não tem suporte em ambientes de produção e só deve ser usado para depuração e teste.
- nodeType: Destination
name: output
destinationSettings:
endpointRef: default
dataDestination: telemetry/processed
Para o roteamento dinâmico de tópicos com base no conteúdo da mensagem, consulte Encaminhar mensagens para tópicos diferentes.
Verifique se o grafo de fluxo de dados está funcionando
Depois de implantar um grafo de fluxo de dados, verifique se ele está em execução:
Na experiência de operações, selecione seu gráfico de fluxo de dados para visualizar seu status. Um grafo saudável mostra um estado em execução.
Use az iot ops dataflowgraph show para exibir os detalhes de um grafo:
az iot ops dataflowgraph show \
--name temperature-processing \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP
Para listar todos os grafos de fluxo de dados associados a um perfil, use az iot ops dataflowgraph list:
az iot ops dataflowgraph list \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP
Verifique o status do DataflowGraph recurso:
az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE
Importante
O uso de manifestos de implantação do Kubernetes não tem suporte em ambientes de produção e só deve ser usado para depuração e teste.
kubectl get dataflowgraph temperature-processing -n azure-iot-operations
Verifique se há erros nos logs do pod:
kubectl logs -l app=dataflow -n azure-iot-operations --tail=50
Conteúdo relacionado
