データ フロー グラフで WebAssembly (WASM) を使用する

1. 操作エクスペリエンスでデータ フロー グラフを作成するには、[データ フロー] タブに移動します。

2. [+ 作成] の横にあるドロップダウン メニューを選択し、[データ フロー グラフの作成] を選択します

   

3. プレースホルダー名 new-data-flow を選択して、データ フローのプロパティを設定します。 データ フロー グラフの名前を入力し、使用するデータ フロー プロファイルを選択します。

4. データ フロー図で、[ ソース ] を選択してソース ノードを構成します。 [ ソースの詳細] で、[ 資産 ] または [ データ フロー エンドポイント] を選択します。

   

   1. [資産] を選択した場合は、データをプルする資産を選択し、[適用] をクリックします。

   2. [データ フロー エンドポイント] を選択した場合は、次の詳細を入力し、[適用] をクリックします。

      Setting	Description
      データ フロー エンドポイント	既定の MQTT メッセージ ブローカー エンドポイントを使用するには、"既定値" を選択します。
      トピック	着信メッセージをサブスクライブするためのトピック フィルター。 [トピック]>[行の追加] を使用して、複数のトピックを追加します。
      メッセージ スキーマ	受信メッセージの逆シリアル化に使用するスキーマ。

5. データ フロー ダイアグラムで、[ グラフ変換の追加 ] (省略可能) を選択して、グラフ処理ノードを追加します。 [グラフの選択] ウィンドウで、グラフの単純な 1 を選択し、[適用] をクリックします。

   

6. ダイアグラム内のグラフ ノードを選択することで、グラフ演算子の設定を構成できます。 たとえば、モジュール温度/マップ演算子を選択し、key2値example-value-2入力できます。 [ 適用 ] をクリックして変更を保存します。

   

7. データ フローダイアグラムで、[ 宛先 ] を選択して宛先ノードを構成します。

8. データ フロー グラフ名の下にある [保存] を選択して、データ フロー グラフを保存します。

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'

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

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'temperature-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'temperature-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'graph-simple:1.0.0'
          configuration: [
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'temperature-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'sensor/temperature/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'temperature-source'
        }
        to: {
          name: 'temperature-processor'
        }
      }
      {
        from: {
          name: 'temperature-processor'
        }
        to: {
          name: 'temperature-destination'
        }
      }
    ]
  }
}

次の構成で YAML ファイルを作成します。

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: temperature-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw

    - nodeType: Graph
      name: temperature-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: graph-simple:1.0.0
        configuration:
          - key: key1
            value: example-value

    - nodeType: Destination
      name: temperature-destination
      destinationSettings:
        endpointRef: default
        dataDestination: sensor/temperature/processed

  nodeConnections:
    - from:
        name: temperature-source
      to:
        name: temperature-processor

    - from:
        name: temperature-processor
      to:
        name: temperature-destination

構成を dataflow-graph.yaml として保存し、クラスターに適用します。

kubectl apply -f dataflow-graph.yaml

1. 操作エクスペリエンスでデータ フロー グラフを作成するには、[データ フロー] タブに移動します。

2. [+ 作成] の横にあるドロップダウン メニューを選択し、[データ フロー グラフの作成] を選択します

   

3. プレースホルダー名 new-data-flow を選択して、データ フローのプロパティを設定します。 データ フロー グラフの名前を入力し、使用するデータ フロー プロファイルを選択します。

4. データ フロー図で、[ ソース ] を選択してソース ノードを構成します。 [ ソースの詳細] で、[ 資産 ] または [ データ フロー エンドポイント] を選択します。

   

   1. [資産] を選択した場合は、データをプルする資産を選択し、[適用] をクリックします。

   2. [データ フロー エンドポイント] を選択した場合は、次の詳細を入力し、[適用] をクリックします。

      Setting	Description
      データ フロー エンドポイント	既定の MQTT メッセージ ブローカー エンドポイントを使用するには、"既定値" を選択します。
      トピック	着信メッセージをサブスクライブするためのトピック フィルター。 [トピック]>[行の追加] を使用して、複数のトピックを追加します。
      メッセージ スキーマ	受信メッセージの逆シリアル化に使用するスキーマ。

