データフローグラフ内の異なるMQTTトピックへのメッセージのルーティング

一部のシナリオでは、メッセージの内容に応じて、さまざまな MQTT トピックにメッセージが届く必要があります。たとえば、重大なしきい値を超えるセンサーの読み取り値は、 alerts トピックに移動する必要がある一方で、通常の読み取り値は historian トピックに移動します。データフローグラフを使用すると、データフローに 1 つの宛先がある場合でも、出力トピックを動的に設定できます。

動的トピックルーティングは map変換に基づいて構築された技術です。mapルールがターゲットトピックをメッセージメタデータに書き込み、宛先がそのトピックに公開します。代わりに、グラフ内の異なる処理経路にメッセージをルーティングするには、「フィルタ、ブランチ、マージデータ」を参照してください。

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

前提条件

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

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

この記事のAzure CLI例は環境変数を使っており、各値を一度設定してからコマンドをコピー&ペーストできます as-is。クイックスタートのAzure IoT Operations Codespaces環境を使っている場合、これらの変数はすでに設定済みなので、このステップを省略できます。そうでなければ、コマンドを実行する前にシェル内で以下の環境変数を設定してください。

以下のスクリプトは、最も一般的に使われる環境変数を設定します。

環境変数	Description
`SUBSCRIPTION_ID`	Azure IoT Operationsインスタンスを含むサブスクリプションのIDです。
`RESOURCE_GROUP`	あなたのAzure IoT Operationsインスタンスを含むリソースグループの名前です。
`AIO_INSTANCE_NAME`	あなたのAzure IoT Operationsインスタンスの名前です。インスタンスを挙げるには、 `az iot ops list -o table`を実行してください。
`CLUSTER_NAME`	あなたのインスタンスをホストしているAzure Arc対応のKubernetesクラスターの名前です。
`LOCATION`	例えば`eastus`など、新しい資源を活用するためのAzure地域。

Bash
PowerShell

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

この記事で使っている変数を設定するだけで十分です。この記事では、選択したリソース名に追加の環境変数を使う場合があります。記事では、導入された場所の位置をどう設定するか説明しています。

動的トピックルーティングの仕組み

マップ変換では、 $metadata.topic 出力パスを使用して、MQTT トピックを含むメッセージメタデータに書き込むことができます。その後、変換先では ${outputTopic} 変数を使用して、変換で設定された任意のトピックに発行します。

次の 2 つの要素が連携して動作します。

変換内: マップルールは文字列値を $metadata.topicに書き込みます。
変換先: dataDestination フィールドは ${outputTopic}を参照し、これは変換によって書き込まれた値に解決されます。

変換は式言語を用いて値、テスト条件、参照フィールドを計算します。式は名前ではなく位置で入力を参照します。 inputs リストの最初の入力は $1、2番目は $2、という具合です。 cToFのような組み込み関数がこれらの値を変換・操作します。

演算子、関数、データ型、メタデータフィールドの完全なリストについては、Expressionsリファレンスを参照してください。

この記事はメッセージのメタデータに書き込みます。読み書きできるメタデータパスについては、メタデータフィールドを参照してください。

オプション1:単一のマップ変換と条件付き式によるルート

最も簡単な方法では、トピックを選択する if 式で 1 つのマップ変換を使用します。

操作エクスペリエンスで、データフローグラフを作成します。

から読み取るsensors/temperatureを追加します。
次の 2 つのルールを使用して マップ 変換を追加します。
- ワイルドカードパススルー規則 (入力 *、出力 *)。
- 入力 temperature、出力 $metadata.topic、および式 if($1 > 1000, "alerts", "historian")を含むコンピューティングルール。
目的地をで追加します。

マップ変換が"alerts"を$metadata.topicに書き込むと、出力先はfactory/${outputTopic}をfactory/alertsに解決します。

Azure CLIは単一のJSON設定ファイルからのデータフローグラフを使用します。グラフプロパティを使用して graph.json ファイルを作成します。 graph.jsonファイルでは、各変換がvalueフィールドにエスケープされたJSON文字列としてルールを格納します。各変換のルールの読みやすい形については、その変換タイプのハウツー記事を参照してください。

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensors/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "route-by-temperature",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"description\":\"Set topic based on temperature threshold\",\"inputs\":[\"temperature\"],\"output\":\"$metadata.topic\",\"expression\":\"if($1 > 1000, \\\"alerts\\\", \\\"historian\\\")\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "factory/${outputTopic}"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "route-by-temperature"
      }
    },
    {
      "from": {
        "name": "route-by-temperature"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Tip

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

構成ファイルを適用します。

az iot ops dataflowgraph apply \
  --name dynamic-topic-routing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'dynamic-topic-routing'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'sensors/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'route-by-temperature'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"description":"Set topic based on temperature threshold","inputs":["temperature"],"output":"$metadata.topic","expression":"if($1 > 1000, \\"alerts\\", \\"historian\\")"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'factory/${outputTopic}'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'route-by-temperature' } }
      { from: { name: 'route-by-temperature' }, to: { name: 'output' } }
    ]
  }
}

Important

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

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: dynamic-topic-routing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensors/temperature

    - nodeType: Graph
      name: route-by-temperature
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "description": "Set topic based on temperature threshold",
                    "inputs": ["temperature"],
                    "output": "$metadata.topic",
                    "expression": "if($1 > 1000, \"alerts\", \"historian\")"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: "factory/${outputTopic}"

  nodeConnections:
    - from: { name: sensors }
      to: { name: route-by-temperature }
    - from: { name: route-by-temperature }
      to: { name: output }

