Gegevens beperken in gegevensstroomgrafieken

Een throttle-transformatiesnelheid beperkt hoe vaak berichten over een MQTT-onderwerp worden doorgestuurd. In plaats van berichten te laten vallen op basis van hun inhoud, laat de throttle-transformatie berichten vallen op basis van wanneer hij ze verwerkt, waarbij maximaal één bericht per topicpatroon binnen elk geconfigureerd interval wordt doorgestuurd. Gebruik throttling om downstream-systemen te beschermen tegen bursty of hoogfrequente bronnen zonder de inhoud van het bericht te veranderen.

Zie Het overzicht van gegevensstroomgrafieken voor een overzicht van gegevensstroomgrafieken en hoe transformaties in een pijplijn worden samengesteld.

Prerequisites

Een exemplaar van Azure IoT-bewerkingen geïmplementeerd in een Kubernetes-cluster. Zie Deploy Azure IoT-bewerkingen voor meer informatie.

Deployment maakt automatisch een standaard register-endpoint aan met de naam default die wijst naar mcr.microsoft.com. De ingebouwde transformaties gebruiken dit eindpunt.

Schaalbaarheidsbeperking voor statusbehoudende grafen

Important

Raam- en gashendeltransformaties zijn stateful. Elke instantie behoudt zijn eigen staat en instanties delen die toestand niet met elkaar. Wanneer het aantal instanties in het dataflowprofiel groter is dan één, verdelen gedeelde abonnementen berichten over instanties , zodat elke instantie slechts een deelverzameling van de berichten ziet. Een windowtransform berekent vervolgens aggregaties zoals gemiddelden, sommen en tellingen over een gedeeltelijke dataset, en een throttle-transformatie handhaaft de geconfigureerde snelheidslimiet onafhankelijk in elke instantie in plaats van over de hele pijplijn.

Stel het aantal instanties van het dataflowprofiel in op 1 voor elke datastroomgrafiek die een window- of throttle-transformatie gebruikt. Stateless dataflow-grafieken die alleen map-, filter-, branch- en concatenate-transformaties gebruiken, kunnen veilig hogere instance-tellingen gebruiken om de doorvoersnelheid te verhogen.

Hoe de gashendeltransformatie werkt

De throttle-transformatie evalueert het onderwerp van elk binnenkomend bericht aan de hand van een geordende lijst van regels per onderwerp:

Eerste wedstrijd wint. De transformatie evalueert regels in volgorde. De eerste regel waarvan het topic patroon overeenkomt met het onderwerp van het bericht, bepaalt of de transformatie het bericht doorstuurt. De transformatie controleert latere regels niet, zelfs als ze ook zouden overeenkomen.
Niet-overeenkomende onderwerpen worden doorgegeven. Als geen regel overeenkomt met het onderwerp van het bericht, stuurt de transformatie het bericht door zonder enige throttling.
Doorsturen is tijdsgebaseerd, niet telgebaseerd. Voor een matched rule stuurt de transformatie het eerste bericht dat het verwerkt door, en laat vervolgens elk volgend bericht op dat patroon vallen totdat het geconfigureerde interval (1000 / maxMessagesPerSecond milliseconden, afgerond naar boven) is verstreken sinds het laatst doorgestuurde bericht. Dit gedrag beperkt de maximale snelheid, maar laat bursts niet toe om eerdere drops te compenseren.
Gedeelde toestand per patroon, niet per onderwerp. Wanneer het topic-patroon van een regel een jokerteken gebruikt, delen alle specifieke topics die ermee overeenkomen dezelfde throttle-status binnen een transformatie-instantie. Bijvoorbeeld, een enkele sensors/+ regel beperkt de gecombineerde snelheid over sensors/temperature en sensors/humidity, niet elk onafhankelijk.
0 Laat alles vallen. Als je maxMessagesPerSecond instelt op 0, worden alle berichten die overeenkomen met het patroon van die regel verwijderd.
De timing is gebaseerd op verwerkingstijd, niet op de inhoud van het bericht. De transformatie gebruikt zijn monotone klok wanneer hij elk bericht verwerkt. Het leest geen tijdstempelveld uit de berichtpayload.

Note

De throttle-transformatie bepaalt alleen of een bericht wordt doorgestuurd of wordt verworpen. Het verandert nooit de inhoud van berichten.

