データフローグラフにおけるデータのフィルタリング、分岐、マージ

データフローグラフには、パイプラインを通過するメッセージフローを制御する 2 つの方法が用意されています。 フィルター 変換は不要なメッセージをドロップし、分岐変換は条件に基づいて各メッセージを 2 つのパスの 1 つ下にルーティングします。分岐後、連結変換によりパスが再統合されます。

これらの変換はグラフ内でメッセージをルーティングします。代わりに、内容に基づいて異なるMQTTトピックへのメッセージをルーティングするには、「異なるMQTTトピックへのメッセージのルーティング」をご覧ください。

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

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

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

前提条件

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

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

この記事の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>"

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

フィルター変換

フィルター変換は、受信した各メッセージを 1 つ以上のルールに対して評価し、メッセージがパイプラインを通過するか破棄されるかを決定します。

Important

フィルター式は、保持するメッセージではなく、削除するメッセージを選択します。式が真の場合はメッセージがドロップされます。この挙動は、保持される値を計算するマップ式の反対です。

条件に一致するメッセージを保持するには、式を反転させます。例えば、90以上に限定するには $1 <= 90でフィルタリングします。

フィルタールールのしくみ

各フィルター規則には、次のプロパティがあります。

財産	必須	説明
`inputs`	はい	受信メッセージから読み取るフィールドパスの一覧。
`expression`	はい	入力値に適用される数式。ブール値を返す必要があります。 trueが戻ると、メッセージは削除されます。
`description`	いいえ	エラーメッセージで使用される人間が判読できるラベル。

各入力は順序に基づいて位置変数に対応します。最初の入力は $1、2番目は $2、という具合です。

複数のルールを定義すると、 OR ロジックが使用されます。 いずれかの ルールが true と評価された場合、メッセージは削除されます。ルールが一致すると、エンジンはショートサーキットします。

主な制約:

式の入力が必要です。 すべてのフィルター規則に expressionを含める必要があります。
filter は配列を引数に取る。 単一のルールでも、 "filter": [ { ... } ]JSON配列としてルールを提供しましょう。裸のオブジェクトを渡すと変換が読み込まれず、エラーはルールのペイロードではなくアーティファクトやレジストリに指向します。この制約は、単一のオブジェクトを扱う branchとは異なります。
ワイルドカード入力なし。 各入力は、特定のフィールドパスを参照する必要があります。
フィールドが見つからないと、エラーが発生します。 inputsで参照されているフィールドが存在しない場合、フィルターはメッセージをサイレントで渡すのではなく、エラーを返します。
ブール値以外の結果ではエラーが発生します。 式がブール値以外の値 (文字列や数値など) を返す場合、フィルターはエラーを返します。

条件別にメッセージを削除する

温度が 100 を超えるメッセージを削除するには:

フィルター変換の構成で、次の規則を追加します。

Setting	価値
入力	`temperature`
式	`$1 > 100`

CLI は、1 つの構成ファイルからグラフ全体を適用します。このスニペットを graph.json の対応する場所に付けて、 az iot ops dataflowgraph applyを使って適用してください。

"filter": [
  {
    "inputs": [
      "temperature"
    ],
    "expression": "$1 > 100"
  }
]

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 > 100'
  }
]

Important

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

- inputs:
    - temperature     # $1
  expression: "$1 > 100"

温度が 100 以下のメッセージが通過します。 100 を超すメッセージは削除されます。

条件ごとにメッセージを保持してください

多くの場合、逆の結果を求めます:条件に合致するメッセージだけを残すこと。フィルター式が削除すべきものを選択するため、比較を逆にします。

90以上に抑えるには、90以下の数値をすべて下げてください:

フィルター変換の構成で、次の規則を追加します。

Setting	価値
入力	`temperature`
式	`$1 <= 90`
Description	`Drop readings at or below 90`

"filter": [
  {
    "inputs": [
      "temperature"
    ],
    "expression": "$1 <= 90",
    "description": "Drop readings at or below 90"
  }
]

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 <= 90'
    description: 'Drop readings at or below 90'
  }
]

Important

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

- inputs:
    - temperature     # $1
  expression: "$1 <= 90"
  description: "Drop readings at or below 90"

90以上のメッセージのみがパイプラインを通過します。ここに $1 > 90 書くと逆効果になります。90度以上の数値はすべて下がり、温度は低いままにしてしまいます。

ヒント

