Filtra, ramifica e unisci i dati nei grafici di flusso dati

I grafici del flusso di dati offrono due modi per controllare i messaggi che passano attraverso la pipeline: i filtri eliminano i messaggi indesiderati e i branch trasformano e instradano ogni messaggio verso uno dei due percorsi in base a una condizione. Dopo la diramazione, una trasformazione concatenata unisce nuovamente i percorsi.

Queste trasformano i messaggi di instradamento all'interno del grafo. Per instradare invece i messaggi verso diversi argomenti MQTT in base al loro contenuto, vedi Instradare i messaggi verso diversi argomenti MQTT.

Per una panoramica dei grafici del flusso di dati e della composizione delle trasformazioni in una pipeline, vedere Panoramica dei grafici del flusso di dati.

Le trasformazioni utilizzano un linguaggio di espressione per calcolare valori, condizioni di test e campi di riferimento. Le espressioni si riferiscono agli input per posizione, non per nome: il primo input nella inputs lista è $1, il secondo è $2, e così via. Funzioni integrate come cToF convertono e manipolano tali valori.

Per l'elenco completo di operatori, funzioni, tipi di dati e campi di metadati, consulta il riferimento Expressions.

Prerequisiti

  • Un endpoint del Registro di sistema predefinito denominato default che punta a mcr.microsoft.com viene creato automaticamente durante la distribuzione. Le trasformazioni predefinite usano questo endpoint.

Gli esempi interfaccia della riga di comando di Azure in questo articolo usano variabili di ambiente così puoi impostare ogni valore una volta e poi copiare e incollare i comandi as-is. Se stai usando l'ambiente Operazioni di Azure IoT Codespaces dal quickstart, queste variabili sono già impostate per te e puoi saltare questo passaggio. Altrimenti, imposta le seguenti variabili di ambiente nella tua shell prima di eseguire i comandi.

I seguenti script impostano le variabili di ambiente più comunemente utilizzate:

Variabile di ambiente Descrizione
SUBSCRIPTION_ID L'ID dell'abbonamento che contiene la tua istanza Operazioni di Azure IoT.
RESOURCE_GROUP Il nome del gruppo di risorse che contiene la tua istanza Operazioni di Azure IoT.
AIO_INSTANCE_NAME Il nome della tua istanza Operazioni di Azure IoT. Per elencare le tue istanze, esegui az iot ops list -o table.
CLUSTER_NAME Il nome del cluster Kubernetes abilitato Azure Arc che ospita la tua istanza.
LOCATION La regione Azure da utilizzare per nuove risorse, ad esempio eastus. 
SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

Devi solo impostare le variabili utilizzate in questo articolo. Questo articolo potrebbe utilizzare variabili ambientali aggiuntive per i nomi delle risorse che scegli. L'articolo spiega come posizionarli dove vengono introdotti.

Trasformazione filtro

Una trasformazione filtro valuta ogni messaggio in arrivo rispetto a una o più regole e decide se il messaggio continua attraverso la pipeline o viene eliminato.

Importante

Un'espressione filtro seleziona i messaggi da rimuovere, non quelli da mantenere. Quando l'espressione è vera, il messaggio viene eliminato. Questo comportamento è l'opposto di un'espressione di mapping, che calcola un valore che viene conservato.

Per mantenere i messaggi che corrispondono a una condizione, inverti l'espressione. Ad esempio, per mantenere solo letture sopra 90, filtra su $1 <= 90.

Funzionamento delle regole di filtro

Ogni regola di filtro ha queste proprietà:

Proprietà Obbligatorio Descrizione
inputs Sì Elenco dei percorsi dei campi da leggere dal messaggio in arrivo.
expression Sì Formula applicata ai valori di input. Deve restituire un valore booleano. Quando ritorna vero, il messaggio viene eliminato.
description No Etichetta leggibile usata nei messaggi di errore.

Ogni input si mappa in una variabile posizionale in base al suo ordine: il primo input è $1, il secondo è $2, e così via.

Quando si definiscono più regole, usano la logica OR: se una regola restituisce true, il messaggio viene eliminato. Il motore va in cortocircuito una volta che una regola corrisponde.

Vincoli di chiave:

  • L'espressione è obbligatoria. Ogni regola di filtro deve includere un oggetto expression.
  • filter richiede una matrice. Fornire regole come array JSON, "filter": [ { ... } ], anche per una singola regola. Se si passa un oggetto semplice, non si riesce a caricare la trasformazione e l'errore risultante fa riferimento all'artefatto e al registro invece che al payload delle regole. Questo vincolo differisce da branch, che prende un solo oggetto.
  • Nessun input con caratteri jolly. Ogni input deve fare riferimento a un percorso di campo specifico.
  • I campi mancanti causano errori. Se un campo a cui si fa riferimento in inputs non esiste, il filtro restituisce un errore anziché passare automaticamente il messaggio.
  • I risultati non booleani causano errori. Se un'espressione restituisce un valore non booleano,ad esempio una stringa o un numero, il filtro restituisce un errore.

