Begränsa data i dataflödesgrafer

En throttle-transformhastighet begränsar hur ofta den vidarebefordrar meddelanden om ett MQTT-ämne. Istället för att släppa meddelanden baserat på deras innehåll, släpper throttle-transformen meddelanden baserat på när den bearbetar dem, och vidarebefordrar högst ett meddelande per ämnesmönster inom varje konfigurerat intervall. Använd strypning för att skydda nedströms system från burstiga eller högfrekventa källor utan att ändra meddelandeinnehållet.

En översikt över dataflödesdiagram och hur transformeringar består i en pipeline finns i Översikt över dataflödesdiagram.

Prerequisites

En instans av Azure IoT Operations distribuerad i ett Kubernetes-kluster. Mer information finns i Deploy Azure IoT Operations.

Distributionen skapar automatiskt en standardregister-endpoint med namnet default som pekar på mcr.microsoft.com. De inbyggda transformeringarna använder den här slutpunkten.

Skalningsbegränsning för tillståndskänsliga grafer

Important

Fönster- och gasreglage transformerar är statele. Varje instans behåller sitt eget tillstånd och instanserna delar inte det tillståndet med varandra. När antalet dataflödesprofiler är fler än ett distribuerar delade prenumerationer meddelanden mellan instanserna, så varje instans ser endast en delmängd av meddelandena. En fönstertransform beräknar sedan aggregationer som medelvärden, summor och räkningar över en partiell dataset, och en throttle-transform upprätthåller den konfigurerade hastighetsgränsen oberoende i varje instans istället för över hela pipelinen.

Sätt antalet data flow-profiler till 1 för alla dataflödesgrafer som använder en fönster- eller throttle-transform. Tillståndslösa dataflödesgrafer som endast använder mapp-, filter-, gren- och koncatenationstransformer kan säkert använda högre instansantal för att öka genomströmningen.

Hur strypningstransformeringen fungerar

throttle-transform utvärderar varje inkommande meddelandes ämne mot en ordnad lista med ämnesspecifika regler:

Första matchen vinner. Transformen utvärderar regler i ordning. Den första regeln vars topic mönster matchar meddelandets ämne avgör om transformen vidarebefordrar meddelandet. Transformeringen kontrollerar inte efterföljande regler, även om de också skulle matcha.
Omatchade ämnen passerar förbi. Om ingen regel matchar meddelandets ämne, vidarebefordrar transformen meddelandet utan någon strypning.
Vidarebefordran är tidsbaserad, inte räknebaserad. För en matchad regel vidarebefordrar transformen det första meddelandet den bearbetar, och släpper sedan varje efterföljande meddelande på det mönstret tills det konfigurerade intervallet (1000 / maxMessagesPerSecond millisekunder, avrundat uppåt) har gått sedan det senaste vidarebefordrade meddelandet. Detta beteende begränsar maxhastigheten men tillåter inte burstar att kompensera för tidigare dropp.
Delat tillstånd per mönster, inte per ämne. När ett regelmönster topic använder ett jokerkort delar alla konkreta ämnen som matchar det samma throttle-tillstånd inom en transforminstans. Till exempel begränsar en enda sensors/+ regel den kombinerade hastigheten över sensors/temperature och sensors/humidity, inte varje regel oberoende av varandra.
0 tappar allt. Att ställa in maxMessagesPerSecond till 0 för en regel gör att alla meddelanden som matchar regelns matchningsmönster ignoreras.
Tidpunkten baseras på bearbetningstid, inte meddelandeinnehåll. Transformen använder sin monotona klocka när den bearbetar varje meddelande. Den läser inte ett tidsstämpelfält från meddelandens nyttolast.

Note

Gasregulatorn avgör bara om ett meddelande ska vidarebefordras eller släppas. Den ändrar aldrig meddelandeinnehållet.

Konfigurera begränsningsregler per ämne

Definiera spakregler i konfigurationsnyckeln throttle (inte rules) som ett JSON-objekt med en perTopicThrottles array.

I konfigurationen för begränsningstransformering lägger du till en eller flera begränsningsregler. För varje regel anger du:

Inställning	Description
Topic	MQTT-ämnesmönster för att matcha. Stöder `+` (enkelnivå) och `#` (flernivå) jokerkort.
Maxmeddelanden per sekund	Den maximala vidarebefordringshastigheten för ämnen som matchar detta mönster. Ange `0` för att kassera alla matchande meddelanden.

CLI tillämpar hela diagrammet från en konfigurationsfil. Lägg till den här konfigurationen i throttle-transformnodens configuration i din graph.json-fil, och tillämpa den med az iot ops dataflowgraph apply.

Konfigurationen är ett JSON-objekt:

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

Lägg till den här konfigurationen i fältet value som en sträng med escape-tecken, under nyckeln throttle:

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

Tips/Råd

För att generera den escaped-strängen sparar du reglerna i en fil med namn som rules.json, kör jq -c . rules.json och klistrar in utdata på en enda rad i fältet value.

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

Important

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

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

Med denna regel vidarebefordrar transformen det första sensors/temperature meddelandet den bearbetar, och släpper sedan eventuella ytterligare meddelanden om ämnet som den bearbetar mindre än 100 millisekunder senare (1000 / 10). Denna regel påverkar inte andra ämnen än sensors/temperature.

Varje post i perTopicThrottles har dessa egenskaper:

Property	Required	Description
`topic`	Yes	MQTT-topicmönster att matcha mot. Stöder jokertecknen `+` (enkelnivå) och `#` (flernivå, endast i slutet).
`maxMessagesPerSecond`	Yes	Maximal vidarebefordran för ämnen som matchar detta mönster, i meddelanden per sekund. Transformen tillåter bråkdelsvärden. Till exempel `0.1` tillåter ett meddelande var tionde sekund. Måste vara noll eller ett positivt, ändligt tal. Ange `0` för att kassera alla meddelanden som matchar mönstret. Timing har 1 millisekunds precision, så värden större än `1000` har samma effektiva gräns som `1000`.

