Filtrar, desviar e fundir dados em grafos de fluxo de dados

Os grafos de fluxo de dados proporcionam duas formas de controlar quais mensagens fluem através do seu pipeline: as transformações de filtro eliminam mensagens indesejadas, e as transformações de ramificação encaminham cada mensagem por um de dois caminhos com base em uma condição. Após a ramificação, uma transformação de concatenação junta novamente os caminhos.

Estas transformam as mensagens de encaminhamento dentro do grafo. Para em vez disso, encaminhar mensagens para diferentes tópicos MQTT com base no seu conteúdo, veja Encaminhar mensagens para diferentes tópicos MQTT.

Para uma visão geral dos gráficos de fluxo de dados e de como as transformações se compõem num pipeline, consulte Visão Geral dos Gráficos de Fluxo de Dados.

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

  • Um endpoint de registo predefinido chamado default que aponta mcr.microsoft.com é criado automaticamente durante a implementação. As transformações incorporadas usam este endpoint.

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>

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.

Transformação de filtro

Uma transformação de filtro avalia cada mensagem recebida de acordo com uma ou mais regras e decide se a mensagem continua pelo pipeline ou se é eliminada.

Importante

Uma expressão de filtro seleciona as mensagens a remover, não as a manter. Quando a expressão é verdadeira, a mensagem é omitida. Este comportamento é o oposto de uma expressão de mapa, que calcula um valor que é mantido.

Para manter as mensagens que correspondem a uma condição, inverta a expressão. Por exemplo, para manter apenas leituras acima de 90, filtre em $1 <= 90.

Como funcionam as regras de filtro

Cada regra de filtro tem estas propriedades:

Propriedade Obrigatório Descrição
inputs Sim Lista de caminhos de campo a ler da mensagem recebida.
expression Sim Fórmula aplicada aos valores de entrada. Tem de devolver um booleano. Quando isso retorna verdadeiro, a mensagem é descartada.
description No Etiqueta legível por humanos usada em mensagens de erro.

Cada entrada corresponde a uma variável posicional com base na sua ordem: a primeira entrada é $1, a segunda é $2, e assim sucessivamente.

Quando defines múltiplas regras, elas usam lógica OR: se alguma regra for avaliada como verdadeira, a mensagem é descartada. O motor entra em curto-circuito assim que uma regra corresponde.

Restrições principais:

  • A expressão é necessária. Cada regra de filtro deve incluir um expression.
  • filter Aceita uma matriz. Fornece regras como um array JSON, "filter": [ { ... } ], mesmo para uma única regra. Passar um objeto simples não consegue carregar a transformação, e o erro resultante aponta para o artefacto e o repositório, em vez do payload de regras. Esta restrição difere de branch, que toma um único objeto.
  • Sem entradas imprevisíveis. Cada entrada deve referenciar um caminho de campo específico.
  • Campos em falta causam erros. Se um campo referenciado em inputs não existir, o filtro devolve um erro em vez de passar a mensagem silenciosamente.
  • Resultados não booleanos causam erros. Se uma expressão devolver um valor não booleano (como uma cadeia ou número), o filtro devolve um erro.

Descartar mensagens por condição

Para descartar mensagens se a temperatura ultrapassar 100:

Na configuração de transformação de filtro, adicione uma regra:

Configuração Value
Input temperature
Expressão $1 > 100

Mensagens em que a temperatura é 100 graus ou menos passam. Mensagens acima de 100 são eliminadas.

Manter mensagens por condição

Muitas vezes, queres o resultado oposto: manter apenas as mensagens que correspondem a uma condição. Como uma expressão de filtro seleciona o que remover, inverta a comparação.

Para manter apenas leituras acima de 90, reduza tudo a 90 ou abaixo:

Na configuração de transformação de filtro, adicione uma regra:

Configuração Value
Input temperature
Expressão $1 <= 90
Description Drop readings at or below 90

