Aggregare i dati nel tempo 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.

Una trasformazione finestra raccoglie i messaggi in un intervallo di tempo fisso e produce un singolo messaggio di output con valori aggregati. Invece di inoltrare ogni lettura singolarmente, è possibile calcolare statistiche come medie, minimi o conteggi e inviare un risultato consolidato downstream.

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.

Quando usare una trasformazione di finestra

Usare una trasformazione a finestra quando si ricevono dati del sensore ad alta frequenza e si vuole ridurre il volume prima di inviarli a valle. Gli scenari comuni includono:

Medie di calcolo: un sensore di temperatura pubblica ogni secondo, ma l'applicazione cloud richiede solo una media di 30 secondi.
Tenere traccia degli estremi: si desidera che vengano rilevate le letture di pressione minima e massima in ogni intervallo di un minuto.
Conteggio eventi: è necessario conoscere il numero di eventi di apertura delle porte negli ultimi cinque minuti.

Funzionamento della trasformazione della finestra

La trasformazione della finestra ha due passaggi interni connessi in sequenza:

Ritardo: imposta la durata della finestra scorrevole. I messaggi in arrivo vengono assegnati a un limite di finestra in base al timestamp.
Accumula: applica le regole di aggregazione alla chiusura della finestra. Tutti i messaggi nella finestra vengono ridotti a un singolo messaggio di output.

Annotazioni

Il passaggio di ritardo allinea i timestamp dei messaggi ai limiti della finestra. Se un messaggio arriva a 7 secondi dall'inizio di una finestra di 10 secondi, viene assegnato al limite temporale di 10 secondi.

Configurare la durata della finestra

Il passaggio di ritardo controlla per quanto tempo dura ogni finestra a cascata. La chiave di configurazione è delay (non rules).

Nella configurazione della trasformazione della finestra impostare la durata della finestra in secondi. Ad esempio, impostarlo su 30 per una finestra scorrevole di 30 secondi.

configuration: [
  {
    key: 'delay'
    value: '{"type":"duration","delaySeconds":30}'
  }
]

{
  "type": "duration",
  "delaySeconds": 30
}

Proprietà	Tipo	Descrizione
`type`	string	Deve essere `"duration"`.
`delaySeconds`	uint64	Dimensione della finestra a cascata in secondi. Deve essere maggiore di 0.

Definire le regole di accumulo

Ogni regola di accumulo specifica come ridurre una finestra di messaggi in un singolo valore di output. La chiave di configurazione è rules.

Nella configurazione della trasformazione della finestra, aggiungere le regole di accumulo. Per ogni regola, specificare:

Impostazione	Descrizione
Inserimento	Campo da aggregare, ad esempio `temperature`.
Risultato	Nome del campo di output , ad esempio `avgTemperature`.
Expression	Funzione di aggregazione , ad esempio `average($1)`.

{
  key: 'rules'
  value: '{"accumulate":[{"inputs":["temperature"],"output":"avgTemperature","expression":"average($1)"}]}'
}

{
  "accumulate": [
    {
      "inputs": ["temperature"],
      "output": "avgTemperature",
      "expression": "average($1)"
    }
  ]
}

Proprietà	Obbligatorio	Descrizione
`inputs`	Sì	Elenco dei percorsi dei campi da leggere da ogni messaggio in arrivo.
`output`	Sì	Percorso del campo per il risultato aggregato. Ogni regola deve avere un output univoco.
`expression`	Sì	Formula che riduce i valori di input nella finestra a un singolo scalare. Deve contenere almeno una funzione di aggregazione.
`description`	No	Descrizione leggibile dagli umani

A differenza delle regole della mappa, expression è necessario per ogni regola di accumulo. L'uso $1 da solo non è valido perché fa riferimento a una raccolta di valori, non a un singolo scalare. Devi incapsularlo in una funzione di aggregazione come average($1).

Funzione di aggregazione

Funzione	Restituzioni	Comportamento della finestra vuota
`average`	Media dei valori numerici	Error
`sum`	Somma dei valori numerici	0.0
`min`	Valore numerico minimo	Error
`max`	Valore numerico massimo	Error
`count`	Numero di messaggi in cui esiste il campo	0
`first`	Primo valore nella finestra	Error
`last`	Ultimo valore nella finestra	Error