descriptionフィールドを使って、ルールが何を落とすかの意図を記録してください。 Drop readings at or below 90のような説明は正確を保ちますが、Keep hot readingsは逆さまの表現ミスを招き、エラーメッセージに現れて逆読みになります。

複数の条件を使用する

複数のルールを定義すると、次 のいずれかの ルールが一致すると、フィルターによってメッセージが削除されます。

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

入力	Expression	説明
`temperature`	`$1 > 100`	高温下げる
`humidity`	`$1 > 95`	高湿度を落とす

"filter": [
  {
    "inputs": [
      "temperature"
    ],
    "expression": "$1 > 100",
    "description": "Drop high temperature"
  },
  {
    "inputs": [
      "humidity"
    ],
    "expression": "$1 > 95",
    "description": "Drop high humidity"
  }
]

filter: [
  {
    inputs: [ 'temperature' ]
    expression: '$1 > 100'
    description: 'Drop high temperature'
  }
  {
    inputs: [ 'humidity' ]
    expression: '$1 > 95'
    description: 'Drop high humidity'
  }
]

Important

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

- inputs:
    - temperature     # $1
  expression: "$1 > 100"
  description: "Drop high temperature"

- inputs:
    - humidity     # $1
  expression: "$1 > 95"
  description: "Drop high humidity"

メッセージ	温度ルール	湿度ルール	結果
`{"temperature": 150, "humidity": 60}`	真実	false	Dropped
`{"temperature": 80, "humidity": 98}`	false	真実	Dropped
`{"temperature": 80, "humidity": 60}`	false	false	Passes

ヒント

フィールド間で AND ロジックが必要な場合は、1 つのルールで複数の入力を使用します。独立した条件間で OR ロジックが必要な場合は、複数のルールを使用します。

複雑な式の使用

1 つのルール内の複数のフィールドを参照し、論理演算子と組み合わせます。

入力 temperature と humidity、および式の $1 > 30 && $2 < 60を含むルールを追加します。

"filter": [
  {
    "inputs": [
      "temperature",
      "humidity"
    ],
    "expression": "$1 > 30 && $2 < 60",
    "description": "Drop hot and dry readings"
  }
]

filter: [
  {
    inputs: [ 'temperature', 'humidity' ]
    expression: '$1 > 30 && $2 < 60'
    description: 'Drop hot and dry readings'
  }
]

Important

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

- inputs:
    - temperature     # $1
    - humidity        # $2
  expression: "$1 > 30 && $2 < 60"
  description: "Drop hot and dry readings"

演算子と関数の完全な一覧については、「式のリファレンス」を参照してください。

スキーマに対してフィルターメッセージを検証する

フィルター変換を設定し、フィルタールールを実行する前にJSONスキーマに対して受信メッセージを検証します。このプロセスはスキーマに適合しないメッセージをドロップします。

スキーマ検証を有効にするには、フィルター構成で validateSchema を true に設定します。有効にすると、フィルターは受信ノード接続のschemaRef (フィルターノードにフィードするfromエントリのnodeConnections側) からスキーマを取得します。

フィルター変換の構成には、[ スキーマの検証 ] チェックボックスが含まれています。しかし、現在、Operationsの体験はノード接続上で schemaRef の設定や閲覧をサポートしていません。スキーマ検証を使用するには、Bicep マニフェストまたは Kubernetes マニフェストを使用して、ノード接続の schemaRef を構成します。

CLI は、1 つの構成ファイルからグラフ全体を適用するため、これを graph.json 内の対応する場所に追加し、 az iot ops dataflowgraph applyで適用します。 graph.json ファイルでは、各変換のルールがエスケープされた JSON 文字列として value フィールドに格納されます。各変換のルールの読み取り可能な形式については、その変換の種類のハウツーを参照してください。

"nodes": [
  {
    "nodeType": "Graph",
    "name": "schema-filter",
    "graphSettings": {
      "registryEndpointRef": "default",
      "artifact": "azureiotoperations/graph-dataflow-filter:1.0.0",
      "configuration": [
        {
          "key": "rules",
          "value": "{\"validateSchema\":true,\"filter\":[]}"
        }
      ]
    }
  }
],
"nodeConnections": [
  {
    "from": {
      "name": "sensors",
      "schema": {
        "schemaRef": "aio-sr://my-namespace/sensor-schema:1",
        "serializationFormat": "Json"
      }
    },
    "to": {
      "name": "schema-filter"
    }
  }
]

