Um grafo de fluxo de dados é um pipeline de processamento componível que transforma os dados à medida que estes se deslocam entre fontes e destinos. Um fluxo de dados padrão segue uma sequência fixa de enriquecimento, filtro e mapa. Um grafo de fluxo de dados permite encadear transformações em qualquer ordem, ramificar em caminhos paralelos e agregar dados ao longo de janelas temporais.
Este artigo explica passo a passo a processo a criação de um gráfico de fluxo de dados. Para uma visão geral dos gráficos de fluxo de dados e das transformadas disponíveis, consulte Visão Geral dos Gráficos de Fluxo de Dados.
Atualmente, os grafos de fluxo de dados suportam apenas os endpoints MQTT, Kafka e OpenTelemetry. Não suportam outros tipos de endpoints como Data Lake, Microsoft Fabric OneLake, Azure Data Explorer e Local Storage.
As transformações utilizam uma linguagem de expressões 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 sucessivamente. Funções incorporadas como cToF convertem e manipulam esses valores.
Para a lista completa de operadores, funções, tipos de dados e campos de metadados, consulte a referência Expressões.
Pré-requisitos
O CLI do Azure instalado na sua máquina de desenvolvimento. Verifique as extensões disponíveis do CLI do Azure para a versão mínima necessária para usar a extensão azure-iot-ops. Utilize az --version para verificar a sua versão e az upgrade para atualizar, se necessário. Para mais informações, consulte Instale o CLI do Azure.
A extensão Operações IoT do Azure para a CLI do Azure. Use o seguinte comando para adicionar a extensão ou atualizá-la para a 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 possas definir cada valor uma vez e depois copiar e colar os comandos as-is. Se estiver a usar o ambiente Operações IoT do Azure Codespaces do quickstart, estas variáveis já estão definidas para si e pode saltar este passo. 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 usadas:
| Variável de ambiente |
Descrição |
SUBSCRIPTION_ID |
O ID da subscrição que contém a sua instância Operações IoT do Azure. |
RESOURCE_GROUP |
O nome do grupo de recursos que contém a sua instância do Operações IoT do Azure. |
AIO_INSTANCE_NAME |
O nome da sua instância do Operações IoT do Azure. Para listar as suas instâncias, execute az iot ops list -o table. |
CLUSTER_NAME |
O nome do cluster Kubernetes com Azure Arc que hospeda a sua instância. |
LOCATION |
A região do Azure deve ser usada para novos recursos, por exemplo eastus. |
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>"
Só precisa de definir as variáveis que este artigo utiliza. Este artigo pode usar variáveis adicionais de ambiente para nomes de recursos que 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. Define cada um antes de executares 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. Liga-os na ordem em que queres que os dados fluam.
Na experiência de operações, vá à sua instância Operações IoT do Azure.
Selecionar grafo >Criar grafo de fluxo de dados.
Introduza um nome para o gráfico de fluxo de dados e selecione um perfil de fluxo de dados. O perfil padrão é selecionado automaticamente.
Construa o seu pipeline adicionando elementos à tela:
Adicionar uma fonte: Selecione o endpoint de origem e configure os tópicos a subscrever para as mensagens recebidas.
Adicionar transformações: Selecione uma ou mais transformações para processar os dados. As transformações disponíveis são map, filter, branch, concatenate, window e throttle. Para detalhes sobre cada tipo de transformação, consulte Visão Geral dos Gráficos de Fluxo de Dados.
Adicionar um destino: Selecione o endpoint de destino e configure o tópico ou caminho para onde enviar os dados processados.
Liga os elementos pela ordem em que queres que os dados fluam.
Selecione Guardar para implementar o gráfico de fluxo de dados.
O CLI do Azure utiliza um grafo de fluxo de dados a partir de um único ficheiro de configuração JSON que contém todos os nós e ligações. Use az iot ops dataflowgraph apply para criar ou substituir o gráfico. O exemplo seguinte lê dados de temperatura, converte-os para Fahrenheit e envia-os para um tópico de destino.
Crie um graph.json ficheiro com as propriedades do grafo de fluxo de dados. No graph.json ficheiro, armazene as regras de cada transformação no value campo como uma string JSON escapada. Para a forma legível das regras de cada transformação, veja o tutorial 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"
}
}
]
}
Aplica o ficheiro 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 ficheiro Bicep .bicep com a seguinte estrutura. Este exemplo cria um gráfico de fluxo de dados que lê dados de temperatura, converte-os para Fahrenheit e envia-os 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' }
}
]
}
}
Implementar o ficheiro Bicep:
az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep
Importante
A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.
Crie um ficheiro de manifesto .yaml Kubernetes com a seguinte estrutura. Este exemplo cria um gráfico de fluxo de dados que lê dados de temperatura, converte-os para Fahrenheit e envia-os 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 endpoint e um ou mais tópicos.
No editor de grafos de fluxo de dados, selecione o elemento de origem e configure:
| Configuração |
Descrição |
|
Ponto final |
O ponto final de fluxo de dados a usar. Selecione o padrão para o corretor local do MQTT. |
| Tópicos |
Um ou mais tópicos para subscrever para mensagens recebidas. |
A CLI aplica todo o grafo de uma só vez, por isso configure a origem como um nó Source no ficheiro de configuração graph.json e, em seguida, 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
A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.
- nodeType: Source
name: sensors
sourceSettings:
endpointRef: default
dataSources:
- telemetry/temperature
- telemetry/humidity
Transforma os dados do processo entre a origem e o destino. Cada transformação faz referência a um artefacto incorporado e tem regras de configuração.
As transformações incorporadas disponíveis são:
| Transformar |
Artefacto |
Descrição |
|
Mapa |
azureiotoperations/graph-dataflow-map:1.0.0 |
Renomear, reestruturar, calcular e copiar campos. |
|
Filtro |
azureiotoperations/graph-dataflow-filter:1.0.0 |
Descartar mensagens que correspondam a uma condição. |
|
Sucursal |
azureiotoperations/graph-dataflow-branch:1.0.0 |
Encaminhe cada mensagem para um caminho true ou false com base numa condição. |
|
Concatenate |
azureiotoperations/graph-dataflow-concatenate:1.0.0 |
Una dois ou mais caminhos num único caminho. |
|
Janela |
azureiotoperations/graph-dataflow-window:1.0.0 |
Recolhe mensagens ao longo de um intervalo de tempo e depois agregue. |
|
Throttle |
azureiotoperations/graph-dataflow-throttle:1.0.0 |
Limite a taxa de mensagens por padrão de tópico MQTT. |
Para mais informações sobre como enriquecer mensagens com dados externos, consulte Enriquecer com dados externos.
No editor de grafos de fluxo de dados, selecione Adicionar transformação e escolha o tipo de transformação. Configura as regras no editor visual.
Cada transformação é um nó com nodeType definido como Graph no ficheiro de configuração graph.json. As regras da transformada são um objeto JSON, tal como este mapa que converte a temperatura em Fahrenheit:
{
"map": [
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
A configuration propriedade aceita estas regras como uma sequência, por isso escapa do JSON das regras e coloca-a no value campo. Aplique o gráfico 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)\"}]}"
}
]
}
}
Sugestão
Para gerar a string escapada, guarde as regras num ficheiro como rules.json, execute jq -c . rules.json, e cole a saída de linha única no value campo.
Cada transformada é um nó com nodeType: 'Graph'. A configuration propriedade transmite regras como uma cadeia 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
A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.
Cada transformada é um nó com nodeType: Graph. A configuration propriedade transmite regras como uma cadeia 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)"
}
]
}
Encadeia qualquer número de transformações. Liga-os na nodeConnections secção pela ordem em que queres que os dados fluam:
Arraste as ligações entre transformações na tela para definir a ordem de processamento.
Defina a ordem de processamento na nodeConnections secção do seu graph.json ficheiro de configuração:
"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
A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.
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 vão os dados processados. Especifique uma referência de endpoint e um tópico ou caminho.
Selecione o elemento de destino e configure:
| Configuração |
Descrição |
|
Ponto final |
O ponto final do fluxo de dados para envio. |
|
Topic |
O tópico ou caminho para publicar dados processados. |
Configure o destino como um nó Destination no seu ficheiro 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
A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.
- nodeType: Destination
name: output
destinationSettings:
endpointRef: default
dataDestination: telemetry/processed
Para encaminhamento dinâmico de tópicos com base no conteúdo da mensagem, veja Encaminhar mensagens para diferentes tópicos.
Verifique se o gráfico de fluxo de dados está a funcionar
Depois de implementar um gráfico de fluxo de dados, verifique se está a correr:
Na experiência de operações, selecione o seu gráfico de fluxo de dados para ver o seu estado. Um gráfico saudável mostra um estado em execução.
Use az iot ops dataflowgraph show para visualizar os detalhes de um gráfico:
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 estado 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
A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.
kubectl get dataflowgraph temperature-processing -n azure-iot-operations
Verifique os registos do pod para eventuais erros:
kubectl logs -l app=dataflow -n azure-iot-operations --tail=50
Conteúdo relacionado
