Filtrar, desviar y fusionar datos en grafos de flujo de datos

Los gráficos de flujo de datos proporcionan dos maneras de controlar qué mensajes fluyen a través de la canalización: las transformaciones de filtrado eliminan los mensajes no deseados, y las transformaciones de rama enrutan cada mensaje por una de las dos rutas en función de una condición. Tras la bifurcación, una transformación concatenar vuelve a unir las rutas.

Estas transforman los mensajes de enrutamiento dentro del grafo. Para enrutar mensajes a diferentes temas de MQTT según su contenido, véase Enrutar mensajes a diferentes temas de MQTT.

Para obtener información general sobre los gráficos de flujo de datos y cómo las transformaciones se componen en una canalización, consulte Introducción a los gráficos de flujo de datos.

Las transformadas utilizan un lenguaje de expresiones para calcular valores, condiciones de prueba y campos de referencia. Las expresiones se refieren a las entradas por posición, no por nombre: la primera entrada en la inputs lista es $1, la segunda es $2, y así sucesivamente. Las funciones integradas, como cToF, convierten y manipulan esos valores.

Para la lista completa de operadores, funciones, tipos de datos y campos de metadatos, consulte la referencia Expressions.

Prerrequisitos

  • Un punto de conexión del Registro predeterminado denominado default que apunta a mcr.microsoft.com se crea automáticamente durante la implementación. Las transformaciones integradas usan este endpoint.

Los CLI de Azure ejemplos de este artículo usan variables de entorno para que puedas establecer cada valor una vez y luego copiar y pegar los comandos as-is. Si usas el entorno Operaciones de IoT de Azure Codespaces del quickstart, estas variables ya están configuradas para ti y puedes saltarte este paso. De lo contrario, configura las siguientes variables de entorno en tu shell antes de ejecutar los comandos.

Los siguientes scripts establecen las variables de entorno más usadas:

Variable del entorno Descripción
SUBSCRIPTION_ID El ID de la suscripción que contiene tu instancia de Operaciones de IoT de Azure.
RESOURCE_GROUP El nombre del grupo de recursos que contiene tu instancia de Operaciones de IoT de Azure.
AIO_INSTANCE_NAME El nombre de tu instancia de Operaciones de IoT de Azure. Para listar tus instancias, ejecuta az iot ops list -o table.
CLUSTER_NAME El nombre del clúster Kubernetes habilitado para Azure Arc que aloja tu instancia.
LOCATION La región de Azure para usar para nuevos recursos, por ejemplo eastus. 
SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

Solo necesitas establecer las variables que utiliza este artículo. Este artículo podría utilizar variables de entorno adicionales para los nombres de recursos que elijas. El artículo explica cómo situarlos donde se introducen.

Transformación de filtro

Una transformación de filtro evalúa cada mensaje entrante con una o varias reglas y decide si el mensaje continúa a través de la tubería o se descarta.

Importante

Una expresión de filtro selecciona los mensajes que se deben eliminar, no los que se deben conservar. Cuando la expresión es verdadera, el mensaje se omite. Este comportamiento es lo opuesto a una expresión de mapa, que calcula un valor que se mantiene.

Para conservar los mensajes que coinciden con una condición, invierte la expresión. Por ejemplo, para mantener solo lecturas por encima de 90, filtra en $1 <= 90.

Funcionamiento de las reglas de filtro

Cada regla de filtro tiene estas propiedades:

Propiedad Obligatorio Descripción
inputs Sí Lista de rutas de campo para leer del mensaje entrante.
expression Sí Fórmula aplicada a los valores de entrada. Debe devolver un valor booleano. Cuando vuelve a ser verdadero, el mensaje se omite.
description No Etiqueta legible que se usa en los mensajes de error.

Cada entrada se asigna a una variable posicional según su orden: la primera entrada es $1, la segunda es $2, y así sucesivamente.

Al definir varias reglas, usan lógica OR: si alguna regla se evalúa como true, se quita el mensaje. El motor se cortocircuita cuando una regla coincide.

Restricciones de clave:

  • Se requiere expresión. Cada regla de filtro debe incluir un expression.
  • filter acepta una matriz. Proporciona reglas como un array JSON, "filter": [ { ... } ], incluso para una sola regla. Pasar un objeto simple hace que no se cargue la transformación, y el error resultante apunta al artefacto y al registro en lugar de al contenido de las reglas. Esta restricción difiere de branch, que toma un solo objeto.
  • No hay entradas de carácter comodín. Cada entrada debe hacer referencia a una ruta de acceso de campo específica.
  • Los campos que faltan provocan errores. Si no existe un campo al que se hace referencia en inputs , el filtro devuelve un error en lugar de pasar el mensaje de forma silenciosa.
  • Los resultados no booleanos provocan errores. Si una expresión devuelve un valor no booleano (como una cadena o un número), el filtro devuelve un error.

