Limiter les données dans les graphes de flux de données

Une transformation de manette limite la fréquence de transmission des messages sur un sujet MQTT. Au lieu d’écarter les messages en fonction de leur contenu, la transformation de limitation de débit écarte les messages en fonction du moment où elle les traite, en n’en transférant qu’un seul au maximum par modèle de rubrique au cours de chaque intervalle configuré. Utilisez la limitation de la vitesse pour protéger les systèmes en aval des sources en rafale ou à haute fréquence sans modifier le contenu du message.

Pour obtenir une vue d’ensemble des graphiques de flux de données et la façon dont les transformations composent dans un pipeline, consultez vue d’ensemble des graphiques de flux de données.

Prerequisites

Instance de Opérations Azure IoT déployée dans un cluster Kubernetes. Pour plus d’informations, consultez Deploy Opérations Azure IoT.

Le déploiement crée automatiquement un point de terminaison de registre par défaut nommé default qui pointe vers mcr.microsoft.com. Les transformations intégrées utilisent ce point de terminaison.

Limitation de mise à l’échelle des graphes à état

Important

Les transformations de fenêtre et de manette des gaz sont avec état. Chaque instance maintient son propre état et les instances ne partagent pas cet état entre elles. Lorsque le nombre d’instances du profil de flux de données est supérieur à un, les abonnements partagés répartissent les messages entre les instances, de sorte que chaque instance ne voit qu’un sous-ensemble des messages. Une transformée de fenêtre calcule ensuite des agrégations telles que les moyennes, les sommes et les comptages sur un jeu de données partiel, et une transformée de puissance impose la limite de débit configurée indépendamment dans chaque instance au lieu de couvrir l’ensemble du pipeline.

Réglez le nombre d’instances du profil de flux de données à 1 pour tout graphique de flux de données utilisant une transformation de fenêtre ou de régulation. Les graphes de flux de données sans état qui utilisent uniquement les transformations de cartographie, filtre, branchement et concaténation peuvent utiliser en toute sécurité des comptes d’instances plus élevés pour augmenter le débit.

Fonctionnement de la transformation throttle

La transformation de manette évalue le sujet de chaque message entrant selon une liste ordonnée de règles par sujet :

Première victoire de match. La transformation évalue les règles dans l’ordre. La première règle dont topic le motif correspond au sujet du message détermine si la transformation redirige le message. La transformation ne vérifie pas les règles ultérieures, même si elles correspondraient aussi.
Des sujets incomparables passent à travers. Si aucune règle ne correspond au sujet du message, la transformation transmet le message sans aucune limitation.
Le transfert est basé sur le temps, pas sur le comptage. Pour une règle correspondante, la transformation transmet le premier message qu’elle traite, puis supprime chaque message suivant sur ce motif jusqu’à ce que l’intervalle configuré (1000 / maxMessagesPerSecond millisecondes, arrondi vers le haut) se soit écoulé depuis le dernier message transféré. Ce comportement limite le taux maximal mais ne permet pas aux rafales de compenser les chutes précédentes.
État partagé par motif, pas par sujet. Lorsque le motif de la règle topic utilise un caractère générique, toutes les rubriques spécifiques qui y correspondent partagent le même état de limitation du débit dans une instance de transformation. Par exemple, une règle unique sensors/+ limite le taux combiné à travers sensors/temperature et sensors/humidity, et non chacun indépendamment.
0 laisse tout tomber. En réglant maxMessagesPerSecond0 pour une règle, chaque message correspondant au schéma de cette règle disparaît.
Le timing est basé sur le temps de traitement, pas sur le contenu du message. La transformation utilise son horloge monotone lorsqu’elle traite chaque message. Il ne lit pas de champ d’horodatage depuis la charge utile du message.

Note

La transformation de limitation de débit sert uniquement à déterminer s’il faut transférer ou ignorer un message. Il ne modifie jamais le contenu du message.

Configurez les règles de limitation du débit par sujet

