Azure IoT Operations でデータフローグラフを作成する

データフローグラフは、ソースと変換先の間を移動するときにデータを変換する、構成可能な処理パイプラインです。標準データフローは、固定エンリッチ、フィルター、マップシーケンスに従います。データフローグラフを使用すると、変換を任意の順序でチェーンし、並列パスに分岐し、時間枠の経過と同時にデータを集計できます。

この記事では、データフローグラフを段階的に作成する手順について説明します。データフローグラフと使用可能な変換の概要については、「データフローグラフの概要」を参照してください。

データフローグラフでは現在、MQTT、Kafka、OpenTelemetry エンドポイントのみがサポートされています。 Data Lake、Microsoft Fabric OneLake、Azure Data Explorer、ローカルストレージなど、他のエンドポイントタイプはサポートしていません。

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

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

前提条件

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

開発用コンピューターにインストールされているAzure CLI。利用可能な Azure CLI 拡張機能で、azure-iot-ops 拡張機能を使用するために必要な最小バージョンを確認してください。 az --version を使用してバージョンを確認し、必要に応じて az upgrade を使用して更新します。詳細については、「 Azure CLI のインストール」を参照してください。
Azure CLIのAzure IoT Operations拡張機能。次のコマンドを使用して、拡張機能を追加するか、それを最新バージョンに更新します。
```
az extension add --upgrade --name azure-iot-ops
```

Azure IoT Operations バージョン 1.2 以降
データフロープロファイル。既定のプロファイルを使用できます。
ソースと宛先のデータフローエンドポイント。既定の MQTT ブローカーエンドポイントは、作業を開始するために機能します。

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

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