ヒント

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

フィルター規則 JSON に validateSchema を含め、受信ノード接続で schemaRef を構成します。

nodes: [
  {
    nodeType: 'Graph'
    name: 'schema-filter'
    graphSettings: {
      registryEndpointRef: 'default'
      artifact: 'azureiotoperations/graph-dataflow-filter:1.0.0'
      configuration: [
        {
          key: 'rules'
          value: '{"validateSchema":true,"filter":[]}'
        }
      ]
    }
  }
]
nodeConnections: [
  {
    from: {
      name: 'sensors'
      schema: {
        schemaRef: 'aio-sr://my-namespace/sensor-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: { name: 'schema-filter' }
  }
]

Important

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

nodes:
  - nodeType: Graph
    name: schema-filter
    graphSettings:
      registryEndpointRef: default
      artifact: azureiotoperations/graph-dataflow-filter:1.0.0
      configuration:
        - key: rules
          value: |
            {
              "validateSchema": true,
              "filter": []
            }

nodeConnections:
  - from:
      name: sensors
      schema:
        schemaRef: "aio-sr://my-namespace/sensor-schema:1"
        serializationFormat: Json
    to:
      name: schema-filter

ガイドライン：

パイプラインごとに 1 つの検証フィルターのみを使用します。
検証フィルターを最初に配置して、他の処理の前に無効なメッセージが削除されるようにします。
フィルター規則は、スキーマ検証に合格した後も適用されます。スキーマ検証のみが必要な場合は、フィルター規則を空のままにします。
schemaRefは、スキーマレジストリ内のスキーマを指している必要があります。 serializationFormatでは、スキーマ形式 (たとえば、Json) を指定します。

スキーマの構成については、「メッセージスキーマについて」を参照してください。

外部データを使用してフィルター規則を強化する

フィルタールールではデータセットがサポートされています。このデータセットを使用すると、外部状態ストアのデータと値を比較できます。データセットの構成の詳細については、「外部データを使用したエンリッチ」を参照してください。

フルフィルター構成

フィルター変換の構成で、入力とブール式を含む 1 つ以上のルールを追加します。必要に応じて、スキーマの検証を有効にし、エンリッチメント参照用のデータセットを構成します。

CLI は、1 つの構成ファイルからグラフ全体を適用するため、これをconfigurationの変換ノードのgraph.jsonに追加し、az iot ops dataflowgraph applyと共に適用します。

規則は JSON オブジェクトです。

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": ["temperature"],
      "expression": "$1 > 100",
      "description": "Drop high temperature readings"
    },
    {
      "inputs": ["rawValue", "$context(limits).maxValue"],
      "expression": "$1 > $2",
      "description": "Drop readings above device-specific limit"
    }
  ]
}

これらのルールは、エスケープされた文字列として value フィールドに入ります。

"configuration": [
  {
    "key": "rules",
    "value": "{\"datasets\":[{\"key\":\"device_limits as limits\",\"inputs\":[\"$source.deviceId\",\"$context.deviceId\"],\"expression\":\"$1 == $2\"}],\"filter\":[{\"inputs\":[\"temperature\"],\"expression\":\"$1 > 100\",\"description\":\"Drop high temperature readings\"},{\"inputs\":[\"rawValue\",\"$context(limits).maxValue\"],\"expression\":\"$1 > $2\",\"description\":\"Drop readings above device-specific limit\"}]}"
  }
]

フィルター規則のJSONは、valueキーのrulesとして渡されます。

configuration: [
  {
    key: 'rules'
    value: '{"datasets":[{"key":"device_limits as limits","inputs":["$source.deviceId","$context.deviceId"],"expression":"$1 == $2"}],"filter":[{"inputs":["temperature"],"expression":"$1 > 100","description":"Drop high temperature readings"},{"inputs":["rawValue","$context(limits).maxValue"],"expression":"$1 > $2","description":"Drop readings above device-specific limit"}]}'
  }
]

Important

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

{
  "datasets": [
    {
      "key": "device_limits as limits",
      "inputs": ["$source.deviceId", "$context.deviceId"],
      "expression": "$1 == $2"
    }
  ],
  "filter": [
    {
      "inputs": ["temperature"],
      "expression": "$1 > 100",
      "description": "Drop high temperature readings"
    },
    {
      "inputs": ["rawValue", "$context(limits).maxValue"],
      "expression": "$1 > $2",
      "description": "Drop readings above device-specific limit"
    }
  ]
}

