Limitare i dati nei grafici di flusso di dati

Una trasformata di gas limita la frequenza con cui inoltra i messaggi su un argomento MQTT. Invece di scartare i messaggi in base al loro contenuto, la trasformazione throttle scarta i messaggi in base al momento in cui li elabora, inoltrando al massimo un messaggio per ciascun modello di topic all'interno di ogni intervallo configurato. Usa il throttling per proteggere i sistemi a valle da fonti a raffica o ad alta frequenza senza modificare il contenuto dei messaggi.

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.

Prerequisites

Istanza di Operazioni di Azure IoT distribuita in un cluster Kubernetes. Per altre informazioni, vedere Deploy Operazioni di Azure IoT.

La distribuzione crea automaticamente un endpoint di registro predefinito chiamato default che punta a mcr.microsoft.com. Le trasformazioni predefinite usano questo endpoint.

Limitazione del ridimensionamento per grafici con stato

Importante

Le trasformazioni di finestra e acceleratore sono stateful. Ogni istanza mantiene il proprio stato e le istanze non condividono quello stato tra loro. Quando il conteggio delle istanze del profilo di flusso dati è superiore a uno, gli abbonamenti condivisi distribuiscono i messaggi tra istanze, così che ogni istanza veda solo un sottoinsieme dei messaggi. Una trasformata finestra calcola quindi aggregazioni come medie, somme e conteggi su un dataset parziale, e una trasformata di gas impone il limite di velocità configurato in modo indipendente in ogni istanza invece che su tutta la pipeline.

Imposta il conteggio delle istanze del profilo di flusso dati a 1 per qualsiasi grafico di flusso dati che utilizza una trasformata a finestra o a manetta. I grafici di flusso dati senza stato che utilizzano solo trasformate mappa, filtro, ramo e concatenazione possono utilizzare in sicurezza conteggi di istanze più elevati per aumentare la produttività.

Come funziona la trasformazione dell'acceleratore

La trasformazione di limitazione valuta l'argomento di ogni messaggio in arrivo rispetto a un elenco ordinato di regole per argomento:

La prima partita vince. La trasformazione valuta le regole in ordine. La prima regola il cui topic pattern corrisponde all'argomento del messaggio determina se la trasformazione inoltra il messaggio. La trasformazione non controlla le regole successive, anche se anch’esse corrisponderebbero.
Gli argomenti non corrispondenti eseguono il pass-through. Se nessuna regola corrisponde all'argomento del messaggio, la trasformazione inoltra il messaggio senza alcuna limitazione della velocità.
L'inoltro è basato sul tempo, non sul conteggio. Per una regola abbinata, la trasformazione inoltra il primo messaggio che elabora, poi elimina ogni messaggio successivo su quel pattern fino a quando l'intervallo configurato (1000 / maxMessagesPerSecond millisecondi, arrotondato al su) non trascorre dall'ultimo messaggio inoltrato. Questo comportamento limita il tasso massimo ma non permette alle raffiche di compensare i drop precedenti.
Stato condiviso per criterio, non per argomento. Quando il criterio topic di una regola utilizza un carattere jolly, tutti gli argomenti concreti che lo corrispondono condividono lo stesso stato di limitazione all'interno di un'istanza di trasformazione. Ad esempio, una singola sensors/+ regola limita la velocità combinata tra sensors/temperature e sensors/humidity, non ciascuna in modo indipendente.
0 elimina tutto. Impostando maxMessagesPerSecond per 0 una regola elimina ogni messaggio che corrisponde al pattern di quella regola.
Il tempismo si basa sul tempo di elaborazione, non sul contenuto del messaggio. La trasformazione utilizza il proprio clock monotono quando elabora ciascun messaggio. Non legge un campo timestamp dal payload del messaggio.

Note

La trasformazione di limitazione decide solo se inoltrare o eliminare un messaggio. Non modifica mai il contenuto dei messaggi.

Configurare le regole di limitazione per argomento

Definisci le regole di accelerazione nella throttle chiave di configurazione (non rules) come un oggetto JSON con un perTopicThrottles array.

Nella configurazione della trasformazione di limitazione, aggiungere una o più regole di limitazione. Per ogni regola, specificare:

Setting	Description
Topic	Il criterio di argomento MQTT da corrispondere. Supporta i caratteri jolly `+` (a livello singolo) e `#` (a più livelli).
Massimo di messaggi al secondo	Il tasso massimo di inoltro per argomenti che corrispondono a questo schema. Impostare su `0` per eliminare tutti i messaggi corrispondenti.

La CLI applica l'intero grafo a partire da un file di configurazione. Aggiungere questa configurazione alla configuration del nodo della trasformazione di limitazione nel file graph.json e applicarla utilizzando az iot ops dataflowgraph apply.

La configurazione è un oggetto JSON:

{
  "perTopicThrottles": [
    {
      "topic": "sensors/temperature",
      "maxMessagesPerSecond": 10
    }
  ]
}

Aggiungi questa configurazione nel campo value come stringa con caratteri di escape, sotto la chiave throttle:

"configuration": [
  {
    "key": "throttle",
    "value": "{\"perTopicThrottles\":[{\"topic\":\"sensors/temperature\",\"maxMessagesPerSecond\":10}]}"
  }
]

Suggerimento

Per generare la stringa sfuggita, salva le regole in un file come rules.json, esegui jq -c . rules.json, e incolla l'output a singola riga nel value campo.

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"sensors/temperature","maxMessagesPerSecond":10}]}'
  }
]

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          {
            "topic": "sensors/temperature",
            "maxMessagesPerSecond": 10
          }
        ]
      }

Con questa regola, la trasformazione inoltra il primo sensors/temperature messaggio che elabora, poi elimina eventuali ulteriori messaggi su quell'argomento che elabora meno di 100 millisecondi dopo (1000 / 10). Questa regola non influisce sugli argomenti diversi che sensors/temperature.

Ogni voce in perTopicThrottles possiede queste proprietà:

Proprietà	Required	Description
`topic`	Sì	Modello di argomento MQTT da abbinare. Supporta i caratteri jolly `+` (a livello singolo) e `#` (a più livelli, solo in posizione finale).
`maxMessagesPerSecond`	Sì	Velocità massima di inoltro per argomenti che corrispondono a questo schema, in messaggi al secondo. La trasformata consente valori frazionari. Ad esempio, `0.1` permette un messaggio ogni 10 secondi. Deve essere zero o un numero positivo e finito. Impostare su `0` per scartare tutti i messaggi che corrispondono al modello. Il tempismo ha una precisione di 1 millisecondo, quindi valori maggiori di `1000` hanno lo stesso limite effettivo di `1000`.

Importante

Ogni schema in topic deve essere univoco. La trasformata rifiuta la configurazione della stessa stringa di pattern più di una volta durante l'inizializzazione.

Usa più regole per argomento

Poiché la trasformazione valuta le regole in ordine e vince la prima partita, elenca schemi più specifici prima di quelli più generali se vuoi che abbiano un proprio limite di tasso:

Aggiungi due regole, in quest'ordine:

Order	Topic	Massimo di messaggi al secondo
1	`sensors/temperature`	`10`
2	`sensors/#`	`1`

{
  "perTopicThrottles": [
    {
      "topic": "sensors/temperature",
      "maxMessagesPerSecond": 10
    },
    {
      "topic": "sensors/#",
      "maxMessagesPerSecond": 1
    }
  ]
}

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"sensors/temperature","maxMessagesPerSecond":10},{"topic":"sensors/#","maxMessagesPerSecond":1}]}'
  }
]

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          { "topic": "sensors/temperature", "maxMessagesPerSecond": 10 },
          { "topic": "sensors/#", "maxMessagesPerSecond": 1 }
        ]
      }