Eliminare i messaggi in base alla condizione

Per eliminare i messaggi in cui la temperatura supera 100:

Nella configurazione della trasformazione del filtro aggiungere una regola:

Impostazione Valore
Inserimento temperature
Expression $1 > 100

I messaggi in cui la temperatura è di 100 o meno vengono trasmessi. I messaggi superiori a 100 vengono eliminati.

Conserva i messaggi per condizione

Spesso si vuole il risultato opposto: mantenere solo i messaggi che corrispondono a una condizione. Poiché un'espressione filtrante seleziona cosa rimuovere, inverti il confronto.

Per mantenere solo valori sopra 90, abbassa tutto a 90 o sotto:

Nella configurazione della trasformazione del filtro aggiungere una regola:

Impostazione Valore
Inserimento temperature
Expression $1 <= 90
Description Drop readings at or below 90

Solo i messaggi con valori superiori a 90 continuano attraverso la pipeline. Scrivendo $1 > 90 qui si otterrebbe l'opposto di quello che si vorrebbe: farebbe eliminarer ogni lettura sopra i 90 e manterrebbe quelle più fredde.

Suggerimento

Usare il campo description per indicare l'intento della regola in termini di ciò che elimina. Una descrizione come Drop readings at or below 90 rimane accurata, mentre Keep hot readings invita all'errore di espressione invertita e appare nei messaggi di errore che poi vengono letti al contrario.

Usare più condizioni

Quando si definiscono più regole, il filtro elimina il messaggio se una regola corrisponde:

Aggiungere due regole:

Inserimento Expression Descrizione
temperature $1 > 100 Riduzione della temperatura elevata
humidity $1 > 95 Eliminare l'umidità elevata
Message regola della temperatura regola di umidità Result
{"temperature": 150, "humidity": 60} true falso Dropped
{"temperature": 80, "humidity": 98} falso true Dropped
{"temperature": 80, "humidity": 60} falso falso Pass

Suggerimento

Usare più input in una regola quando è necessaria la logica AND tra i campi. Usare più regole quando è necessaria la logica OR in condizioni indipendenti.

Usare espressioni complesse

Fare riferimento a più campi in una singola regola e combinarli con operatori logici:

Aggiungere una regola con gli input temperature e humidity, e l'espressione $1 > 30 && $2 < 60.

Per l'elenco completo di operatori e funzioni, vedere Riferimenti alle espressioni.

Valida i messaggi di filtro contro uno schema

Configura una trasformazione di filtro per convalidare i messaggi in arrivo rispetto a uno schema JSON prima che le regole di filtro vengano eseguite. Il processo rilascia messaggi che non rispettano lo schema.

Per abilitare la convalida dello schema, impostare su validateSchematrue nella configurazione del filtro. Se abilitato, il filtro recupera lo schema da schemaRef nella connessione al nodo in ingresso (il lato from della voce nodeConnections che alimenta il nodo filtro).

La configurazione della trasformazione filtro include una casella di controllo Convalida schema . Tuttavia, l'esperienza operativa attualmente non supporta la configurazione o la visualizzazione delle schemaRef connessioni on-node. Per usare la convalida dello schema, configurare la connessione del nodo schemaRef tramite Bicep o manifesti Kubernetes.

Istruzioni:

  • Usare un solo filtro di convalida per ogni pipeline.
  • Posizionare prima il filtro di convalida in modo che i messaggi non validi vengano eliminati prima di altre elaborazioni.
  • Le regole di filtro vengono comunque applicate dopo il passaggio della convalida dello schema. Se è necessaria solo la convalida dello schema, lasciare vuote le regole di filtro.
  • schemaRef deve puntare a uno schema nel registro degli schemi. serializationFormat Specifica il formato dello schema , ad esempio Json.

Per informazioni sulla configurazione degli schemi, vedere Informazioni sugli schemi dei messaggi.

Arricchire le regole di filtro con dati esterni

Le regole di filtro supportano i set di dati, che consentono di confrontare i valori con i dati di un archivio stati esterno. Per informazioni dettagliate sulla configurazione dei set di dati, vedere Arricchire con dati esterni.

Configurazione completa del filtro

Nella configurazione della trasformazione del filtro aggiungere una o più regole con input ed espressioni booleane. Facoltativamente, abilitare la convalida dello schema e configurare i set di dati per le ricerche di arricchimento.