鍵	必須	説明
`filter`	はい	フィルター規則の配列。
`datasets`	いいえ	エンリッチメント検索のデータセット定義の配列。
`validateSchema`	いいえ	`true`する場合は、フィルター規則を実行する前に、JSON スキーマに対してメッセージを検証します。既定値は `false` です。

ブランチ変換

分岐変換は、各受信メッセージの条件を評価し、 true または falseの 2 つの出力パスのいずれかにルーティングします。 (メッセージを削除する) フィルターとは異なり、ブランチはすべてのメッセージを保持し、適切なパスに転送します。

分岐のしくみ

すべてのメッセージは、2 つのパスのいずれかになります。何も削除されません。

主な制約:

分岐式は ブール値を返す必要があります。非ブール値の結果はエラーを引き起こします。
ワイルドカード入力なし。
分岐規則が 1 つだけです。 branch キーは、配列ではなく 1 つのオブジェクトを受け取ります。

Important

分岐はメッセージを別々の処理パスに分割しますが、すべてのパスは目的地に到達する前に連結変換を用いて再び統合しなければなりません。分岐は、複数のエンドポイントにルーティングする方法ではなく、異なるメッセージに異なる変換を適用する方法と考えてください。

分岐規則を定義する

重大度のしきい値に基づいてメッセージを分岐するには:

ブランチ変換の構成で、次の設定を行います。

Setting	価値
入力	`severity`
式	`$1 > 5`

規則は JSON オブジェクトです。

{
  "branch": {
    "inputs": ["severity"],
    "expression": "$1 > 5",
    "description": "Route high-severity messages"
  }
}

これらのルールは、エスケープされた文字列として value フィールドに入ります。

"configuration": [
  {
    "key": "rules",
    "value": "{\"branch\":{\"inputs\":[\"severity\"],\"expression\":\"$1 > 5\",\"description\":\"Route high-severity messages\"}}"
  }
]

configuration: [
  {
    key: 'rules'
    value: '{"branch":{"inputs":["severity"],"expression":"$1 > 5","description":"Route high-severity messages"}}'
  }
]

Important

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

{
  "branch": {
    "inputs": ["severity"],
    "expression": "$1 > 5",
    "description": "Route high-severity messages"
  }
}

severityが 5 より大きいメッセージは、true パスに移動します。その他はすべて、 false パスに移動します。

ブランチメッセージをスキーマに対して検証する

バージョン 1.1.0からは、ブランチ変換を設定して、JSONスキーマに対して受信メッセージを検証してからブランチ式を評価することができます。

スキーマ検証を有効にするには、分岐設定で validateSchema を true に設定してください。 validateSchemaフィールドはオプションで、デフォルトではfalseです。有効にすると、ブランチは受信ノード接続上の schemaRef からスキーマを取得します（ブランチノードに入力される from エントリの nodeConnections 側）。

スキーマ検証を通過したメッセージは分岐評価に続きます。
スキーマ検証に失敗したメッセージは false パスに送られます。

分岐変換の設定には 「スキーマの検証 」チェックボックスが含まれています。しかし、現在、Operationsの体験はノード接続上で schemaRef の設定や閲覧をサポートしていません。スキーマ検証を使用するには、Bicep マニフェストまたは Kubernetes マニフェストを使用して、ノード接続の schemaRef を構成します。

"nodes": [
  {
    "nodeType": "Graph",
    "name": "schema-branch",
    "graphSettings": {
      "registryEndpointRef": "default",
      "artifact": "azureiotoperations/graph-dataflow-branch:1.1.0",
      "configuration": [
        {
          "key": "rules",
          "value": "{\"validateSchema\":true,\"branch\":{\"inputs\":[\"severity\"],\"expression\":\"$1 > 5\"}}"
        }
      ]
    }
  }
],
"nodeConnections": [
  {
    "from": {
      "name": "sensors",
      "schema": {
        "schemaRef": "aio-sr://my-namespace/sensor-schema:1",
        "serializationFormat": "Json"
      }
    },
    "to": {
      "name": "schema-branch"
    }
  }
]

ヒント

分岐ルールのJSONに validateSchema を含め、受信ノード接続で schemaRef を設定します:

nodes: [
  {
    nodeType: 'Graph'
    name: 'schema-branch'
    graphSettings: {
      registryEndpointRef: 'default'
      artifact: 'azureiotoperations/graph-dataflow-branch:1.1.0'
      configuration: [
        {
          key: 'rules'
          value: '{"validateSchema":true,"branch":{"inputs":["severity"],"expression":"$1 > 5"}}'
        }
      ]
    }
  }
]
nodeConnections: [
  {
    from: {
      name: 'sensors'
      schema: {
        schemaRef: 'aio-sr://my-namespace/sensor-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: { name: 'schema-branch' }
  }
]

Important

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

nodes:
  - nodeType: Graph
    name: schema-branch
    graphSettings:
      registryEndpointRef: default
      artifact: azureiotoperations/graph-dataflow-branch:1.1.0
      configuration:
        - key: rules
          value: |
            {
              "validateSchema": true,
              "branch": {
                "inputs": ["severity"],
                "expression": "$1 > 5"
              }
            }

nodeConnections:
  - from:
      name: sensors
      schema:
        schemaRef: "aio-sr://my-namespace/sensor-schema:1"
        serializationFormat: Json
    to:
      name: schema-branch

分岐出力を接続する

パイプライン構成で、ノード名の後に .output.true または .output.false を使用して、各パスをダウンストリーム変換に接続します。

データフローグラフエディターで、分岐変換の true および false 出力から適切なダウンストリーム変換に接続をドラッグします。

CLI は、1 つの構成ファイルからグラフ全体を適用するため、これを graph.json 内の対応する場所に追加し、 az iot ops dataflowgraph applyで適用します。

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "severity-check"
    }
  },
  {
    "from": {
      "name": "severity-check.output.true"
    },
    "to": {
      "name": "alert-transform"
    }
  },
  {
    "from": {
      "name": "severity-check.output.false"
    },
    "to": {
      "name": "normal-transform"
    }
  }
]

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'severity-check' } }
  { from: { name: 'severity-check.output.true' }, to: { name: 'alert-transform' } }
  { from: { name: 'severity-check.output.false' }, to: { name: 'normal-transform' } }
]

Important

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

nodeConnections:
  - from: { name: sensors }
    to: { name: severity-check }
  - from: { name: severity-check.output.true }
    to: { name: alert-transform }
  - from: { name: severity-check.output.false }
    to: { name: normal-transform }

連結を使用してパスをマージする

すべての分岐パスは、宛先に到達する前に収束する必要があります。連結変換はそれらをマージします。構成も規則もありません。接続されているすべての入力からのメッセージは、変更されていない状態で通過します。

キャンバスに連結変換を追加し、両方の分岐パスをそれに接続してから、連結を宛先に接続します。

CLI は、1 つの構成ファイルからグラフ全体を適用するため、これを graph.json 内の対応する場所に追加し、 az iot ops dataflowgraph applyで適用します。

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

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

Important

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

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

例: フィルター、分岐、マージ

このエンド・ツー・エンドの例では、不正な読み取り値を除外し、重大度に応じて分岐し、各パスに異なるマップ変換を適用して、結果をマージします。

このパイプラインをオペレーション体験で構築するには:

データフローグラフを作成し、から読み取るtelemetry/sensorsを追加します。
フィルター変換を追加。 temperature > 1000メッセージを削除するルールを構成します。
ブランチ変換を追加します。重大度の高いメッセージを true パスにルーティングするように条件 severity > 5 を構成します。
真のパスに マップ 変換を追加します。 deviceIdの名前をidに変更し、temperatureにtempし、alertに設定trueフィールドを追加するルールを構成します。
false パスに マップ 変換を追加します。 deviceIdの名前をidに変更し、temperatureにtempするようにルールを構成します。
両方のパスをマージする連結変換を追加します。
宛先を追加してtelemetry/processedに送信します。
要素を接続: ソース → フィルター → ブランチ → (true パス: アラートマップ、false パス: ノーマルマップ) → 連結 → 宛先。

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

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/sensors"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "remove-bad-data",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-filter:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"filter\":[{\"inputs\":[\"temperature\"],\"expression\":\"$1 > 1000\",\"description\":\"Drop impossible temperature readings\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "severity-check",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-branch:1.1.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"branch\":{\"inputs\":[\"severity\"],\"expression\":\"$1 > 5\",\"description\":\"Route high-severity messages\"}}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "alert-transform",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"deviceId\"],\"output\":\"id\"},{\"inputs\":[\"temperature\"],\"output\":\"temp\"},{\"inputs\":[],\"output\":\"alert\",\"expression\":\"true\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "normal-transform",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"deviceId\"],\"output\":\"id\"},{\"inputs\":[\"temperature\"],\"output\":\"temp\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "merge",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-concatenate:1.0.0"
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/processed"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "remove-bad-data"
      }
    },
    {
      "from": {
        "name": "remove-bad-data"
      },
      "to": {
        "name": "severity-check"
      }
    },
    {
      "from": {
        "name": "severity-check.output.true"
      },
      "to": {
        "name": "alert-transform"
      }
    },
    {
      "from": {
        "name": "severity-check.output.false"
      },
      "to": {
        "name": "normal-transform"
      }
    },
    {
      "from": {
        "name": "alert-transform"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "normal-transform"
      },
      "to": {
        "name": "merge"
      }
    },
    {
      "from": {
        "name": "merge"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'alert-routing'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/sensors' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'remove-bad-data'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-filter:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"filter":[{"inputs":["temperature"],"expression":"$1 > 1000","description":"Drop impossible temperature readings"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'severity-check'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-branch:1.1.0'
          configuration: [
            {
              key: 'rules'
              value: '{"branch":{"inputs":["severity"],"expression":"$1 > 5","description":"Route high-severity messages"}}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'alert-transform'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["deviceId"],"output":"id"},{"inputs":["temperature"],"output":"temp"},{"inputs":[],"output":"alert","expression":"true"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'normal-transform'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["deviceId"],"output":"id"},{"inputs":["temperature"],"output":"temp"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'merge'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-concatenate:1.0.0'
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/processed'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
      { from: { name: 'remove-bad-data' }, to: { name: 'severity-check' } }
      { from: { name: 'severity-check.output.true' }, to: { name: 'alert-transform' } }
      { from: { name: 'severity-check.output.false' }, to: { name: 'normal-transform' } }
      { from: { name: 'alert-transform' }, to: { name: 'merge' } }
      { from: { name: 'normal-transform' }, to: { name: 'merge' } }
      { from: { name: 'merge' }, to: { name: 'output' } }
    ]
  }
}

Important

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

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

    - nodeType: Graph
      name: remove-bad-data
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-filter:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "filter": [
                  {
                    "inputs": ["temperature"],
                    "expression": "$1 > 1000",
                    "description": "Drop impossible temperature readings"
                  }
                ]
              }

    - nodeType: Graph
      name: severity-check
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-branch:1.1.0
        configuration:
          - key: rules
            value: |
              {
                "branch": {
                  "inputs": ["severity"],
                  "expression": "$1 > 5",
                  "description": "Route high-severity messages"
                }
              }

    - nodeType: Graph
      name: alert-transform
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["deviceId"], "output": "id" },
                  { "inputs": ["temperature"], "output": "temp" },
                  { "inputs": [], "output": "alert", "expression": "true" }
                ]
              }

    - nodeType: Graph
      name: normal-transform
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  { "inputs": ["deviceId"], "output": "id" },
                  { "inputs": ["temperature"], "output": "temp" }
                ]
              }

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

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

  nodeConnections:
    - from: { name: sensors }
      to: { name: remove-bad-data }
    - from: { name: remove-bad-data }
      to: { name: severity-check }
    - from: { name: severity-check.output.true }
      to: { name: alert-transform }
    - from: { name: severity-check.output.false }
      to: { name: normal-transform }
    - from: { name: alert-transform }
      to: { name: merge }
    - from: { name: normal-transform }
      to: { name: merge }
    - from: { name: merge }
      to: { name: output }

フィードバック

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

Last updated on 2026-08-03

データフローグラフにおけるデータのフィルタリング、分岐、マージ

前提条件

フィルター変換

フィルター ルールのしくみ

条件別にメッセージを削除する

条件ごとにメッセージを保持してください

複数の条件を使用する

複雑な式の使用

スキーマに対してフィルターメッセージを検証する

外部データを使用してフィルター 規則を強化する

フル フィルター構成

ブランチ変換

分岐のしくみ

分岐規則を定義する

ブランチメッセージをスキーマに対して検証する

分岐出力を接続する

連結を使用してパスをマージする

例: フィルター、分岐、マージ

関連するコンテンツ

フィードバック

その他のリソース

フィルタールールのしくみ

外部データを使用してフィルター規則を強化する

フルフィルター構成