Filtrare e instradare i dati nei grafici del flusso di dati

Importante

Questa pagina include istruzioni per la gestione dei componenti Azure IoT Operations usando i manifesti di distribuzione kubernetes, disponibile in ANTEPRIMA. Questa funzionalità viene fornita con diverse limitazioni e non deve essere usata per i carichi di lavoro di produzione.

Consultare i Termini supplementari di utilizzo per le anteprime di Microsoft Azure per i termini legali applicabili alle funzionalità di Azure in versione beta, anteprima o non ancora rilasciate in versione generale.

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 concat unisce nuovamente i percorsi.

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.

Prerequisiti

Istanza di Azure IoT Operations distribuita in un cluster Kubernetes abilitato per Arc. Per altre informazioni, vedere Deploy Azure IoT Operations.
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.

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.

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.
`description`	No	Etichetta leggibile usata nei messaggi di errore.

Gli input vengono assegnate variabili posizionali in base all'ordine: il primo input è $1, il secondo è $2e 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.
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`

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 > 100'
  }
]

- inputs:
    - temperature     # $1
  expression: "$1 > 100"

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

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

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 > 100'
    description: 'Drop high temperature'
  }
  {
    inputs: [ 'humidity' ]
    expression: '$1 > 95'
    description: 'Drop high humidity'
  }
]

- inputs:
    - temperature     # $1
  expression: "$1 > 100"
  description: "Drop high temperature"

- inputs:
    - humidity     # $1
  expression: "$1 > 95"
  description: "Drop high humidity"

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.

filter: [
  {
    inputs: [ 'temperature', 'humidity' ]
    expression: '$1 > 30 && $2 < 60'
    description: 'Drop hot and dry readings'
  }
]

- inputs:
    - temperature     # $1
    - humidity        # $2
  expression: "$1 > 30 && $2 < 60"
  description: "Drop hot and dry readings"

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

Convalidare i messaggi in base a uno schema

È possibile configurare una trasformazione di filtro per convalidare i messaggi in ingresso in base a uno schema JSON prima dell'esecuzione delle regole di filtro. I messaggi che non sono conformi allo schema vengono eliminati immediatamente.

Per abilitare la convalida dello schema, impostare su validateSchematrue nella configurazione del filtro. Se abilitato, il filtro recupera lo schema dall'impostazione dell'origine del flusso di schemaRef dati.

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.

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.

Il codice JSON delle regole di filtro viene passato come value per la chiave rules.

configuration: [
  {
    key: 'rules'
    value: '{"datasets":[{"key":"device_limits as limits","inputs":["$source.deviceId","$context.deviceId"],"expression":"$1 == $2"}],"filter":[{"inputs":["temperature"],"expression":"$1 > 100","description":"Drop high temperature readings"},{"inputs":["rawValue","$context(limits).maxValue"],"expression":"$1 > $2","description":"Drop readings above device-specific limit"}]}'
  }
]

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": ["temperature"],
      "expression": "$1 > 100",
      "description": "Drop high temperature readings"
    },
    {
      "inputs": ["rawValue", "$context(limits).maxValue"],
      "expression": "$1 > $2",
      "description": "Drop readings above device-specific limit"
    }
  ]
}

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 (a differenza del filtro, che genera errori anche per i risultati non booleani).
Nessun input con caratteri jolly.
Esattamente una regola di ramificazione. La branch chiave accetta un singolo oggetto, non una matrice.

Importante

La diramazione suddivide i messaggi in percorsi di elaborazione separati, ma tutti i percorsi devono essere uniti usando una trasformazione concat 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`

configuration: [
  {
    key: 'rules'
    value: '{"branch":{"inputs":["severity"],"expression":"$1 > 5","description":"Route high-severity messages"}}'
  }
]

{
  "branch": {
    "inputs": ["severity"],
    "expression": "$1 > 5",
    "description": "Route high-severity messages"
  }
}

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

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.

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'severity-check' } }
  { from: { name: 'severity-check.output.true' }, to: { name: 'alert-transform' } }
  { from: { name: 'severity-check.output.false' }, to: { name: 'normal-transform' } }
]