環境変数	説明
`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>"

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

本記事では、データフローグラフとプロファイル名に環境変数 GRAPH_NAME と PROFILE も使用しています。関連するコマンドを実行する前にそれぞれ設定してください。

データフローグラフを作成する

データフローグラフには、データを取り込む ソース 、データを処理する変換、送信先の 3 種類の要素が含まれています。データをフローさせる順序で接続します。

運用体験では、Azure IoT Operationsインスタンスに移動します。
データフローグラフ>作成データフローグラフを選択します。
データフローグラフの名前を入力し、データフロープロファイルを選択します。既定では、既定のプロファイルが選択されています。
キャンバスに要素を追加してパイプラインを構築します。
1. ソースの追加: ソース エンドポイントを選択し、受信メッセージをサブスクライブするトピックを構成します。
2. 変換の追加: 1 つ以上の変換を選択してデータを処理します。利用可能な変換には、マップ、フィルター、ブランチ、コンカテネート、ウィンドウ、スロットルなどがあります。各変換の種類の詳細については、「データフローグラフの概要」を参照してください。
3. 宛先を追加する: 宛先エンドポイントを選択し、処理されたデータを送信するトピックまたはパスを構成します。
データをフローさせる順序で要素を接続します。
[保存] を選択してデータフローグラフをデプロイします。

Azure CLIは、すべてのノードと接続を含む単一のJSON設定ファイルからのデータフローグラフを使用します。 az iot ops dataflowgraph applyを使用してグラフを作成または置換します。以下の例は温度データを読み取り、それを華氏(F)に変換して宛先トピックに送信します。

データフローグラフのプロパティを使用して graph.json ファイルを作成します。 graph.jsonファイルでは、各変換のルールをvalueフィールドにエスケープされたJSON文字列として保存します。各変換のルールの読み取り可能な形式については、その変換の種類のハウツーを参照してください。

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "convert",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/converted"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "convert"
      }
    },
    {
      "from": {
        "name": "convert"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

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

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

このグラフは default データフロープロファイルを使用しています。別のプロファイルを使用するには、 --profile <PROFILE_NAME>を追加します。

次の構造の Bicep .bicep ファイルを作成します。次の使用例は、温度データを読み取り、華氏に変換して宛先トピックに送信するデータフローグラフを作成します。

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'

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

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: 'temperature-processing'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    profileRef: 'default'
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'telemetry/temperature'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'convert'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/converted'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'sensors' }
        to: { name: 'convert' }
      }
      {
        from: { name: 'convert' }
        to: { name: 'output' }
      }
    ]
  }
}

Bicep ファイルをデプロイします。

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep

Important

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

次の構造の Kubernetes マニフェスト .yaml ファイルを作成します。次の使用例は、温度データを読み取り、華氏に変換して宛先トピックに送信するデータフローグラフを作成します。

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

    - nodeType: Graph
      name: convert
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "temperature_f",
                    "expression": "cToF($1)"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/converted

  nodeConnections:
    - from:
        name: sensors
      to:
        name: convert
    - from:
        name: convert
      to:
        name: output

マニフェストを適用します。

kubectl apply -f main.yaml

ソースを構成する

ソースは、データがパイプラインに入る場所を定義します。エンドポイント参照と 1 つ以上のトピックを指定します。

データフローグラフエディターで、ソース要素を選択し、次の構成を行います。

Setting	説明
エンドポイント	使用するデータフローエンドポイント。ローカル MQTT ブローカーの既定値を選択します。
トピック	受信メッセージのために 1 つまたは複数のトピックを登録する。

CLI はグラフ全体を一度に適用するため、ソースを Source 構成ファイルのgraph.json ノードとして構成してから、az iot ops dataflowgraph apply実行します。

{
  "nodeType": "Source",
  "name": "sensors",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "telemetry/temperature",
      "telemetry/humidity"
    ]
  }
}

{
  nodeType: 'Source'
  name: 'sensors'
  sourceSettings: {
    endpointRef: 'default'
    dataSources: [
      'telemetry/temperature'
      'telemetry/humidity'
    ]
  }
}

Important

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

- nodeType: Source
  name: sensors
  sourceSettings:
    endpointRef: default
    dataSources:
      - telemetry/temperature
      - telemetry/humidity

変換を追加する

ソースと変換先の間でプロセスデータを変換します。各変換は組み込みのアーティファクトを参照し、構成ルールを持っています。

使用可能な組み込み変換は次のとおりです。

変換	アーティファクト	説明
地図	`azureiotoperations/graph-dataflow-map:1.0.0`	フィールドの名前変更、再構築、計算、およびコピーを行います。
フィルター	`azureiotoperations/graph-dataflow-filter:1.0.0`	条件に一致するメッセージを削除します。
ブランチ	`azureiotoperations/graph-dataflow-branch:1.0.0`	条件に基づいて、各メッセージを `true` または `false` パスにルーティングします。
連結	`azureiotoperations/graph-dataflow-concatenate:1.0.0`	2つ以上のパスを再び1つにまとめます。
ウィンドウ	`azureiotoperations/graph-dataflow-window:1.0.0`	一定期間にわたってメッセージを収集し、集計します。
スロットル	`azureiotoperations/graph-dataflow-throttle:1.0.0`	MQTTトピックパターンごとのメッセージレートを制限してください。

外部データでメッセージをエンリッチする方法の詳細については、「外部データを使用したエンリッチメント」を参照してください。

データフローグラフエディターで、[変換の 追加] を選択し、変換の種類を選択します。ビジュアルエディターでルールを構成します。

各 transform は、nodeType 設定ファイルで Graph が graph.json に設定されたノードです。変換のルールは、温度を華氏に変換する次のマップのような JSON オブジェクトです。

{
  "map": [
    {
      "inputs": ["temperature"],
      "output": "temperature_f",
      "expression": "cToF($1)"
    }
  ]
}

configurationプロパティはこれらのルールを文字列として受け取るので、ルールJSONを脱出してvalueフィールドに入れてください。 az iot ops dataflowgraph applyを使用してグラフ全体を適用します。

{
  "nodeType": "Graph",
  "name": "convert",
  "graphSettings": {
    "registryEndpointRef": "default",
    "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
    "configuration": [
      {
        "key": "rules",
        "value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
      }
    ]
  }
}

Tip

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

各変換は、 nodeType: 'Graph'を持つノードです。 configuration プロパティは、JSON 文字列として規則を渡します。

{
  nodeType: 'Graph'
  name: 'convert'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
    configuration: [
      {
        key: 'rules'
        value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
      }
    ]
  }
}

Important

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

各変換は、 nodeType: Graphを持つノードです。 configuration プロパティは、JSON 文字列として規則を渡します。

- nodeType: Graph
  name: convert
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-map:1.0.0
    configuration:
      - key: rules
        value: |
          {
            "map": [
              {
                "inputs": ["temperature"],
                "output": "temperature_f",
                "expression": "cToF($1)"
              }
            ]
          }

複数の変換を連結する

任意の数の変換を連鎖します。データの流れを指定した順番で、nodeConnections セクションで接続してください。

キャンバス上の変換間の接続をドラッグして、処理順序を定義します。

nodeConnections構成ファイルの graph.json セクションで処理順序を定義します。

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "remove-bad-data"
    }
  },
  {
    "from": {
      "name": "remove-bad-data"
    },
    "to": {
      "name": "convert"
    }
  },
  {
    "from": {
      "name": "convert"
    },
    "to": {
      "name": "output"
    }
  }
]

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
  { from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
  { from: { name: 'convert' }, to: { name: 'output' } }
]

Important

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

nodeConnections:
  - from: { name: sensors }
    to: { name: remove-bad-data }
  - from: { name: remove-bad-data }
    to: { name: convert }
  - from: { name: convert }
    to: { name: output }

宛先を構成する

宛先は処理されたデータがどこに行くかを定義します。エンドポイント参照とトピックまたはパスを指定します。

変換先の要素を選択し、次の構成を行います。

Setting	説明
エンドポイント	データを送信するデータフローエンドポイント。
Topic	処理されたデータを発行するトピックまたはパス。

Destination構成ファイルのgraph.json ノードとして宛先を構成し、az iot ops dataflowgraph applyを使用して完全なグラフを適用します。

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "telemetry/processed"
  }
}

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'telemetry/processed'
  }
}

Important

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

- nodeType: Destination
  name: output
  destinationSettings:
    endpointRef: default
    dataDestination: telemetry/processed

メッセージの内容に基づく動的なトピックルーティングについては、「さまざまなトピックにメッセージをルーティングする」を参照してください。

データフローグラフが動作していることを確認する

データフローグラフをデプロイした後、実行されていることを確認します。

運用体験では、データフローグラフを選択してその状態を確認します。正常なグラフには 、実行中 の状態が表示されます。

グラフの詳細を表示するには、 az iot ops dataflowgraph show を使用します。

az iot ops dataflowgraph show \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

プロファイルに関連付けられているすべてのデータフローグラフを一覧表示するには、次の az iot ops dataflowgraph listを使用します。

az iot ops dataflowgraph list \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

DataflowGraph リソースの状態を確認します。

az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE

Important

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

kubectl get dataflowgraph temperature-processing -n azure-iot-operations

ポッドログでエラーがないか確認します。

kubectl logs -l app=dataflow -n azure-iot-operations --tail=50

フィードバック

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

Last updated on 2026-07-31

Azure IoT Operations でデータ フロー グラフを作成する

前提条件

データ フロー グラフを作成する

ソースを構成する

変換を追加する

複数の変換を連結する

宛先を構成する

データ フロー グラフが動作していることを確認する

関連するコンテンツ

フィードバック

その他のリソース

Azure IoT Operations でデータフローグラフを作成する

データフローグラフを作成する

データフローグラフが動作していることを確認する