Apenas mensagens com valores superiores a 90 continuam através do pipeline. Escrever $1 > 90 aqui faria o oposto do que queres: descartaria todas as leituras acima de 90 e manteria as mais baixas.

Sugestão

Use o campo description para registar a intenção da regra relativamente ao que esta descarta. Uma descrição como Drop readings at or below 90 mantém-se precisa, enquanto Keep hot readings convida ao erro de expressão invertida e aparece em mensagens de erro que depois se lêem ao contrário.

Usar múltiplas condições

Quando definires mais do que uma regra, o filtro elimina a mensagem se qualquer regra for correspondida.

Adicione duas regras:

Entrada Expression Descrição
temperature $1 > 100 Queda de temperatura elevada
humidity $1 > 95 Redução da humidade elevada
Mensagem Regra da temperatura Regra da humidade Result
{"temperature": 150, "humidity": 60} verdadeiro falso Caiu
{"temperature": 80, "humidity": 98} falso verdadeiro Caiu
{"temperature": 80, "humidity": 60} falso falso Passes

Sugestão

Usa múltiplas entradas numa regra quando precisares de lógica AND entre campos. Use múltiplas regras quando precisar de lógica OU em condições independentes.

Usar expressões complexas

Referenciar múltiplos campos numa única regra e combiná-los com operadores lógicos:

Adicione uma regra com entradas temperature e humidity, e expressão $1 > 30 && $2 < 60.

Para a lista completa de operadores e funções, veja Referência de expressões.

Validar mensagens de filtro contra um esquema

Configure uma transformação de filtro para validar mensagens recebidas contra um esquema JSON antes da execução das regras de filtro. O processo deixa de enviar mensagens que não seguem o esquema.

Para permitir a validação do esquema, defina validateSchema como true na configuração do filtro. Quando ativado, o filtro recupera o esquema da schemaRef ligação ao nó de entrada (o from lado da nodeConnections entrada que alimenta o nó do filtro).

A configuração da transformação do filtro inclui uma caixa de verificação Validar esquema. No entanto, a experiência operacional atualmente não suporta configurar ou visualizar as schemaRef ligações on-node. Para usar a validação de esquema, configure o schemaRef da ligação ao nó através de manifestos Bicep ou Kubernetes.

Orientações:

  • Use apenas um filtro de validação por pipeline.
  • Coloque primeiro o filtro de validação para que mensagens inválidas sejam descartadas antes de outros processamentos.
  • As regras de filtro continuam a aplicar-se após a validação do esquema ser aprovada. Se só precisares de validação do esquema, deixa as regras de filtro vazias.
  • O schemaRef deve apontar para um esquema no registo de esquemas. O serializationFormat especifica o formato do esquema (por exemplo, Json).

Para aprender sobre a configuração de esquemas, veja Compreender esquemas de mensagens.

Enriquecer regras de filtro com dados externos

As regras de filtro suportam conjuntos de dados, que permitem comparar valores com dados de um armazenamento de estados externo. Para detalhes sobre a configuração de conjuntos de dados, consulte Enriquecer com dados externos.

Configuração completa do filtro

Na configuração de transformação de filtro, adiciona-se uma ou mais regras com entradas e expressões booleanas. Opcionalmente, ative a validação de esquemas e configure conjuntos de dados para consultas de enriquecimento.

Key Obrigatório Descrição
filter Sim Conjunto de regras de filtro.
datasets No Array de definições de conjuntos de dados para consultas de enriquecimento.
validateSchema No Quando true, valida mensagens contra um esquema JSON antes da execução das regras de filtro. O padrão é false.

Transformada de ramificação

Uma transformada de desvio avalia uma condição em cada mensagem de entrada e encaminha-a para um de dois caminhos de saída: true ou false. Ao contrário de um filtro (que elimina mensagens), um ramo preserva todas as mensagens e encaminha-as pelo caminho apropriado.

Como funciona a ramificação

Cada mensagem vai exatamente para um dos dois caminhos. Nada é deixado cair.