Important

Varje mönster topic i perTopicThrottles måste vara unikt. Transformen avvisar att samma mönstersträng konfigureras mer än en gång under initialiseringen.

Använd flera regler per ämne

Eftersom transformen utvärderar regler i ordning och den första matchen vinner, lista mer specifika mönster innan mer generella om du vill att de ska ha sin egen hastighetsgräns:

Lägg till två regler, i denna ordning:

Order	Topic	Maxmeddelanden per sekund
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

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

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

Meddelanden på sensors/temperature motsvarar den första regeln och är begränsade till 10 meddelanden per sekund. Meddelanden om något annat sensors/* ämne (till exempel sensors/humidity) matchar den andra regeln och delar en sammanlagd gräns på 1 meddelande per sekund.

Important

Ordningen är viktig. Om regeln sensors/# listades först skulle den också matcha sensors/temperature meddelanden, och transformen skulle aldrig nå den mer specifika regeln.

Använd jokertecken för att begränsa grupper av ämnen

Mönstret topic stöder samma jokerkort som MQTT-ämnesfilter:

Jokertecken	Matchningar	Example
`+`	Precis en ämnesnivå	`sensors/+/status` matchar `sensors/line1/status` men inte `sensors/line1/sub/status`
`#`	Noll eller fler kvarvarande ämnesnivåer, och måste vara det sista segmentet	`sensors/#` matchar `sensors`, `sensors/temperature` och `sensors/line1/temperature`

Alla specifika ämnen som matchar samma wildcard-regel delar samma begränsningstillstånd.

Lägg till en regel med ämne sensors/+ och max meddelanden per sekund 1.

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

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

Important

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

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

Med denna regel vidarebefordras inte ett meddelande på sensors/temperature och ett meddelande på sensors/humidity som bearbetas med 500 millisekunders mellanrum – det andra tas bort eftersom det räknas mot samma gemensamma 1-sekundersintervall som det första, oavsett vilket konkret ämne det handlar om.

Note

Ett rent # mönster matchar varje ämne och applicerar en enda, kombinerad hastighetsgräns inom varje transforminstans.

Ta bort alla meddelanden för ett ämne

Ställ maxMessagesPerSecond in på 0 att släppa varje meddelande som matchar ett mönster, utan att ta bort regeln eller ämnet från din pipeline:

Lägg till en regel med ämne debug/# och max meddelanden per sekund 0.

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

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

Important

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

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

Distribuera en dataflödesgraf med throttle

För att tillämpa strypning från början till slut, distribuera en dataflödesgraf som kopplar ihop en källa, en throttle-transform och en destination. Använd verktyget som matchar ditt arbetsflöde.

I Operations-miljön skapar du en dataflödesgraf med en strypningstransformering:

Lägg till en källa som läser från ditt MQTT-ämne.
Lägg till en strypningsomvandling. Lägg till en eller flera regler per ämne, ordnade från mest till minst specifika.
Lägg till ett mål som skickar till ditt utdataämne.

Azure CLI tillämpar en dataflödesgraf från en enda JSON-konfigurationsfil. Skapa en graph.json fil med grafegenskaperna. I graph.json filen lagrar varje transform sin konfiguration i value fältet som en undkommet JSON-sträng. För den läsbara formen av gasreglagskonfigurationen, se Konfigurera regler för spjällning per ämne. Applicera det genom att använda 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

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

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

Ändrar inte meddelanden. Gasreglaget vidarebefordrar eller släpper bara meddelanden; Det förändrar inte meddelandets innehåll.
Första matchen vinner. Transformen tillämpar endast den första regeln vars topic mönster matchar meddelandets ämne. Lista mer specifika mönster innan mer generella.
Delat tillstånd enligt mönster. En wildcard-regel delar sin hastighetsgräns över varje konkret ämne den matchar. Det finns ingen separat gräns per ämne.
Ingen burst-ersättning. Transformen upprätthåller en minsta tid mellan vidarebefordrade meddelanden per matchat mönster. Den ackumulerar inte oanvänd kapacitet från tidigare, långsammare perioder.
Millisekundprecision. Minsta gasintervall är 1 millisekund, så värden som är högre än 1000 för maxMessagesPerSecond ökar inte den effektiva vidarebefordrande hastigheten över 1 000 meddelanden per sekund.
Tidpunkten baseras på bearbetningstid, inte meddelandeinnehåll. Transformen använder tiden den bearbetar varje meddelande, inte tiden som mäklaren tog emot det eller ett tidsstämpelfält i nyttolasten.
Delstaten är lokal och i minnet. Varje dataflödesprofilinstans behåller sitt eget throttle-tillstånd. Att starta om eller konfigurera om transformen återställer det tillståndet. Om en profil har flera instanser upprätthåller varje instans den konfigurerade hastigheten oberoende.
Dubbletter av ämnesmönster är inte tillåtna. Att konfigurera samma topic sträng mer än en gång i perTopicThrottles misslyckas när transformen initieras.

Feedback

Var den här sidan till hjälp?

Last updated on 2026-08-14

Begränsa data i dataflödesgrafer

Prerequisites

Skalningsbegränsning för tillståndskänsliga grafer

Hur strypningstransformeringen fungerar

Konfigurera begränsningsregler per ämne

Använd flera regler per ämne

Använd jokertecken för att begränsa grupper av ämnen

Ta bort alla meddelanden för ett ämne

Distribuera en dataflödesgraf med throttle

Limitations

Relaterat innehåll

Feedback

Ytterligare resurser