Stel snelheidsbeperkingsregels per onderwerp in

Definieer de throttle-regels in de throttle configuratiesleutel (niet rules) als een JSON-object met een perTopicThrottles array.

Voeg in de gaskleptransformatieconfiguratie één of meer gasregels toe. Geef voor elke regel het volgende op:

Setting	Description
onderwerp	Het MQTT-themapatroon dat overeenkomt. Ondersteunt de `+`-wildcard (één niveau) en de `#`-wildcard (meerdere niveaus).
Maximaal berichten per seconde	De maximale doorstuursnelheid voor onderwerpen die aan dit patroon voldoen. Stel in op `0` om elk overeenkomend bericht te verwerpen.

De CLI past de hele grafiek uit één configuratiebestand toe. Voeg deze configuratie toe aan de throttle transform-nodes configuration in je graph.json bestand en pas deze toe met az iot ops dataflowgraph apply.

De configuratie is een JSON-object:

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

Voeg deze configuratie toe in het value-veld, als een ge-escapete tekenreeks, onder de sleutel throttle:

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

Tip

Om de ge-escapete tekenreeks te genereren, sla je de regels op in een bestand zoals rules.json, voer je jq -c . rules.json uit en plak je de eenregelige uitvoer in het veld value.

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

Important

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Met deze regel stuurt de transformatie het eerste sensors/temperature bericht dat hij verwerkt door, waarna verdere berichten over dat onderwerp die hij verwerkt minder dan 100 milliseconden later (1000 / 10) laat vallen. Deze regel heeft geen invloed op andere onderwerpen dan sensors/temperature.

Elke invoer in perTopicThrottles heeft de volgende eigenschappen:

Property	Required	Description
`topic`	Ja	Het MQTT-themapatroon om te matchen. Ondersteunt de jokertekens `+` (één niveau) en `#` (meerdere niveaus, alleen aan het einde).
`maxMessagesPerSecond`	Ja	Maximale doorstuursnelheid voor onderwerpen die aan dit patroon voldoen, in berichten per seconde. De transformatie maakt fractionele waarden mogelijk. Bijvoorbeeld, `0.1` staat één bericht elke 10 seconden toe. Moet nul zijn of een positief, eindig getal. Stel in op `0` om elk bericht te verwijderen dat overeenkomt met het patroon. Timing heeft een precisie van 1 milliseconde, dus waarden groter dan `1000` hebben dezelfde effectieve limiet als `1000`.

Important

Elk topic patroon in perTopicThrottles moet uniek zijn. De transformatie weigert het configureren van dezelfde patroonstring meer dan eens tijdens de initialisatie.

Gebruik meerdere regels per onderwerp

Omdat de transformatie de regels in volgorde beoordeelt en de eerste wedstrijd wint, vermeld dan meer specifieke patronen voordat je algemenere patronen ziet als je wilt dat ze hun eigen rate-limiet hebben:

Voeg twee regels toe, in deze volgorde:

Order	Onderwerp	Maximaal berichten per seconde
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}]}'
  }
]

