Daten in Datenflussdiagrammen drosseln

Eine Throttle-Transformationsrate begrenzt, wie oft sie Nachrichten zu einem MQTT-Thema weiterleitet. Anstatt Nachrichten basierend auf ihrem Inhalt zu löschen, wirft die Throttle-Transformation Nachrichten basierend darauf, wann sie sie verarbeitet, und leitet höchstens eine Nachricht pro Thema-Muster innerhalb jedes konfigurierten Intervalls weiter. Verwenden Sie Drosselung, um nachgelagerte Systeme vor Bursty- oder Hochfrequenzquellen zu schützen, ohne den Nachrichteninhalt zu ändern.

Eine Übersicht über Datenflussdiagramme und die Art und Weise, wie Transformationen in einer Pipeline erstellt werden, finden Sie in der Übersicht über Datenflussdiagramme.

Prerequisites

• Eine Instanz von Azure IoT Einsatz, die in einem Kubernetes-Cluster bereitgestellt wird. Weitere Informationen finden Sie unter Deploy Azure IoT Einsatz.

• Die Bereitstellung erstellt automatisch einen Standard-Registry-Endpunkt namens default , der auf mcr.microsoft.comverweist. Die integrierten Transformationen verwenden diesen Endpunkt.

Skalierungsbeschränkung für zustandsbehaftete Diagramme

Wie die Gashebeltransformation funktioniert

Die Drosselungstransformation bewertet das Thema jeder eingehenden Nachricht anhand einer geordneten Liste von Regeln pro Thema:

• Der Gewinner ist derjenige, der das erste Spiel gewinnt. Die Transformation bewertet die Regeln in der Reihenfolge. Die erste Regel, deren topic Muster mit dem Thema der Nachricht übereinstimmt, bestimmt, ob die Transformation die Nachricht weiterleitet. Die Transformation prüft spätere Regeln nicht, selbst wenn sie ebenfalls übereinstimmen würden.
• Nicht übereinstimmende Themen werden durchgelassen. Wenn keine Regel mit dem Thema der Nachricht übereinstimmt, leitet die Transformation die Nachricht ohne Throttling weiter.
• Die Weiterleitung erfolgt zeitbasiert, nicht zählbasiert. Bei einer abgestimmten Regel leitet die Transformation die erste von ihr verarbeitete Nachricht weiter und lässt dann jede nachfolgende Nachricht in diesem Muster fallen, bis das konfigurierte Intervall (1000 / maxMessagesPerSecond Millisekunden, aufgerundet) seit der letzten weitergeleiteten Nachricht verstrichen ist. Dieses Verhalten begrenzt die maximale Rate, erlaubt aber keine Bursts, um frühere Drops auszugleichen.
• Gemeinsamer Zustand pro Muster, nicht pro Topic. Wenn das topic-Muster einer Regel einen Platzhalter verwendet, teilen alle konkreten Themen, die ihr entsprechen, innerhalb einer Transformationsinstanz denselben Drosselungszustand. Zum Beispiel begrenzt eine einzelne sensors/+ Regel die kombinierte Rate über sensors/temperature und sensors/humidity, nicht jede einzelne unabhängig voneinander.
• 0 lässt alles fallen. Das Festlegen von maxMessagesPerSecond auf 0 für eine Regel verwirft jede Nachricht, die dem Muster dieser Regel entspricht.
• Das Timing basiert auf der Bearbeitungszeit, nicht auf dem Inhalt der Nachrichten. Die Transformation verwendet ihre monotone Uhr, wenn sie jede Nachricht verarbeitet. Es liest kein Zeitstempelfeld aus dem Message-Payload.

Konfiguriere die Drosselregeln pro Thema

Definiere Drosselregeln im Konfigurationsschlüssel throttle (nicht rules) als JSON-Objekt mit einem perTopicThrottles Array.

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

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

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

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

Mit dieser Regel leitet die Transformation die erste sensors/temperature von ihr verarbeitete Nachricht weiter und entfernt dann alle weiteren Nachrichten zu diesem Thema, die sie weniger als 100 Millisekunden später verarbeitet (1000 / 10). Diese Regel betrifft keine anderen Themen als sensors/temperature.

Jeder Eintrag in perTopicThrottles besitzt folgende Eigenschaften:

Verwenden Sie mehrere Regeln pro Thema

