データフローグラフにおけるスロットルデータ

スロットル変換はMQTTトピックでメッセージを転送する頻度を制限します。内容に基づいてメッセージをドロップするのではなく、スロットル変換は処理タイミングに基づいてメッセージをドロップし、各設定された区間内のトピックパターンごとに最大1つのメッセージを転送します。スロットリングを使って、メッセージ内容を変えずに下流システムをバーストや高周波のソースから保護しましょう。

データフローグラフの概要と、パイプラインでの変換の構成方法については、データフローグラフの概要に関するページを参照してください。

Prerequisites

Kubernetes クラスターにデプロイされた Azure IoT Operations のインスタンス。詳細については、「Deploy Azure IoT Operations」をご覧ください。

デプロイメントは自動的にmcr.microsoft.comを指すdefaultという名前のデフォルトレジストリエンドポイントを作成します。組み込みの変換では、このエンドポイントが使用されます。

ステートフルグラフのスケーリング制限

Important

ウィンドウ変換とスロットル変換は ステートフルです。各インスタンスは独自の状態を維持し、インスタンス同士がその状態を共有しません。データフロープロファイルのインスタンス数が1を超えると、共有サブスクリプションはインスタンス間でメッセージを分散するため、各インスタンスはメッセージの一部しか見られません。ウィンドウ変換は部分データセットに対して平均値、和値、カウントなどの集計を計算し、スロットル変換はパイプライン全体ではなく各インスタンスで独立して設定レート制限を強制します。

ウィンドウやスロットル変換を使用するデータフローグラフについては、データフロープロファイルのインスタンスカウントを 1 に設定します。マップ、フィルタ、分岐、連結変換のみを使用する状態なしデータフローグラフは、より多くのインスタンス数を安全に活用してスループットを増加させることができます。

スロットル変換の仕組み

スロットル変換は、各受信メッセージのトピックを、順序付けられたトピックごとのルールリストに照らして評価します。

最初に一致したエントリが選択されます。 変換は規則を順番に評価します。 topicパターンがメッセージのトピックと一致する最初のルールが、変換がメッセージを転送するかどうかを決定します。変身は後のルールをチェックしません。たとえそれらも一致していてもです。
一致しないトピックはそのまま通過します。 もしメッセージのトピックに一致するルールがなければ、トランスフォームはスロットリングなしでメッセージを転送します。
転送はカウントベースではなく時間ベースです。 一致したルールについては、トランスフォームは最初に処理するメッセージを転送し、その後は、最後に転送されたメッセージから設定された間隔（1000 / maxMessagesPerSecond ミリ秒、切り上げ）が経過するまで、そのパターンに一致する後続のメッセージをすべて破棄します。この挙動は最大レートを制限しますが、バーストで以前の落ちを補うことはできません。
パターンごとの共有状態であり、トピックごとの共有ではありません。 ルールの topic パターンがワイルドカードを使う場合、それに合うすべての具体的なトピックはトランスフォームインスタンス内で同じスロットル状態を共有します。例えば、単一の sensors/+ ルールが sensors/temperature と sensors/humidity の合計レートを制限するものであり、それぞれが独立して制限するわけではありません。
0 すべてを捨てる。 ルールの maxMessagesPerSecond を 0 に設定すると、そのルールのパターンに合致するすべてのメッセージが削除されます。
タイミングはメッセージの内容ではなく処理時間に基づいています。 変換は各メッセージを処理する際に単調クロックを使用します。メッセージのペイロードからタイムスタンプフィールドを読み取ることはありません。

Note

スロットル変換はメッセージを転送するかドロップするかを決めるだけです。メッセージの内容は一切変更されません。

トピックごとのスロットルルールを設定する

スロットルルールは throttle 設定キー( rulesではなく)でJSONオブジェクトと perTopicThrottles 配列として定義してください。

スロットル変換構成では、1つ以上のスロットルルールを追加します。ルールごとに、次の値を指定します。