Important

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Berichten op sensors/temperature voldoen aan de eerste regel en zijn beperkt tot 10 berichten per seconde. Berichten over een ander sensors/* onderwerp (bijvoorbeeld sensors/humidity) voldoen aan de tweede regel en delen een gecombineerde limiet van 1 bericht per seconde.

Important

Volgorde is belangrijk. Als de sensors/# regel als eerste werd vermeld, zou die ook overeenkomen sensors/temperature met berichten, en zou de transformatie nooit de meer specifieke regel bereiken.

Gebruik jokertekens om groepen topics te beperken

Het topic patroon ondersteunt dezelfde wildcards als MQTT-themafilters:

Wildcard	Overeenkomsten	Example
`+`	Precies één onderwerpniveau	`sensors/+/status` komt overeen met `sensors/line1/status` maar niet met `sensors/line1/sub/status`
`#`	Nul of meer overgebleven onderwerpniveaus, en het moet het laatste segment zijn	`sensors/#` komt overeen met `sensors`, `sensors/temperature` en `sensors/line1/temperature`

Alle concrete onderwerpen die aan dezelfde wildcard-regel voldoen, delen één gaspedaalstaat.

Voeg een regel toe met onderwerp sensors/+ en maximale berichten per seconde 1.

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

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

Important

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Met deze regel worden een bericht op sensors/temperature en een bericht op sensors/humidity verwerkt met 500 milliseconden uit elkaar niet allebei doorgestuurd—het tweede wordt verwijderd, omdat het meetelt tegen hetzelfde gedeelde interval van 1 seconde als het eerste, ongeacht over welk concreet onderwerp het gaat.

Note

Een kale # patroon komt overeen met elk onderwerp en past een enkele, gecombineerde snelheidslimiet toe binnen elke transformatie-instantie.

Verwijder alle berichten van een onderwerp

Stel maxMessagesPerSecond in op 0 om elk bericht te verwijderen dat overeenkomt met een patroon, zonder de regel of het onderwerp uit uw pijplijn te verwijderen:

Voeg een regel toe met onderwerp debug/# en maximale berichten per seconde 0.

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

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

Important

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Een datastroomgrafiek met throttling implementeren

Om throttling end-to-end toe te passen, zet je een datastroomgrafiek uit die een bron, een throttle-transformatie en een bestemming verbindt. Gebruik de tool die bij jouw workflow past.

Maak in Operations een datastroomgrafiek met een throttletransformatie:

Voeg een bron toe die wordt gelezen uit uw MQTT-onderwerp.
Voeg een gaskleptransformatie toe. Voeg één of meer regels per onderwerp toe, gerangschikt van meest naar minst specifiek.
Voeg een bestemming toe die naar uw uitvoeronderwerp wordt verzonden.

De Azure CLI past een dataflowgrafiek toe vanuit één JSON-configuratiebestand. Maak een graph.json bestand met de grafiekeigenschappen. In het graph.json bestand slaat elke transformatie zijn configuratie in het value veld op als een ontsnapte JSON-string. Voor de leesbare vorm van de smoorconfiguratie, zie Smoorregels per topic configureren. Pas het toe door gebruik te maken van 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' } }
    ]
  }
}

Important

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Ze wijzigt geen berichten. De throttle-transformatie stuurt berichten alleen door of laat ze vallen; de inhoud van berichten wordt niet gewijzigd.
Eerste wedstrijd wint. De transformatie past alleen de eerste regel toe waarvan het topic patroon overeenkomt met het onderwerp van het bericht. Noem meer specifieke patronen vóór meer algemene.
Gedeelde toestand volgens patroon. Een wildcardregel deelt zijn snelheidslimiet over elk concreet onderwerp dat het overeenkomt. Er is geen aparte limiet per onderwerp.
Geen piektoeslag. De transformatie handhaaft een minimale tijd tussen doorgestuurde berichten per gematcht patroon. Het verzamelt geen ongebruikte capaciteit uit eerdere, langzamere periodes.
Milliseconde precisie. Het minimale gasklepinterval is 1 milliseconde, dus waarden groter dan 1000 voor maxMessagesPerSecond verhogen de effectieve doorstuursnelheid niet boven de 1.000 berichten per seconde.
De timing is gebaseerd op verwerkingstijd, niet op de inhoud van het bericht. De transformatie gebruikt de tijd waarin elk bericht verwerkt wordt, niet de tijd waarop de broker het ontving of een tijdstempelveld in de payload.
De staat is lokaal en in het geheugen. Elk exemplaar van een gegevensstroomprofiel heeft zijn eigen snelheidsbeperkingsstatus. Het herstarten of herconfigureren van de transformatie reset die toestand. Als een profiel meerdere instanties heeft, handhaaft elke instantie onafhankelijk de geconfigureerde snelheid.
Dubbele themapatronen zijn niet toegestaan. Het configureren van dezelfde topic string meer dan eens in perTopicThrottles faalt wanneer de transformatie initialiseert.

Feedback

Is deze pagina nuttig?

Last updated on 2026-08-14

Gegevens beperken in gegevensstroomgrafieken

Prerequisites

Schaalbaarheidsbeperking voor statusbehoudende grafen

Hoe de gashendeltransformatie werkt

Stel snelheidsbeperkingsregels per onderwerp in

Gebruik meerdere regels per onderwerp

Gebruik jokertekens om groepen topics te beperken

Verwijder alle berichten van een onderwerp

Een datastroomgrafiek met throttling implementeren

Limitations

Verwante onderwerpen

Feedback

Aanvullende resources