Définissez les règles de limitation dans la clé de configuration throttle (pas rules) sous la forme d’un objet JSON avec un tableau perTopicThrottles.

Dans la configuration de la transformation de limitation du débit, ajoutez une ou plusieurs règles de limitation du débit. Pour chaque règle, spécifiez :

Setting	Description
Topic	Le modèle de sujet MQTT à faire correspondre. Prend en charge les caractères génériques `+` (à un niveau) et `#` (à plusieurs niveaux).
Nombre maximal de messages par seconde	Le taux maximal de transfert pour les sujets correspondant à ce schéma. Réglez sur `0` pour rejeter tous les messages correspondants.

L’interface CLI applique l’ensemble du graphique à partir d’un fichier config. Ajoutez cette configuration au configuration du nœud de transformation de l’accélérateur dans votre fichier graph.json, et appliquez-la à l’aide de az iot ops dataflowgraph apply.

La configuration est un objet JSON :

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

Ajoutez cette configuration dans le champ value sous forme de chaîne échappée, sous la clé throttle :

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

Conseil / Astuce

Pour générer la chaîne échappée, sauvegardez les règles dans un fichier comme rules.json, exécutez jq -c . rules.json, puis collez la sortie ligne unique dans le value champ.

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

Important

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.

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

Avec cette règle, la transformation transmet le premier sensors/temperature message traité, puis supprime tout autre message sur ce sujet traité moins de 100 millisecondes plus tard (1000 / 10). Cette règle n’affecte pas les sujets autres que sensors/temperature.

Chaque entrée dans perTopicThrottles possède ces propriétés :

Propriété	Obligatoire	Description
`topic`	Yes	Modèle de rubrique MQTT à faire correspondre. Prend en charge les jokers `+` (à niveau unique) et `#` (multiniveau, en position finale uniquement).
`maxMessagesPerSecond`	Yes	Taux maximal de redirection pour les sujets correspondant à ce schéma, en messages par seconde. La transformée permet des valeurs fractionnaires. Par exemple, `0.1` permet un message toutes les 10 secondes. Doit être zéro ou un nombre positif, fini. Réglez pour `0` supprimer tous les messages correspondant au schéma. Le timing a une précision de 1 milliseconde, donc les valeurs supérieures à `1000` ont la même limite effective que `1000`.

Important

Chaque topic motif dans perTopicThrottles doit être unique. La transformation n’accepte pas qu’une même chaîne de caractères de motif soit configurée plus d’une fois lors de l’initialisation.

Utilisez plusieurs règles par sujet

Comme la transformation évalue les règles dans l’ordre et que la première partie gagne, listez des motifs plus spécifiques avant des modèles plus généraux si vous voulez qu’ils aient leur propre limite de taux :

Ajoutez deux règles, dans cet ordre :

Commande	Sujet	Nombre maximal de messages par 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

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.

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