Setting	説明
Topic	MQTTのトピックパターンもそれに合わせて。 `+`(シングルレベル)および`#`(マルチレベル)ワイルドカードをサポートしています。
最大1秒あたりのメッセージ数	このパターンに合致するトピックの最大転送速度。 `0`に設定して、一致するすべてのメッセージをドロップします。

CLI は、1 つの構成ファイルからグラフ全体を適用します。この設定をconfigurationファイルのスロットル変換ノードのgraph.jsonに追加し、az iot ops dataflowgraph applyを使って適用してください。

構成はJSONオブジェクトです:

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

この構成を value フィールドにエスケープ文字列として、 throttle キーの下に加えます:

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

ヒント

エスケープ文字列を生成するには、ルールを rules.jsonのようなファイルに保存し、 jq -c . rules.jsonを実行し、1行の出力を value フィールドに貼り付けます。

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

Important

Kubernetes 配置マニフェストの使用は運用環境ではサポートされていないため、デバッグとテストにのみ使用する必要があります。

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

このルールにより、トランスフォームは処理した最初の sensors/temperature メッセージを転送し、その後100ミリ秒(1000 / 10)以内に処理したそのトピックに関するメッセージを除外します。このルールは sensors/temperature以外の話題には影響しません。

perTopicThrottlesの各項目は以下の性質を持っています。

財産	Required	説明
`topic`	イエス	MQTTのトピックパターンもそれに合わせて。 `+`(シングルレベル)と`#`(マルチレベル、トレーリングのみ)ワイルドカードをサポートしています。
`maxMessagesPerSecond`	イエス	このパターンに合致するトピックの最大転送速度は、1秒あたりのメッセージ数です。変換により分数値も可能です。例えば、 `0.1` 10秒ごとに1つのメッセージを受信できます。ゼロか正の有限数でなければなりません。パターンに一致するすべてのメッセージを破棄するには、`0` に設定します。タイミング精度は1ミリ秒なので、 `1000` を超える値は `1000`と同じ実効限界を持ちます。

Important

topicの各perTopicThrottlesパターンは独自でなければなりません。変換は初期化時に同じパターン文字列を複数回設定することを拒否します。

トピックごとに複数のルールを使う

変換はルールを順番に評価し、最初のマッチが勝つため、独自のレート制限を持たせたいなら、より一般的なパターンより先に具体的なパターンをリストアップしてください。

次の順に2つのルールを追加します。

Order	トピック	最大1秒あたりのメッセージ数
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

Kubernetes 配置マニフェストの使用は運用環境ではサポートされていないため、デバッグとテストにのみ使用する必要があります。

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