Da die Transformation die Regeln der Reihe nach bewertet und das erste Match gewinnt, listen Sie spezifischere Muster auf, bevor sie allgemeinere auflisten, wenn Sie möchten, dass sie ein eigenes Rate-Limit haben:

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

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

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

Nachrichten auf sensors/temperature entsprechen der ersten Regel und sind auf 10 Nachrichten pro Sekunde begrenzt. Nachrichten zu einem anderen sensors/* Thema (zum Beispiel sensors/humidity) entsprechen der zweiten Regel und teilen sich ein kombiniertes Limit von 1 Nachricht pro Sekunde.

Themengruppen mittels Platzhalter drosseln

Das Muster topic unterstützt dieselben Wildcards wie MQTT-Themenfilter:

Alle konkreten Themen, die derselben Platzhalterregel entsprechen, teilen sich einen Drosselungszustand.

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

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

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

Mit dieser Regel werden eine Nachricht auf sensors/temperature und eine Nachricht auf sensors/humidity, die im Abstand von 500 Millisekunden verarbeitet werden, nicht beide weitergeleitet – die zweite wird verworfen, weil sie auf dasselbe gemeinsame 1‑Sekunden‑Intervall wie die erste angerechnet wird, unabhängig davon, zu welchem konkreten Topic sie gehört.

Alle Nachrichten zu einem Thema verwerfen

Setze maxMessagesPerSecond auf 0 so, dass jede Nachricht, die einem Muster entspricht, droppt, ohne die Regel oder das Thema aus deiner Pipeline zu entfernen:

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

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

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

Ein Datenflussdiagramm mit Drosselung bereitstellen

Um End-to-End-Throttling anzuwenden, sollte ein Datenflussgraph bereitgestellt werden, der eine Quelle, eine Throttle-Transformation und ein Ziel verbindet. Verwenden Sie das Tool, das zu Ihrem Arbeitsablauf passt.

{
  "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' } }
    ]
  }
}

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 }

Einschränkungen

• Verändert keine Nachrichten. Die Throttle-Transformation leitet Nachrichten nur weiter oder verwirft sie; sie ändert deren Inhalt nicht.
• Der Gewinner ist derjenige, der das erste Spiel gewinnt. Die Transformation gilt nur für die erste Regel, deren topic Muster mit dem Thema der Nachricht übereinstimmt. Führen Sie spezifischere Muster vor allgemeineren auf.
• Geteilter Zustand gemäß Muster. Eine Wildcard-Regel teilt ihr Ratenlimit für jedes konkrete Thema, das sie trifft. Es gibt kein separates Limit pro Thema.
• Keine Spitzen. Die Transformation setzt eine Mindestzeit zwischen den weitergeleiteten Nachrichten pro abgestimmtem Muster vor. Es sammelt keine ungenutzte Kapazität aus früheren, langsameren Phasen an.
• Millisekundenpräzision. Das minimale Drosselintervall beträgt 1 Millisekunde, daher erhöhen höhere Werte als 1000 für maxMessagesPerSecond die effektive Weiterleitungsrate nicht über 1.000 Nachrichten pro Sekunde.
• Das Timing basiert auf der Bearbeitungszeit, nicht auf dem Inhalt der Nachrichten. Die Transformation verwendet die Zeit, in der jede Nachricht verarbeitet wird, nicht die Zeit, zu der der Broker sie empfangen hat, oder ein Zeitstempelfeld in der Nutzlast.
• Der Zustand ist lokal und im Speicher. Jede Datenflussprofil-Instanz behält ihren eigenen Drosselungszustand. Ein Neustart oder Neukonfiguration der Transformation setzt diesen Zustand zurück. Wenn ein Profil mehrere Instanzen hat, setzt jede Instanz die konfigurierte Rate unabhängig durch.
• Doppelte Themenmuster sind nicht erlaubt. Die Konfiguration desselben topic Strings mehrmals in perTopicThrottles schlägt fehl, wenn die Transformation initialisiert wird.

Verwandte Inhalte

• Transformieren von Daten mit Karte
• Filtern und Weiterleiten von Daten
• Aggregieren von Daten im Laufe der Zeit
• Anreichern mit externen Daten
• Ausdrucksreferenz
• Konfigurieren einer Quelle
• Konfigurieren eines Ziels