Quitar mensajes por condición

Para descartar mensajes en los que la temperatura exceda los 100:

En la configuración de transformación de filtro, agregue una regla:

Configuración Importancia
Input temperature
Expresión $1 > 100

Los mensajes pasan si la temperatura es de 100 o inferior. Los mensajes que superan el número 100 se descartan.

Conservar mensajes por condición

A menudo, se quiere el resultado contrario: conservar solo los mensajes que coincidan con una condición. Como una expresión de filtro selecciona qué eliminar, invierte la comparación.

Para mantener solo las lecturas por encima de 90, descarta todo lo que esté en 90 o por debajo:

En la configuración de transformación de filtro, agregue una regla:

Configuración Importancia
Input temperature
Expresión $1 <= 90
Descripción Drop readings at or below 90

Solo los mensajes con valores superiores a 90 continúan a través de la canalización. Escribir $1 > 90 aquí haría justo lo contrario de lo que quieres hacer: descartaría todos los valores por encima de 90 y conservaría los más bajos.

Sugerencia

Use el campo description para registrar el propósito de la regla en función de lo que descarta. Una descripción como Drop readings at or below 90 sigue siendo precisa, mientras que Keep hot readings propicia el error de inversión de la expresión y da lugar a mensajes de error que luego se leen al revés.

Uso de varias condiciones

Al definir más de una regla, el filtro quita el mensaje si alguna regla coincide:

Agregue dos reglas:

Entrada Expression Descripción
temperature $1 > 100 Reducción de alta temperatura
humidity $1 > 95 Reducir la alta humedad
Mensaje regla de temperatura regla de humedad Resultado
{"temperature": 150, "humidity": 60} verdadero false Dropped
{"temperature": 80, "humidity": 98} false verdadero Dropped
{"temperature": 80, "humidity": 60} false false Pases

Sugerencia

Use varias entradas en una regla cuando necesite lógica AND entre campos. Use varias reglas cuando necesite lógica OR en condiciones independientes.

Uso de expresiones complejas

Haga referencia a varios campos en una sola regla y combínelos con operadores lógicos:

Agregue una regla con entradas temperature y humidity, y expresión $1 > 30 && $2 < 60.

Para obtener la lista completa de operadores y funciones, vea Referencia de expresiones.

Validar mensajes de filtro frente a un esquema

Configura una transformación de filtro para validar los mensajes entrantes contra un esquema JSON antes de ejecutar las reglas de filtro. El proceso suelta mensajes que no se ajustan al esquema.

Para habilitar la validación de esquemas, establezca validateSchema en true en la configuración del filtro. Cuando está habilitado, el filtro recupera el esquema de schemaRef en la conexión del nodo entrante (el lado from de la entrada nodeConnections que se alimenta en el nodo de filtro).

La configuración de transformación de filtro incluye una casilla Validar esquema . Sin embargo, la experiencia de operaciones actualmente no permite configurar ni ver el schemaRef en las conexiones de nodo. Para usar la validación de esquemas, configure la conexión del nodo mediante schemaRef a través de Bicep o manifiestos de Kubernetes.

Directrices:

  • Use solo un filtro de validación por canalización.
  • Coloque primero el filtro de validación para que los mensajes no válidos se quiten antes de otro procesamiento.
  • Las reglas de filtro se siguen aplicando después de que se supere la validación del esquema. Si solo necesita la validación del esquema, deje vacías las reglas de filtro.
  • schemaRef debe apuntar a un esquema en el registro de esquemas. serializationFormat especifica el formato de esquema (por ejemplo, Json).

Para obtener información sobre cómo configurar esquemas, consulte Descripción de los esquemas de mensajes.

Enriquecimiento de reglas de filtro con datos externos

Las reglas de filtro admiten conjuntos de datos, que permiten comparar valores con datos de un almacén de estado externo. Para más información sobre cómo configurar conjuntos de datos, consulte Enriquecimiento con datos externos.

Configuración de filtro completa

En la configuración de transformación de filtro, agregue una o varias reglas con entradas y expresiones booleanas. Opcionalmente, habilite la validación del esquema y configure conjuntos de datos para búsquedas de enriquecimiento.