5. データ フロー ダイアグラムで、[ グラフ変換の追加 ] (省略可能) を選択して、グラフ処理ノードを追加します。 [グラフの選択] ウィンドウで、graph-complex:1 を選択し、[適用] をクリックします。

   

6. ダイアグラム内のグラフ ノードを選択することで、グラフ演算子の設定を構成できます。

   

   Operator	Description
   module-snapshot/branch	画像に対してオブジェクト検出を実行するように snapshot モジュールを構成します。 snapshot_topic構成キーを設定して、画像データの入力トピックを指定できます。
   モジュール-温度/マップ	key2温度値を別のスケールに変換します。

7. [ 適用 ] をクリックして変更を保存します。

8. データ フローダイアグラムで、[ 宛先 ] を選択して宛先ノードを構成します。

9. データ フロー グラフ名の下にある [保存] を選択して、データ フロー グラフを保存します。

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<COMPLEX_GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'

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

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

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

resource complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensor-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
            'sensor/humidity/raw'
            'sensor/images/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'sensor-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'graph-complex:1.0.0'
          configuration: [
            {
              key: 'snapshot_topic'
              value: 'sensor/images/raw'
            }
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'analytics-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'analytics/sensor/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'sensor-source'
        }
        to: {
          name: 'sensor-processor'
        }
      }
      {
        from: {
          name: 'sensor-processor'
        }
        to: {
          name: 'analytics-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <COMPLEX_GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: sensor-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw
          - sensor/humidity/raw
          - sensor/images/raw

    - nodeType: Graph
      name: sensor-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: graph-complex:1.0.0
        configuration:
          - key: snapshot_topic
            value: sensor/images/raw
          - key: key1
            value: example-value

    - nodeType: Destination
      name: analytics-destination
      destinationSettings:
        endpointRef: default
        dataDestination: analytics/sensor/processed

  nodeConnections:
    - from:
        name: sensor-source
      to:
        name: sensor-processor

    - from:
        name: sensor-processor
      to:
        name: analytics-destination

データ フロー グラフを作成または編集するときに、[ データ フローのプロパティ ] ウィンドウで [ データ フローを有効にする] を [はい ] に設定して、モードを Enabledに設定できます。 オフのままにすると、モードは Disabledに設定されます。

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  // ... other properties
  properties: {
    mode: 'Enabled'  // or 'Disabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  mode: Enabled  # or Disabled
  # ... other properties

データ フロー グラフを作成または編集するときに、[データ フローのプロパティ ] ウィンドウで、データ フロー プロファイルを選択します。 既定では、既定のデータ フロー プロファイルが選択されています。 データ フロー プロファイルの詳細については、「データ フロー プロファイルの構成」を参照してください。

Bicep では、データ フロー グラフをプロファイルの子リソースとして作成してプロファイルを指定します。

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  parent: defaultDataflowProfile  // This establishes the profile relationship
  // ... other properties
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  profileRef: default  # Reference to the profile name
  # ... other properties

データ フロー グラフを作成または編集する場合、[ データ フローのプロパティ ] ウィンドウで 、[ データの永続化の要求 ] を [ はい ] にチェックして、要求ディスクの永続化を Enabledに設定できます。 未チェックのままにすると、設定は Disabled。

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  // ... other properties
  properties: {
    requestDiskPersistence: 'Enabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Enabled
  # ... other properties

データ フロー図で、[ ソース ] を選択してソース ノードを構成します。 [ ソースの詳細] で [ データ フロー エンドポイント] を選択し、[ トピック] フィールドを使用して、受信メッセージをサブスクライブする MQTT トピック フィルターを指定します。 [行の追加] を選択し、新しいトピックを入力することで、複数の MQTT トピックを追加できます。

{
  nodeType: 'Source'
  name: 'sensor-data-source'
  sourceSettings: {
    endpointRef: 'default'  // References a data flow endpoint
    dataSources: [
      'sensor/+/temperature'  // MQTT topic with wildcard
      'sensor/+/humidity'
    ]
    assetRef: 'weather-station-asset'  // Optional asset reference
  }
}

- nodeType: Source
  name: sensor-data-source
  sourceSettings:
    endpointRef: default  # References a data flow endpoint
    dataSources:
      - sensor/+/temperature  # MQTT topic with wildcard
      - sensor/+/humidity
    assetRef: weather-station-asset  # Optional asset reference

データ フロー ダイアグラムで、[ グラフ変換の追加 ] (省略可能) を選択して、グラフ処理ノードを追加します。 [グラフの選択] ウィンドウで、目的のグラフ成果物 (単純グラフまたは複雑グラフ) を選択し、[適用] をクリックします。 ダイアグラム内のグラフ ノードを選択することで、グラフ演算子の設定を構成できます。

{
  nodeType: 'Graph'
  name: 'temperature-processor'
  graphSettings: {
    registryEndpointRef: 'my-acr-endpoint'
    artifact: 'temperature-converter:2.1.0'
    configuration: [
      {
        key: 'input-unit'
        value: 'fahrenheit'
      }
      {
        key: 'output-unit' 
        value: 'celsius'
      }
      {
        key: 'precision'
        value: '2'
      }
      {
        key: 'enable-filtering'
        value: 'true'
      }
    ]
  }
}

- nodeType: Graph
  name: temperature-processor
  graphSettings:
    registryEndpointRef: my-acr-endpoint
    artifact: temperature-converter:2.1.0
    configuration:
      - key: input-unit
        value: fahrenheit
      - key: output-unit
        value: celsius
      - key: precision
        value: "2"
      - key: enable-filtering
        value: "true"

1. データ フロー図で、[ 宛先 ] ノードを選択します。
2. [データ フロー エンドポイントの詳細] ドロップダウンから目的の データ フロー エンドポイントを 選択します。
3. [続行] を選択して、宛先を構成します。
4. データの送信先となるトピックやテーブルなど、宛先に 必要な設定 を入力します。 データ変換先フィールドは、エンドポイントの種類に基づいて自動的に解釈されます。 たとえば、データ フロー エンドポイントが MQTT エンドポイントの場合、宛先の詳細ページでトピックを入力するように求められます。

{
  nodeType: 'Destination'
  name: 'cloud-storage-destination'
  destinationSettings: {
    endpointRef: 'azure-storage-endpoint'
    dataDestination: 'processed-data/temperature'
    outputSchemaSettings: {
      serializationFormat: 'Parquet'
      schemaRef: 'temperature-output-schema:1'
    }
  }
}

- nodeType: Destination
  name: cloud-storage-destination
  destinationSettings:
    endpointRef: azure-storage-endpoint
    dataDestination: processed-data/temperature
    outputSchemaSettings:
      serializationFormat: Parquet
      schemaRef: temperature-output-schema:1

グラフ処理ノードを選択すると、操作エクスペリエンスによってノード接続が自動的に作成されます。 グラフの作成後に接続を変更することはできません。

nodeConnections: [
  {
    from: {
      name: 'sensor-data-source'
      schema: {
        schemaRef: 'sensor-input-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'temperature-processor'
    }
  }
  {
    from: {
      name: 'temperature-processor'
    }
    to: {
      name: 'cloud-storage-destination'
    }
  }
]

nodeConnections:
  - from:
      name: sensor-data-source
      schema:
        schemaRef: sensor-input-schema:1
        serializationFormat: Json
    to:
      name: temperature-processor
  - from:
      name: temperature-processor
    to:
      name: cloud-storage-destination