Daten in Datenflussdiagrammen drosseln

In der Drosselungstransformationskonfiguration fügen Sie eine oder mehrere Drosselungsregeln hinzu. Geben Sie für jede Regel Folgendes an:

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Füge diese Konfiguration zum configuration des Throttle-Transform-Knotens in deiner graph.json-Datei hinzu und wende sie mit az iot ops dataflowgraph apply an.

Die Konfiguration ist ein JSON-Objekt:

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

Fügen Sie diese Konfiguration im Feld value als Zeichenfolge mit Escapezeichen unter dem throttle-Schlüssel hinzu:

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

Fügen Sie zwei Regeln hinzu, in folgender Reihenfolge:

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

Füge eine Regel mit Thema sensors/+ hinzu und maximiere Nachrichten pro Sekunde 1.

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

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

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

Füge eine Regel mit Thema debug/# hinzu und maximiere Nachrichten pro Sekunde 0.

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

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

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

Erstellen Sie in der Betriebsumgebung ein Datenflussdiagramm mit einer Drosselungstransformation:

1. Fügen Sie eine Quelle hinzu, die aus Ihrem MQTT-Topic gelesen wird.
2. Fügen Sie eine Gaspedal-Transformation hinzu. Füge pro Thema eine oder mehrere Regeln hinzu, von den am wenigsten bis am wenigsten spezifischen.
3. Fügen Sie ein Ziel hinzu, das an Ihr Ausgabe-Topic sendet.

Die Azure CLI wendet einen Datenflussgraphen aus einer einzelnen JSON-Konfigurationsdatei an. Erstellen Sie eine graph.json Datei mit den Diagrammeigenschaften. In der graph.json Datei speichert jede Transformation ihre Konfiguration im Feld value als entkommene JSON-Zeichenkette. Die lesbare Form der Drosselungskonfiguration finden Sie unter Drosselungsregeln pro Thema konfigurieren. Wende es mit az iot ops dataflowgraph apply an.

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

Wenden Sie die Konfiguration auf das Profil der vorhandenen Instanz default an:

az iot ops dataflowgraph apply \
  --name sensors-throttle \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

Speichere die folgende Vorlage als main.bicep.

param aioInstanceName string
param customLocationName string

resource aioInstance 'Microsoft.IoTOperations/instances@2026-07-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource dataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-07-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-07-01' = {
  name: 'sensors-throttle'
  parent: dataflowProfile
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  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' }
      }
    ]
  }
}

Setze CUSTOM_LOCATION_NAME auf den benutzerdefinierten Standortnamen der bestehenden Instanz und deploye dann in der Ressourcengruppe, die die Instanz und den benutzerdefinierten Standort enthält:

az deployment group create \
  --resource-group $RESOURCE_GROUP \
  --template-file main.bicep \
  --parameters aioInstanceName=$AIO_INSTANCE_NAME customLocationName=$CUSTOM_LOCATION_NAME

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 }

Erstellen Sie in der Betriebsumgebung ein Datenflussdiagramm namens rate-limit-telemetry im Profil default:

1. Füge eine Quelle namens factory-mqtt mit Endpunkt factory/readings und Thema source hinzu.
2. Fügen Sie die integrierte Drosselungs-Transformation namens throttle hinzu. Füge eine Pro-Topic-Regel mit Thema factory/readings und Maximale Nachrichten pro Sekunde auf 2 gesetzt hinzu.
3. Fügen Sie ein Ziel namens destination hinzu, mit dem Endpunkt factory-mqtt und dem Thema factory/rate-limited.
4. Verbinden Sie source mit throttle, dann throttle mit destination. Speichere und aktiviere das Diagramm.

Die Drosselung leitet Nachrichten unverändert mit mindestens 500 Millisekunden Abstand weiter und verwirft überschüssige Nachrichten. Es puffert keine Nachrichten und wählt nicht den neuesten Wert aus.

Speichere diese vollständigen Grapheigenschaften als graph.json:

{
  "mode": "Enabled",
  "nodes": [
    {
      "name": "source",
      "nodeType": "Source",
      "sourceSettings": {
        "endpointRef": "factory-mqtt",
        "dataSources": ["factory/readings"]
      }
    },
    {
      "name": "throttle",
      "nodeType": "Graph",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-throttle:1.0.0",
        "configuration": [
          {
            "key": "throttle",
            "value": "{\"perTopicThrottles\":[{\"topic\":\"factory/readings\",\"maxMessagesPerSecond\":2}]}"
          }
        ]
      }
    },
    {
      "name": "destination",
      "nodeType": "Destination",
      "destinationSettings": {
        "endpointRef": "factory-mqtt",
        "dataDestination": "factory/rate-limited"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": { "name": "source" },
      "to": { "name": "throttle" }
    },
    {
      "from": { "name": "throttle" },
      "to": { "name": "destination" }
    }
  ]
}

Wenden Sie die Konfiguration auf das Profil der bestehenden Instanz default an:

az iot ops dataflowgraph apply \
  --name rate-limit-telemetry \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

Speichern Sie die folgende Vorlage unter main.bicep:

param aioInstanceName string
param customLocationName string

resource aioInstance 'Microsoft.IoTOperations/instances@2026-07-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource dataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-07-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-07-01' = {
  name: 'rate-limit-telemetry'
  parent: dataflowProfile
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        name: 'source'
        nodeType: 'Source'
        sourceSettings: {
          endpointRef: 'factory-mqtt'
          dataSources: [ 'factory/readings' ]
        }
      }
      {
        name: 'throttle'
        nodeType: 'Graph'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-throttle:1.0.0'
          configuration: [
            {
              key: 'throttle'
              value: '{"perTopicThrottles":[{"topic":"factory/readings","maxMessagesPerSecond":2}]}'
            }
          ]
        }
      }
      {
        name: 'destination'
        nodeType: 'Destination'
        destinationSettings: {
          endpointRef: 'factory-mqtt'
          dataDestination: 'factory/rate-limited'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'source' }
        to: { name: 'throttle' }
      }
      {
        from: { name: 'throttle' }
        to: { name: 'destination' }
      }
    ]
  }
}

Setze CUSTOM_LOCATION_NAME auf den benutzerdefinierten Standortnamen der bestehenden Instanz und deploye dann in der Ressourcengruppe, die die Instanz und den benutzerdefinierten Standort enthält:

az deployment group create \
  --resource-group $RESOURCE_GROUP \
  --template-file main.bicep \
  --parameters aioInstanceName=$AIO_INSTANCE_NAME customLocationName=$CUSTOM_LOCATION_NAME

Speichere dieses vollständige Graph-YAML als rate-limit-telemetry.yaml und wende es zum Debuggen mit kubectl apply -f rate-limit-telemetry.yaml an.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: rate-limit-telemetry
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  nodes:
    - name: source
      nodeType: Source
      sourceSettings:
        endpointRef: factory-mqtt
        dataSources: [factory/readings]
    - name: throttle
      nodeType: Graph
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-throttle:1.0.0
        configuration:
          - key: throttle
            value: |
              {
                "perTopicThrottles": [
                  {"topic":"factory/readings","maxMessagesPerSecond":2}
                ]
              }
    - name: destination
      nodeType: Destination
      destinationSettings:
        endpointRef: factory-mqtt
        dataDestination: factory/rate-limited
  nodeConnections:
    - from: { name: source }
      to: { name: throttle }
    - from: { name: throttle }
      to: { name: destination }