Les messages sur sensors/temperature correspondent à la première règle et sont limités à 10 messages par seconde. Les messages sur tout autre sensors/* sujet (par exemple, sensors/humidity) correspondent à la seconde règle et partagent une limite combinée d’un message par seconde.

Important

L’ordre est important. Si la sensors/# règle était listée en premier, elle correspondrait sensors/temperature aussi aux messages, et la transformation n’atteindrait jamais la règle plus spécifique.

Utilisez des caractères génériques pour réguler des groupes de rubriques

Le topic modèle prend en charge les mêmes caractères génériques que les filtres de rubrique MQTT :

Caractère générique	Correspondances	Exemple
`+`	Exactement un niveau thématique	`sensors/+/status` correspond à `sensors/line1/status` mais pas à `sensors/line1/sub/status`
`#`	Zéro ou plusieurs niveaux de rubrique restants, et doit constituer le dernier segment	`sensors/#` correspond à `sensors`, à `sensors/temperature` et à `sensors/line1/temperature`

Toutes les rubriques concrètes correspondant à la même règle avec caractère générique partagent le même état de limitation de débit.

Ajoutez une règle avec le sujet sensors/+ et un maximum de messages par seconde 1.

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

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

Important

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.

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

Avec cette règle, un message sur sensors/temperature et un message sur sensors/humidity, traités à 500 millisecondes d’intervalle, ne sont pas tous deux transmis : le second est ignoré, car il est comptabilisé dans le même intervalle partagé d’une seconde que le premier, quel que soit le sujet précis auquel il appartient.

Note

Un motif nu # correspond à chaque sujet et applique une limite de débit combinée unique dans chaque instance de transformation.

Supprimer tous les messages pour un sujet

Définissez maxMessagesPerSecond sur 0 pour supprimer tous les messages correspondant à un motif, sans retirer la règle ni le sujet de votre pipeline :

Ajoutez une règle avec le sujet debug/# et un maximum de messages par seconde 0.

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

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

Important

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.

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

Déploie un graphique de flux de données avec accélérateur

Pour appliquer la limitation de bout en bout, déployez un graphique de flux de données reliant une source, une transformation de manette et une destination. Utilisez l’outil qui correspond à votre flux de travail.

Dans l’environnement Opérations, créer un graphe de flux de données avec une transformation de limitation de débit :

Ajoutez une source qui lit depuis votre topic MQTT.
Ajoutez une transformation de limitation du débit. Ajoutez une ou plusieurs règles par sujet, classées du plus au moins spécifique.
Ajoutez une destination qui envoie à votre rubrique de sortie.

L’Azure CLI applique un graphe de flux de données à partir d’un seul fichier de configuration JSON. Créez un graph.json fichier avec les propriétés du graphique. Dans le graph.json fichier, chaque transformation stocke sa configuration dans le value champ sous forme de chaîne JSON échappée. Pour la version lisible de la configuration de limitation du débit, voir Configurer les règles de limitation du débit par rubrique. Appliquez-le en utilisant 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

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le 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

Ça ne modifie pas les messages. La transformation de limitation de débit se contente de transmettre ou de supprimer les messages ; elle n’en modifie pas le contenu.
Première victoire de match. La transformation n’applique que la première règle dont topic le motif correspond au sujet du message. Listez des motifs plus spécifiques avant les plus généraux.
État partagé pour chaque motif. Une règle de la carte sauvage partage sa limite de taux sur chaque sujet concret qu’elle correspond. Il n’y a pas de limite distincte par sujet.
Pas d’autorisation pour les explosions. La transformation impose un temps minimum entre les messages transférés par motif correspondant. Il n’accumule pas la capacité inutilisée des périodes antérieures et plus lentes.
Précision en millisecondes. L’intervalle minimum d’accélération est de 1 milliseconde, donc des valeurs supérieures à 1000 pour maxMessagesPerSecond n’augmentent pas le débit de transfert effectif au-delà de 1 000 messages par seconde.
Le timing est basé sur le temps de traitement, pas sur le contenu du message. La transformation utilise le temps de traitement de chaque message, et non l’heure à laquelle le courtier l’a reçu ni un champ d’horodatage dans la charge utile.
L’État est local et dans la mémoire. Chaque instance de profil de flux de données maintient son propre état de régulation. Redémarrer ou reconfigurer la transformation réinitialise cet état. Si un profil compte plusieurs instances, chaque instance applique le taux configuré indépendamment.
Les modèles de sujet dupliqués ne sont pas autorisés. Configurer la même topic chaîne plusieurs fois dans perTopicThrottles échoue lorsque la transformation s’initialise.

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-08-14

Limiter les données dans les graphes de flux de données

Prerequisites

Limitation de mise à l’échelle des graphes à état

Fonctionnement de la transformation throttle

Configurez les règles de limitation du débit par sujet

Utilisez plusieurs règles par sujet

Utilisez des caractères génériques pour réguler des groupes de rubriques

Supprimer tous les messages pour un sujet

Déploie un graphique de flux de données avec accélérateur

Limitations

Contenu connexe

Commentaires

Ressources supplémentaires