sensors/temperature上のメッセージは最初のルールに一致し、1秒間に10メッセージに制限されています。他の sensors/* どのトピック(例えば sensors/humidity)のメッセージも2つ目のルールに合致し、合計で1秒あたり1つのメッセージの共有制限を共有します。

Important

順序が重要です。もし sensors/# ルールが先に記載されれば、メッセージ sensors/temperature 一致し、トランスフォームはより具体的なルールには到達しません。

ワイルドカードを使ってトピックのグループを制限する

topicパターンはMQTTトピックフィルターと同じワイルドカードをサポートしています:

ワイルドカード	一致	例
`+`	トピックレベルは1つのみ	`sensors/+/status` は `sensors/line1/status` と一致しますが、`sensors/line1/sub/status` とは一致しません
`#`	残りトピックレベルが0つ以上、そして最後のセグメントでなければなりません	`sensors/#` は `sensors`、`sensors/temperature`、および `sensors/line1/temperature` と一致します

同じワイルドカードルールに一致するすべての個々のトピックは、1 つのスロットル状態を共有します。

トピックの sensors/+ 制限と1秒あたりのメッセージ数の上限を 1ルールに追加してください。

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

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

Important

Kubernetes 配置マニフェストの使用は運用環境ではサポートされていないため、デバッグとテストにのみ使用する必要があります。

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

このルールでは、sensors/temperature 上のメッセージと sensors/humidity 上のメッセージが500ミリ秒間隔で処理された場合、両方が転送されることはありません。2つ目のメッセージは、どの具体的なトピック上のメッセージであるかに関係なく、最初のメッセージと同じ共有の1秒間隔にカウントされるため、破棄されます。

Note

裸 # パターンはすべてのトピックにマッチし、各変換インスタンス内に単一の結合レート制限を適用します。

トピックごとにメッセージをすべて落とす

maxMessagesPerSecondを0に設定し、パターンに一致するすべてのメッセージをドロップしますが、ルールやトピックはパイプラインから削除されません:

トピックの debug/# 制限と1秒あたりのメッセージ数の上限を 0ルールに追加してください。

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

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

Important

Kubernetes 配置マニフェストの使用は運用環境ではサポートされていないため、デバッグとテストにのみ使用する必要があります。

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

スロットル付きのデータフローグラフを展開する

スロットリングをエンドツーエンドで適用するには、ソース、スロットル変換、宛先をつなぐデータフローグラフを展開します。自分のワークフローに合ったツールを使いましょう。

Operationsの経験では、スロットル変換を含むデータフローグラフを作成します:

MQTT トピックから読み取る ソース を追加します。
スロットル トランスフォームを追加してください。トピックごとに1つ以上のルールを追加し、その順に、より具体的なものから少ない順に並べてください。
出力トピックに送信する宛先を追加します。

Azure CLIは単一のJSON設定ファイルからデータフローグラフを適用します。グラフプロパティを使用して graph.json ファイルを作成します。 graph.jsonファイルでは、各変換がvalueフィールドに設定をエスケープされたJSON文字列として保存します。スロットル構成の読みやすい形については、「トピックごとのスロットルルールの設定」を参照してください。 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

Kubernetes 配置マニフェストの使用は運用環境ではサポートされていないため、デバッグとテストにのみ使用する必要があります。

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 }

制限事項

メッセージを改変しません。 スロットル変換はメッセージの転送や削除のみを行います。メッセージの内容は変わりません。
最初に一致したエントリが選択されます。 変換は、 topic パターンがメッセージのトピックと一致する最初のルールのみを適用します。より一般的なパターンの前に、より具体的なパターンを挙げます。
パターンごとの共有状態。 ワイルドカードルールは、対応するすべての具体的なトピックでレート上限を共有します。トピックごとに別々の制限はありません。
バースト許容量はありません。 変換はマッチしたパターンごとに転送メッセージ間の最小時間を強制します。それ以前の、より遅い時期の未使用容量が蓄積されることもありません。
ミリ秒単位の精度です。 最小スロットル間隔は1ミリ秒なので、1000の値をmaxMessagesPerSecondより大きくしても、実効転送速度は1,000件/秒を超えません。
タイミングはメッセージの内容ではなく処理時間に基づいています。 変換は、ブローカーがメッセージを受信した時間やペイロード内のタイムスタンプフィールドではなく、各メッセージを処理する時間を利用します。
州は地域であり、記憶に残っています。 各データフロープロファイルインスタンスは独自のスロットル状態を維持します。トランスフォームを再起動したり設定したりすると、その状態がリセットされます。プロファイルに複数のインスタンスがある場合、各インスタンスが独立して設定されたレートを強制します。
トピックパターンの重複は禁止されています。 同じtopic文字列を複数回設定すると、変換が初期化される際に失敗しますperTopicThrottles

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-14

データフローグラフにおけるスロットルデータ

Prerequisites

ステートフル グラフのスケーリング制限

スロットル変換の仕組み

トピックごとのスロットルルールを設定する

トピックごとに複数のルールを使う

ワイルドカードを使ってトピックのグループを制限する

トピックごとにメッセージをすべて落とす

スロットル付きのデータフローグラフを展開する

制限事項

関連するコンテンツ

フィードバック

その他のリソース

ステートフルグラフのスケーリング制限