Ogni funzione accetta una singola variabile posizionale come argomento ($1 per il primo input, $2 per il secondo e così via).

Valori non numerici: le averagefunzioni , summin, e max ignorano automaticamente i valori non numerici.

Funzioni basate sulla presenza: count, firste last operano sulla presenza del campo indipendentemente dal tipo di valore.

Combinare le aggregazioni

È possibile combinare più funzioni di aggregazione in una singola espressione:

Aggiungere una regola con gli input temperature e humidity, e l'espressione average($1) + max($2).

{
  inputs: [ 'temperature', 'humidity' ]
  output: 'tempHumidityIndex'
  expression: 'average($1) + max($2)'
}

- inputs:
    - temperature     # $1
    - humidity        # $2
  output: tempHumidityIndex
  expression: "average($1) + max($2)"

Per convertire un valore aggregato, applicare la funzione di conversione all'esterno dell'aggregazione. Ad esempio, cToF(average($1)) converte la temperatura media in Fahrenheit.

Ogni funzione di aggregazione deve fare riferimento direttamente a una singola variabile posizionale. average($1) + max($2) è valido, ma average($1 + $2) non lo è.

Differenze rispetto alle regole della mappa

Capability	Le regole della mappa	Regole di accumulo
Espressione obbligatoria	No	Sì
Input con caratteri jolly	Supportato	Non supportato
`$metadata` Accesso	Supportato	Non supportato
`$context` Arricchimento	Supportato	Non supportato
`? $last` direttiva	Supportato	Non supportato
Tipo di contenuto di output	Corrisponde all'input	Sempre `application/json`

Esempio di configurazione completa

Configurazione della finestra completa che calcola le statistiche relative alla temperatura in una finestra di 30 secondi.

Se la finestra riceve questi tre messaggi:

{ "temperature": 21.5 }
{ "temperature": 23.0 }
{ "temperature": 19.8 }

Il messaggio di output è:

{
  "avgTemperature": 21.433333333333334,
  "minTemperature": 19.8,
  "maxTemperature": 23.0,
  "readingCount": 3,
  "tempRange": 3.2
}

Nell'ambiente Operazioni, crea un grafico del flusso di dati con una trasformazione di finestra:

Aggiungere una sorgente che legga da telemetry/temperature.
Aggiungi una trasformazione finestra. Impostare la durata della finestra su 30 secondi. Configurare le regole di accumulo per media, min, max, conteggio e intervallo nel temperature campo.
Aggiungere una destinazione che invia a telemetry/aggregated.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  name: 'temperature-window'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'aggregate'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-window:1.0.0'
          configuration: [
            {
              key: 'delay'
              value: '{"type":"duration","delaySeconds":30}'
            }
            {
              key: 'rules'
              value: '{"accumulate":[{"inputs":["temperature"],"output":"avgTemperature","expression":"average($1)"},{"inputs":["temperature"],"output":"minTemperature","expression":"min($1)"},{"inputs":["temperature"],"output":"maxTemperature","expression":"max($1)"},{"inputs":["temperature"],"output":"readingCount","expression":"count($1)"},{"inputs":["temperature"],"output":"tempRange","expression":"max($1) - min($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/aggregated'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'aggregate' } }
      { from: { name: 'aggregate' }, to: { name: 'output' } }
    ]
  }
}

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

    - nodeType: Graph
      name: aggregate
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-window:1.0.0
        configuration:
          - key: delay
            value: |
              {
                "type": "duration",
                "delaySeconds": 30
              }
          - key: rules
            value: |
              {
                "accumulate": [
                  {
                    "inputs": ["temperature"],
                    "output": "avgTemperature",
                    "expression": "average($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "minTemperature",
                    "expression": "min($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "maxTemperature",
                    "expression": "max($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "readingCount",
                    "expression": "count($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "tempRange",
                    "expression": "max($1) - min($1)"
                  }
                ]
              }

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

  nodeConnections:
    - from: { name: sensors }
      to: { name: aggregate }
    - from: { name: aggregate }
      to: { name: output }

Passaggi successivi

Trasformare i dati con mappa

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-03-25