I messaggi su sensors/temperature corrispondono alla prima regola e sono limitati a 10 messaggi al secondo. I messaggi su qualsiasi altro sensors/* argomento (ad esempio, sensors/humidity) corrispondono alla seconda regola e condividono un limite combinato di 1 messaggio al secondo.

Importante

L'ordine è importante. Se la sensors/# regola fosse elencata per prima, corrisponderebbe sensors/temperature anche ai messaggi, e la trasformazione non raggiungerebbe mai la regola più specifica.

Usare i caratteri jolly per limitare i gruppi di argomenti

Il criterio topic supporta gli stessi caratteri jolly dei filtri degli argomenti MQTT:

Wildcard (Carattere jolly)	Corrispondenze	Example
`+`	Esattamente un livello tematico	`sensors/+/status` corrisponde a `sensors/line1/status` ma non a `sensors/line1/sub/status`
`#`	Zero o più livelli tematici rimanenti, e deve essere l'ultimo segmento	`sensors/#` corrisponde a `sensors`, `sensors/temperature`e `sensors/line1/temperature`

Tutti gli argomenti concreti che corrispondono alla stessa regola del carattere jolly condividono lo stesso stato di limitazione.

Aggiungi una regola con l'argomento sensors/+ e il numero massimo di messaggi al secondo 1.

{
  "perTopicThrottles": [
    {
      "topic": "sensors/+",
      "maxMessagesPerSecond": 1
    }
  ]
}

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"sensors/+","maxMessagesPerSecond":1}]}'
  }
]

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          { "topic": "sensors/+", "maxMessagesPerSecond": 1 }
        ]
      }

Con questa regola, un messaggio acceso sensors/temperature e un messaggio in sensors/humidity processato a 500 millisecondi di distanza non vengono entrambi inoltrati—il secondo viene eliminato, perché conta contro lo stesso intervallo condiviso di 1 secondo del primo, indipendentemente dall'argomento concreto trattato.

Note

Un pattern nudo # corrisponde a ogni argomento e applica un unico limite di velocità combinato all'interno di ogni istanza di trasformazione.

Elimina tutti i messaggi di un argomento

Imposta maxMessagesPerSecond su 0 per scartare ogni messaggio che corrisponde a uno schema, senza rimuovere la regola o l'argomento dalla pipeline:

Aggiungi una regola con l'argomento debug/# e il numero massimo di messaggi al secondo 0.

{
  "perTopicThrottles": [
    {
      "topic": "debug/#",
      "maxMessagesPerSecond": 0
    }
  ]
}

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"debug/#","maxMessagesPerSecond":0}]}'
  }
]

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          { "topic": "debug/#", "maxMessagesPerSecond": 0 }
        ]
      }

Distribuire un grafico del flusso di dati con limitazione

Per applicare il throttling end-to-end, si implementa un grafico di flusso dati che collega una sorgente, una trasformata di throttle e una destinazione. Usa lo strumento che si adatta al tuo flusso di lavoro.

Nell'esperienza operativa, creare un grafico del flusso di dati con una trasformazione di limitazione:

Aggiungere un'origine che esegue la lettura dall'argomento MQTT.
Aggiungi una trasformazione throttle. Aggiungi una o più regole per argomento, ordinate dal più al meno specifico.
Aggiungere una destinazione che invia all'argomento di output.

La interfaccia della riga di comando di Azure applica un grafo di flusso dati da un singolo file di configurazione JSON. Creare un graph.json file con le proprietà del grafo. Nel graph.json file, ogni trasformazione memorizza la sua configurazione nel value campo come una stringa JSON sfuggita. Per la forma leggibile della configurazione dell'acceleratore, vedi Configura regole per argomento. Applicalo usando az iot ops dataflowgraph apply.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensors/#"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "throttle",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-throttle:1.0.0",
        "configuration": [
          {
            "key": "throttle",
            "value": "{\"perTopicThrottles\":[{\"topic\":\"sensors/temperature\",\"maxMessagesPerSecond\":10},{\"topic\":\"sensors/#\",\"maxMessagesPerSecond\":1}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/throttled"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "throttle"
      }
    },
    {
      "from": {
        "name": "throttle"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'sensors-throttle'
  parent: dataflowProfile
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'sensors/#' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'throttle'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-throttle:1.0.0'
          configuration: [
            {
              key: 'throttle'
              value: '{"perTopicThrottles":[{"topic":"sensors/temperature","maxMessagesPerSecond":10},{"topic":"sensors/#","maxMessagesPerSecond":1}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/throttled'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'throttle' } }
      { from: { name: 'throttle' }, to: { name: 'output' } }
    ]
  }
}

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: sensors-throttle
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensors/#
    - nodeType: Graph
      name: throttle
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-throttle:1.0.0
        configuration:
          - key: throttle
            value: |
              {
                "perTopicThrottles": [
                  { "topic": "sensors/temperature", "maxMessagesPerSecond": 10 },
                  { "topic": "sensors/#", "maxMessagesPerSecond": 1 }
                ]
              }
    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/throttled
  nodeConnections:
    - from: { name: sensors }
      to: { name: throttle }
    - from: { name: throttle }
      to: { name: output }

Limitations

Non modifica i messaggi. La trasformazione dell'acceleratore inoltra o interrompe solo i messaggi; Non cambia il contenuto del messaggio.
La prima partita vince. La trasformazione applica solo la prima regola il cui topic pattern corrisponde all'argomento del messaggio. Elenca modelli più specifici prima di quelli più generali.
Stato condiviso per ogni modello. Una regola jolly condivide il suo limite di tasso su ogni argomento concreto che affronta. Non c'è un limite separato per argomento.
Nessuna autorizzazione di burst. La trasformazione impone un tempo minimo tra i messaggi inoltrati per ogni pattern corrispondente. Non accumula capacità inutilizzata di periodi precedenti e più lenti.
Precisione millisecondo. L'intervallo minimo di acceleratore è di 1 millisecondo, quindi valori superiori a 1000 per maxMessagesPerSecond non aumentano la velocità effettiva di inoltro oltre 1.000 messaggi al secondo.
Il tempismo si basa sul tempo di elaborazione, non sul contenuto del messaggio. La trasformazione utilizza il tempo in cui elabora ogni messaggio, non l'orario in cui il broker lo ha ricevuto o un campo di timestamp nel payload.
Lo stato è locale e in memoria. Ogni istanza di profilo di flusso di dati mantiene il proprio stato di limitazione. Riavviare o riconfigurare la trasformazione resetta quello stato. Se un profilo ha più istanze, ogni istanza applica indipendentemente il tasso configurato.
Non sono permessi schemi di argomento duplicati. La configurazione della stessa stringa topic più di una volta in perTopicThrottles non riesce quando la trasformazione inizializza.

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-08-14

Limitare i dati nei grafici di flusso di dati

Prerequisites

Limitazione del ridimensionamento per grafici con stato

Come funziona la trasformazione dell'acceleratore

Configurare le regole di limitazione per argomento

Usa più regole per argomento

Usare i caratteri jolly per limitare i gruppi di argomenti

Elimina tutti i messaggi di un argomento

Distribuire un grafico del flusso di dati con limitazione

Limitations

Contenuti correlati

Commenti e suggerimenti

Risorse aggiuntive