Restrições principais:

  • A expressão do ramo deve devolver um booleano. Resultados não booleanos causam um erro.
  • Sem entradas imprevisíveis.
  • Exatamente uma regra de ramo. A branch chave leva um único objeto, não um array.

Importante

A ramificação divide as mensagens em caminhos de processamento separados, mas todos os caminhos devem fundir-se novamente através de uma transformação de concatenação antes de chegarem ao destino. Pensa na ramificação como uma forma de aplicar diferentes transformações a diferentes mensagens, não como uma forma de encaminhar para múltiplos endpoints.

Defina uma regra de ramo

Para ramificar mensagens com base num limiar de gravidade:

Na configuração da transformada de ramo, defina:

Configuração Value
Input severity
Expressão $1 > 5

Mensagens onde severity é maior que 5 vão para o true caminho. Todos os outros seguem o false caminho.

Validar mensagens de desvio contra um esquema

A partir da versão 1.1.0, pode configurar uma transformação de ramificação para validar as mensagens recebidas em relação a um esquema JSON antes de avaliar a expressão de ramificação.

Para permitir a validação do esquema, defina validateSchema para true na configuração do ramo. O validateSchema campo é opcional e por defeito é false. Quando ativado, o ramo recupera o esquema da schemaRef ligação do nó de entrada (o from lado da nodeConnections entrada que alimenta o nó do ramo).

  • As mensagens que passam a validação do esquema prosseguem para a avaliação do ramo.
  • As mensagens que falham na validação do esquema vão para o false caminho.

A configuração da transformação de ramificação inclui uma caixa de verificação Validar esquema. No entanto, a experiência operacional atualmente não suporta configurar ou visualizar as schemaRef ligações on-node. Para usar a validação de esquema, configure o schemaRef da ligação ao nó através de manifestos Bicep ou Kubernetes.

Ligar saídas de ramificações

Na configuração do pipeline, use o nome do nó seguido de .output.true ou .output.false para ligar cada caminho a uma transformação a jusante.

No editor de grafos de fluxo de dados, arraste as ligações das saídas verdadeira e falsa da transformada de desvio para as transformadas a jusante apropriadas.

Fundir caminhos com concatenação

Todos os caminhos de ramificação devem convergir antes de chegar a um destino. Uma transformação de concatenação funde-os. Não tem configuração nem regras. As mensagens de todas as entradas ligadas passam sem modificações.

Adicione uma transformação de concatenação ao painel e ligue ambos os caminhos das ramificações a ela, depois ligue a concatenação ao destino.

Exemplo: filtrar, ramificar e fundir

Este exemplo completo filtra leituras erróneas, faz a ramificação segundo a severidade, aplica diferentes transformações de mapa a cada caminho e combina os resultados.

Captura de ecrã da tela de experiência de operações mostrando um filtro, ramificação, mapa, concat e pipeline de destino.

Para construir este pipeline na experiência operacional:

  1. Crie um grafo de fluxo de dados e adicione uma fonte que leia de telemetry/sensors.
  2. Adiciona uma transformação de filtro. Configure uma regra que elimine mensagens onde temperature > 1000.
  3. Adicione uma transformação de ramificação . Configure a condição severity > 5 para encaminhar mensagens de alta gravidade para o caminho verdadeiro.
  4. Adiciona uma transformação de mapa no caminho verdadeiro. Configure regras para renomear deviceId para id, temperature para temp, e adicione um campo alert definido para true.
  5. Adicione uma transformação de mapa no trajeto falso. Configure regras para renomear deviceId para id e temperature para temp.
  6. Adicione uma transformação de concatenação para unir os dois caminhos.
  7. Adicione um destino que envie para telemetry/processed.
  8. Ligue os elementos: fonte → filtro → ramificação → (caminho verdadeiro: mapa de alerta, caminho falso: mapa normal) → concatenar → destino.

Conteúdo relacionado

  • Last updated on