オプション2:分岐、パスごとのマップ、マージを含むルート

(単に異なるトピックではなく) パスごとに異なる変換が必要な場合は、分岐変換を使用してフローを分割し、各アームでマップ変換を使用してトピックを設定し、パス固有のルールを適用し、パスをマージするための連結変換を使用します。

運用体験では、次の操作を行います。

から読み取るsensors/temperatureを追加します。
フィールドに条件$1 > 1000を持つtemperature変換を追加します。
true パスに、ワイルドカードパススルーと、を$metadata.topicに設定するルールを含む"alerts"変換を追加します。
false パスに、ワイルドカードパススルーと、を$metadata.topicに設定するルールを含む"historian"変換を追加します。
両方のパスをマージする連結変換を追加します。
目的地をで追加します。

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensors/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "check-temperature",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-branch:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"branch\":{\"inputs\":[\"temperature\"],\"expression\":\"$1 > 1000\",\"description\":\"Route critical temperatures to alerts\"}}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "set-alerts-topic",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[],\"output\":\"$metadata.topic\",\"expression\":\"\\\"alerts\\\"\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "set-historian-topic",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[],\"output\":\"$metadata.topic\",\"expression\":\"\\\"historian\\\"\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "merge",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-concatenate:1.0.0"
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "factory/${outputTopic}"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "check-temperature"
      }
    },
    {
      "from": {
        "name": "check-temperature.output.true"
      },
      "to": {
        "name": "set-alerts-topic"
      }
    },
    {
      "from": {
        "name": "check-temperature.output.false"
      },
      "to": {
        "name": "set-historian-topic"
      }
    },
    {
      "from": {
        "name": "set-alerts-topic"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "set-historian-topic"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "merge"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

構成ファイルを適用します。

az iot ops dataflowgraph apply \
  --name dynamic-topic-routing-branched \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'dynamic-topic-routing-branched'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'sensors/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'check-temperature'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-branch:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"branch":{"inputs":["temperature"],"expression":"$1 > 1000","description":"Route critical temperatures to alerts"}}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'set-alerts-topic'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":[],"output":"$metadata.topic","expression":"\\"alerts\\""}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'set-historian-topic'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":[],"output":"$metadata.topic","expression":"\\"historian\\""}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'merge'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'factory/${outputTopic}'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'check-temperature' } }
      { from: { name: 'check-temperature.output.true' }, to: { name: 'set-alerts-topic' } }
      { from: { name: 'check-temperature.output.false' }, to: { name: 'set-historian-topic' } }
      { from: { name: 'set-alerts-topic' }, to: { name: 'merge' } }
      { from: { name: 'set-historian-topic' }, to: { name: 'merge' } }
      { from: { name: 'merge' }, to: { name: 'output' } }
    ]
  }
}

Important

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

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: dynamic-topic-routing-branched
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensors/temperature

    - nodeType: Graph
      name: check-temperature
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-branch:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "branch": {
                  "inputs": ["temperature"],
                  "expression": "$1 > 1000",
                  "description": "Route critical temperatures to alerts"
                }
              }

    - nodeType: Graph
      name: set-alerts-topic
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["*"], "output": "*" },
                  { "inputs": [], "output": "$metadata.topic", "expression": "\"alerts\"" }
                ]
              }

    - nodeType: Graph
      name: set-historian-topic
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["*"], "output": "*" },
                  { "inputs": [], "output": "$metadata.topic", "expression": "\"historian\"" }
                ]
              }

    - nodeType: Graph
      name: merge
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-concatenate:1.0.0

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: "factory/${outputTopic}"

  nodeConnections:
    - from: { name: sensors }
      to: { name: check-temperature }
    - from: { name: check-temperature.output.true }
      to: { name: set-alerts-topic }
    - from: { name: check-temperature.output.false }
      to: { name: set-historian-topic }
    - from: { name: set-alerts-topic }
      to: { name: merge }
    - from: { name: set-historian-topic }
      to: { name: merge }
    - from: { name: merge }
      to: { name: output }

単一のマップ変換と分岐のどちらかを選びます

考慮事項	オプション 1 (単一マップ)	オプション 2 (ブランチ + マップ)
簡略	ノードの数が少なく、読み取りが簡単	より多くのノード、より明確に
トピックのみのルーティング	理想	設定には余計な手間がかかりますが、動作します。
パスごとに異なる変換	入れ子になった `if()`で可能、複雑になる	自然: 各分岐には独自のマップルールがあります
パスの追加	チェーン `if()` 呼び出し	入れ子の分岐が必要です

単一の条件に基づく簡単なトピックルーティングの場合、オプション 1 の方が簡単です。各パスにトピック名以外の異なる処理が必要な場合は、オプション 2 を使用します。

outputTopic変数が宛先トピックをどのように解決するか

${outputTopic}のdataDestination変数は、パイプラインの最後の変換によって設定された$metadata.topicの完全な値に解決されます。 ${outputTopic.N} (1 インデックス付き) でセグメントを使用することもできます。たとえば、変換で $metadata.topic が "region/west"に設定されている場合は、次のようになります。

`dataDestination`	解決済みトピック
`factory/${outputTopic}`	`factory/region/west`
`factory/${outputTopic.1}`	`factory/region`
`factory/${outputTopic.2}`	`factory/west`

データフローがトピック変数を解決できない場合(例えば、 $metadata.topic が一度も設定されていない場合)、メッセージをドロップしエラーをログします。

スロットルデータ

フィードバック

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

Last updated on 2026-08-14