Key Obligatorio Descripción
filter Sí Matriz de reglas de filtro.
datasets No Matriz de definiciones de conjunto de datos para búsquedas de enriquecimiento.
validateSchema No Cuando true, valida los mensajes en un esquema JSON antes de que se ejecuten las reglas de filtro. Tiene como valor predeterminado false.

Transformación de rama

Una transformación de rama evalúa una condición en cada mensaje entrante y la enruta a una de las dos rutas de acceso de salida: true o false. A diferencia de un filtro (que elimina mensajes), una rama conserva todos los mensajes y los dirige por la ruta adecuada.

Funcionamiento de la bifurcación

Cada mensaje va a exactamente una de las dos rutas de acceso. Nada se pierde.

Restricciones de clave:

  • La expresión de rama debe devolver un valor booleano. Los resultados no booleanos causan un error.
  • No hay entradas comodín.
  • Exactamente una regla de rama. La branch clave toma un único objeto, no una matriz.

Importante

El ramificado divide los mensajes en rutas de procesamiento separadas, pero todas deben volver a fusionarse mediante una transformación de concatenación antes de llegar al destino. Piense en la bifurcación como una manera de aplicar diferentes transformaciones a distintos mensajes, no como una manera de enrutar a varios puntos de conexión.

Definición de una regla de rama

Para bifurcar mensajes en función de un umbral de gravedad:

En la configuración de transformación de rama, establezca:

Configuración Importancia
Input severity
Expresión $1 > 5

Los mensajes en los que severity es mayor que 5 van a la ruta true. El resto va a la ruta de acceso false.

Validar mensajes de desviación contra un esquema

A partir de la versión 1.1.0, puedes configurar una transformación de rama para validar los mensajes entrantes con respecto a un esquema JSON antes de evaluar la expresión de rama.

Para habilitar la validación del esquema, establece validateSchema en true en la configuración de la rama. El validateSchema campo es opcional y por defecto es false. Cuando está habilitado, la rama recupera el esquema de schemaRef en la conexión del nodo entrante (el lado from de la entrada nodeConnections que se alimenta en el nodo de rama).

  • Los mensajes que pasan la validación del esquema pasan a la evaluación de ramas.
  • Los mensajes que fallan en la validación del esquema van al false camino.

La configuración de la transformación de rama incluye una casilla de verificación Validar esquema. Sin embargo, la experiencia de operaciones actualmente no permite configurar ni ver el schemaRef en las conexiones de nodo. Para usar la validación de esquemas, configure la conexión del nodo mediante schemaRef a través de Bicep o manifiestos de Kubernetes.

Conectar salidas de rama

En la configuración de canalización, use el nombre del nodo seguido de .output.true o .output.false para conectar cada ruta a una transformación descendente.

En el editor de gráficos de flujo de datos, arrastre las conexiones de las salidas True y False de la transformación de rama a las transformaciones de bajada adecuadas.

Combinar rutas con concatenar

Todas las rutas de rama deben converger antes de llegar a un destino. Una transformación de concatenación las fusiona. No tiene ninguna configuración y ninguna regla. Los mensajes de todas las entradas conectadas pasan sin modificación.

Añada una transformación de concatenación al lienzo y conecte ambas rutas de ramificación a ella; a continuación, conecte la concatenación al destino.

Ejemplo: filtrar, ramificar y fusionar

En este ejemplo completo se filtran las lecturas incorrectas, se ramifican según la gravedad, se aplican transformaciones de mapa distintas a cada ruta y se combinan los resultados.

Captura de pantalla del lienzo de operaciones que muestra un filtro, una rama, un mapa, una concatenación y una canalización de destino.

Para construir esta línea de cartas en la experiencia operativa:

  1. Cree un gráfico de flujo de datos y agregue un origen que lea de telemetry/sensors.
  2. Agregue una transformación de filtro . Configure una regla que descarte los mensajes donde temperature > 1000.
  3. Agregue un branch transform. Configure la condición severity > 5 para enrutar los mensajes de alta severidad al camino verdadero.
  4. Agregue una transformación de map en la ruta correcta. Configure las reglas para renombrar deviceId a id, temperature a temp, y agregar un campo alert establecido en true.
  5. Agregue una transformación de map en la ruta falsa. Configure reglas para cambiar el nombre deviceId a id y temperature a temp.
  6. Agregue una transformación concatenada para combinar ambas rutas de acceso.
  7. Agregue un destino que envíe a telemetry/processed.
  8. Conecte los elementos: fuente → filtro → rama → (ruta de acceso verdadera: mapa de alertas, ruta de acceso falsa: mapa normal) → concatenar → destino.

Contenido relacionado

  • Last updated on