nodeConnections:
  - from: { name: sensors }
    to: { name: severity-check }
  - from: { name: severity-check.output.true }
    to: { name: alert-transform }
  - from: { name: severity-check.output.false }
    to: { name: normal-transform }

Unire percorsi con la funzione "concat"

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

Aggiungere una trasformazione concat al canvas e connettere entrambi i percorsi dei rami, quindi connettere il concat alla destinazione.

{
  nodeType: 'Graph'
  name: 'merge'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
  }
}

- nodeType: Graph
  name: merge
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-concatenate:1.0.0

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.

Per compilare questa pipeline nell'esperienza Operazioni:

Creare un grafo del flusso di dati e aggiungere un'origine che legge da telemetry/sensors.
Aggiungere una trasformazione di filtro . Configurare una regola che elimina i messaggi in cui temperature > 1000.
Aggiungere una trasformazione di ramo . Configurare la condizione severity > 5 per instradare i messaggi con gravità elevata al percorso vero.
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.
Aggiungere una trasformazione mappa sul percorso false. Configurare le regole per rinominare deviceId in id e temperature in temp.
Aggiungere una trasformazione concat per unire entrambi i percorsi.
Aggiungere una destinazione che invia a telemetry/processed.
Connettere gli elementi: sorgente → filtro → ramo → (percorso vero: mappa degli avvisi, percorso falso: mappa normale) → concat → destinazione.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  name: 'alert-routing'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/sensors' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'remove-bad-data'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-filter:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"filter":[{"inputs":["temperature"],"expression":"$1 > 1000","description":"Drop impossible temperature readings"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'severity-check'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-branch:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"branch":{"inputs":["severity"],"expression":"$1 > 5","description":"Route high-severity messages"}}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'alert-transform'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["deviceId"],"output":"id"},{"inputs":["temperature"],"output":"temp"},{"inputs":[],"output":"alert","expression":"true"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'normal-transform'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["deviceId"],"output":"id"},{"inputs":["temperature"],"output":"temp"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'merge'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/processed'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
      { from: { name: 'remove-bad-data' }, to: { name: 'severity-check' } }
      { from: { name: 'severity-check.output.true' }, to: { name: 'alert-transform' } }
      { from: { name: 'severity-check.output.false' }, to: { name: 'normal-transform' } }
      { from: { name: 'alert-transform' }, to: { name: 'merge' } }
      { from: { name: 'normal-transform' }, to: { name: 'merge' } }
      { from: { name: 'merge' }, to: { name: 'output' } }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: alert-routing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/sensors

    - nodeType: Graph
      name: remove-bad-data
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-filter:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "filter": [
                  {
                    "inputs": ["temperature"],
                    "expression": "$1 > 1000",
                    "description": "Drop impossible temperature readings"
                  }
                ]
              }

    - nodeType: Graph
      name: severity-check
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-branch:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "branch": {
                  "inputs": ["severity"],
                  "expression": "$1 > 5",
                  "description": "Route high-severity messages"
                }
              }

    - nodeType: Graph
      name: alert-transform
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["deviceId"], "output": "id" },
                  { "inputs": ["temperature"], "output": "temp" },
                  { "inputs": [], "output": "alert", "expression": "true" }
                ]
              }

    - nodeType: Graph
      name: normal-transform
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["deviceId"], "output": "id" },
                  { "inputs": ["temperature"], "output": "temp" }
                ]
              }

    - nodeType: Graph
      name: merge
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-concatenate:1.0.0

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/processed

  nodeConnections:
    - from: { name: sensors }
      to: { name: remove-bad-data }
    - from: { name: remove-bad-data }
      to: { name: severity-check }
    - from: { name: severity-check.output.true }
      to: { name: alert-transform }
    - from: { name: severity-check.output.false }
      to: { name: normal-transform }
    - from: { name: alert-transform }
      to: { name: merge }
    - from: { name: normal-transform }
      to: { name: merge }
    - from: { name: merge }
      to: { name: output }

Passaggi successivi

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-03-31