Chiave Obbligatorio Descrizione
filter Sì Matrice di regole di filtro.
datasets No Matrice di definizioni di set di dati per le ricerche di arricchimento.
validateSchema No Quando true, convalida i messaggi rispetto a uno schema JSON prima dell'esecuzione delle regole di filtro. Il valore predefinito è false.

Trasformazione del ramo

Una trasformazione di ramo valuta una condizione in ogni messaggio in arrivo e la indirizza a uno dei due percorsi di output: true o false. A differenza di un filtro (che elimina i messaggi), un ramo mantiene ogni messaggio e lo indirizza verso il basso nel percorso appropriato.

Funzionamento della diramazione

Ogni messaggio va esattamente a uno dei due percorsi. Non viene eliminato nulla.

Vincoli di chiave:

  • L'espressione di ramo deve restituire un valore booleano. I risultati non booleani causano un errore.
  • Nessun input con caratteri jolly.
  • Esattamente una regola di ramificazione. La branch chiave accetta un singolo oggetto, non una matrice.

Importante

Il ramificamento divide i messaggi in percorsi di elaborazione separati, ma tutti i percorsi devono riunirsi usando una trasformata di concatenazione prima di raggiungere la destinazione. Considerare la diramazione come un modo per applicare trasformazioni diverse a messaggi diversi, non come un modo per instradare verso più endpoint.

Definire una regola di ramo

Per creare un ramo dei messaggi in base a una soglia di gravità:

Nella configurazione della trasformazione del ramo impostare:

Impostazione Valore
Inserimento severity
Expression $1 > 5

I messaggi in cui severity è maggiore di 5 passano al percorso true. Tutti gli altri vanno al percorso false.

Valida i messaggi di branch rispetto a uno schema

A partire dalla versione 1.1.0, puoi configurare una trasformazione di branch per validare i messaggi in arrivo contro uno schema JSON prima di valutare l'espressione di branch.

Per abilitare la validazione dello schema, imposta validateSchema su true nella configurazione del ramo. Il validateSchema campo è opzionale e predefinito su false. Se abilitato, il ramo recupera lo schema da schemaRef nella connessione al nodo in ingresso (il lato from della voce nodeConnections che alimenta il nodo ramo).

  • I messaggi che superano la validazione dello schema continuano la valutazione dei rami.
  • I messaggi che non superano la validazione dello schema vanno al false percorso.

La configurazione della trasformazione del ramo include una casella di spunta Convalida schema. Tuttavia, l'esperienza operativa attualmente non supporta la configurazione o la visualizzazione delle schemaRef connessioni on-node. Per usare la convalida dello schema, configurare la connessione del nodo schemaRef tramite Bicep o manifesti Kubernetes.

Connettere gli output dei rami

Nella configurazione della pipeline usare il nome del nodo seguito da .output.true o .output.false per collegare ogni percorso a una trasformazione downstream.

Nell'editor del grafico del flusso di dati, trascinare le connessioni dagli output vero e falso della trasformazione di ramo alle trasformazioni downstream adatte.

Unire i percorsi tramite concatenazione

Tutti i percorsi dei rami devono convergere prima di raggiungere una destinazione. Una trasformazione concatenata li unisce. Non ha alcuna configurazione e nessuna regola. I messaggi provenienti da tutti gli input connessi passano senza modifiche.

Aggiungere una trasformazione di concatenazione a canvas e connettere entrambi i rami, quindi collegare il modulo di concatenazione alla destinazione.

Esempio: filtrare, diramare e unire

Questo esempio end-to-end filtra letture non valide, si dirama in base alla gravità, applica diverse trasformazioni della mappa a ogni percorso e unisce i risultati.

Screenshot del canvas dell'esperienza operativa che mostra un filtro, un ramo, una mappa, una concat e una pipeline di destinazione.

Per costruire questa pipeline nell'esperienza operativa:

  1. Creare un grafo del flusso di dati e aggiungere un'origine che legge da telemetry/sensors.
  2. Aggiungere una trasformazione di filtro . Configurare una regola che elimina i messaggi in cui temperature > 1000.
  3. Aggiungere una trasformazione di ramo . Configurare la condizione severity > 5 per instradare i messaggi con gravità elevata al percorso vero.
  4. Aggiungere una trasformazione mappa sul percorso vero. Configurare le regole per rinominare deviceId in id, temperature in temp, e aggiungere un campo alert impostato su true.
  5. Aggiungere una trasformazione mappa sul percorso false. Configurare le regole per rinominare deviceId in id e temperature in temp.
  6. Aggiungere una trasformazione concatenata per unire entrambi i percorsi.
  7. Aggiungere una destinazione che invia a telemetry/processed.
  8. Collega gli elementi: origine → filtro → diramazione → (percorso vero: mappa degli avvisi, percorso falso: mappa normale) → concatenazione → destinazione.

Contenuti correlati

  • Last updated on