データフローグラフにおけるウィンドウ変換を用いた集約データ

ウィンドウ変換は受信メッセージをグループ化し、ウィンドウが閉じると集約された値をまとめた単一の出力メッセージを生成します。すべての測定値を個別に転送する代わりに、平均値、最小値、カウントなどの統計値を計算し、統合された結果を下流に送信できます。

現在、ウィンドウは持続時間、カウント、メモリ、トリガー条件に基づいて閉じることができます。

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

注

非期間ベースのウィンドウは azureiotoperations/graph-dataflow-window:1.1.0 以降が必要です。

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

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

ウィンドウ変換は、 average、 min、 maxなどの集計関数を追加し、これらは蓄積ルールでのみ利用可能です。全リストについては「集約関数」を参照してください。

前提条件

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>"

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

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

Important

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

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

ウィンドウ変換を使用するタイミング

高周波センサーデータを受信し、ダウンストリームに送信する前にボリュームを減らしたい場合は、ウィンドウ変換を使用します。一般的なシナリオは、次のとおりです。

コンピューティング平均: 温度センサーは 1 秒ごとに発行されますが、クラウドアプリケーションで必要なのは 30 秒の平均のみです。
極値を追跡する: 1 分間隔ごとに最小および最大の圧力読み取り値が必要です。
イベントのカウント: 過去 5 分間に発生したドアオープンイベントの数を把握する必要があります。
生産バッチを作成する:固定サイズのバッチごとに統計を計算したいです。例えば、充填ラインから出る100個のパッケージごとに。
状態変化に応答する:ミキサーが running から drainingに変わるなど、動作信号が変わるたびに知りたいです。

ウィンドウ変換のしくみ

ウィンドウ変換には、次の 2 つの内部ステップが順番に接続されています。

ウィンドウ:設定されたクロージング条件のいずれかが実行されるまでメッセージをバッファリングします。
累積: ウィンドウが閉じたときに集計ルールを適用します。変換によりウィンドウ内のすべてのメッセージは単一の出力メッセージに縮小されます。

注

ウィンドウ変換は少なくとも1つのクローズニング条件( delay、 count、 memory、または triggers)を設定する必要があります。

ウィンドウクローズ条件の設定

バージョン 1.1.0以降、ウィンドウ変換は既存の delay キーに加えて3つの構成キーを追加します。

構成キー	ウィンドウの種類	Purpose
`delay`	持続時間に基づくウィンドウ	一定時間終了後にウィンドウを閉じてください。
`count`	カウントベースのウィンドウ	一定数のメッセージが届くとウィンドウを閉じます。
`memory`	メモリベースウィンドウ	バッファされたペイロードサイズが上限に達したらウィンドウを閉じます。
`triggers`	トリガーベースのウィンドウ	カスタム式が `true`に評価されたらウィンドウを閉じます。

持続時間に基づくウィンドウ

delay設定を使って、一定時間後にウィンドウを閉じます。この設定は、各ターンブリングウィンドウの持続時間を制御します。

注

遅延ステップは、メッセージのタイムスタンプをウィンドウの境界に合わせます。メッセージが10秒のウィンドウのうち7秒後に到着した場合、それは10秒の境界に属します。

注

delayを提供しない場合、ウィンドウは安全弁として60秒のデフォルトタイムアウトを使用します。

ウィンドウ変換の構成で、ウィンドウの期間を秒単位で設定します。たとえば、30 秒のタンブリングウィンドウの 30 に設定します。

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

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

{
  "type": "duration",
  "delaySeconds": 30
}

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

"configuration": [
  {
    "key": "delay",
    "value": "{\"type\":\"duration\",\"delaySeconds\":30}"
  }
]

Tip

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

configuration: [
  {
    key: 'delay'
    value: '{"type":"duration","delaySeconds":30}'
  }
]

Important

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

{
  "type": "duration",
  "delaySeconds": 30
}

財産	タイプ	説明
`type`	文字列	`"duration"`である必要があります。
`delaySeconds`	uint64	窓が閉まるまでの秒数。 0 より大きくなければなりません。

カウントベースのウィンドウ

count設定を使って、一定数のメッセージが届くとウィンドウを閉じます。

ウィンドウ変換設定で 、Message count を 5 に設定し、境界メッセージの動作を messageInCurrentに設定します。

CLIは1つの設定ファイルからグラフ全体を適用するため、この設定をconfigurationファイルのウィンドウ変換ノードの graph.json に追加してください。そして、 az iot ops dataflowgraph applyを使って塗布します。

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

{
  "type": "messageCount",
  "maxMessageCount": 5,
  "boundaryMessage": "messageInCurrent"
}

valueフィールドにこれらの規則をエスケープ文字列として加えます:

"configuration": [
  {
    "key": "count",
    "value": "{\"type\":\"messageCount\",\"maxMessageCount\":5,\"boundaryMessage\":\"messageInCurrent\"}"
  }
]

Tip

configuration: [
  {
    key: 'count'
    value: '{"type":"messageCount","maxMessageCount":5,"boundaryMessage":"messageInCurrent"}'
  }
]

Important

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

{
  "type": "messageCount",
  "maxMessageCount": 5,
  "boundaryMessage": "messageInCurrent"
}

財産	タイプ	説明
`type`	文字列	`"messageCount"`である必要があります。
`maxMessageCount`	uint64	ウィンドウが閉じるまでにバッファリングすべきメッセージの数。 0 より大きくなければなりません。
`boundaryMessage`	文字列	ウィンドウを閉じるメッセージが現在のウィンドウに残る(`messageInCurrent`)か次のウィンドウを始める(`messageInNext`)か。

メモリベースウィンドウ

バッファされたペイロードサイズが上限に達したときに memory 構成でウィンドウを閉じます。

ウィンドウ変換設定で、 バッファサイズ を 1048576 バイトに設定し、境界メッセージの挙動を messageInNextに設定します。

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

{
  "type": "bufferSize",
  "maxBufferBytes": 1048576,
  "boundaryMessage": "messageInNext"
}

valueフィールドにこれらの規則をエスケープ文字列として加えます:

"configuration": [
  {
    "key": "memory",
    "value": "{\"type\":\"bufferSize\",\"maxBufferBytes\":1048576,\"boundaryMessage\":\"messageInNext\"}"
  }
]

Tip

configuration: [
  {
    key: 'memory'
    value: '{"type":"bufferSize","maxBufferBytes":1048576,"boundaryMessage":"messageInNext"}'
  }
]

Important

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

{
  "type": "bufferSize",
  "maxBufferBytes": 1048576,
  "boundaryMessage": "messageInNext"
}

財産	タイプ	説明
`type`	文字列	`"bufferSize"`である必要があります。
`maxBufferBytes`	uint64	ウィンドウが閉じるまでの最大累積ペイロードバイト数。 0 より大きくなければなりません。
`boundaryMessage`	文字列	ウィンドウを閉じるメッセージが現在のウィンドウに残る(`messageInCurrent`)か次のウィンドウを始める(`messageInNext`)か。

トリガーベースのウィンドウ

ウィンドウが閉じるべき時は、メッセージ内容や現在のウィンドウ内の実行状態に基づいて triggers 設定を使いましょう。

ウィンドウ変換設定では、入力フィールド temperature、式 running_sum($1) + $1 > 100、境界メッセージ振る舞い messageInCurrentを含むトリガールールを追加します。

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

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "running_sum($1) + $1 > 100",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

valueフィールドにこれらの規則をエスケープ文字列として加えます:

"configuration": [
  {
    "key": "triggers",
    "value": "{\"type\":\"expression\",\"rules\":[{\"inputs\":[\"temperature\"],\"trigger\":\"running_sum($1) + $1 > 100\",\"boundaryMessage\":\"messageInCurrent\"}]}"
  }
]

Tip

configuration: [
  {
    key: 'triggers'
    value: '{"type":"expression","rules":[{"inputs":["temperature"],"trigger":"running_sum($1) + $1 > 100","boundaryMessage":"messageInCurrent"}]}'
  }
]

Important

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

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "running_sum($1) + $1 > 100",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

財産	必須	説明
`type`	はい	`"expression"`である必要があります。
`rules`	はい	トリガールールの配列。ルールはメッセージごとに逐次評価されます。最初の一致ルールでウィンドウが閉じます。
`datasets`	いいえ	ステートストアを参照するオプションのステートストアデータセット。

各トリガールールは以下のフィールドをサポートしています:

財産	必須	説明
`inputs`	はい	入力フィールド参照の配列。この表現は、 `$1`、 `$2`などに結びつきます。
`trigger`	はい	`true`に評価するとウィンドウが閉じるブール式。
`boundaryMessage`	はい	ウィンドウを閉じるメッセージが現在のウィンドウに残る(`messageInCurrent`)か次のウィンドウを始める(`messageInNext`)か。

inputsフィールドは、プレーンフィールド、??デフォルト、? $last、$context(key).field、$metadata.*など、データフローグラフの他の場所で使われる同じ入力構文をサポートしています。 $context(key)の利用についての詳細は「外部データで豊かにする」をご覧ください。

トリガー式は、正則グラフ式関数やウィンドウが閉じるとリセットされる以下の実行状態関数を使用できます。

Function	説明
`running_sum($1)`	現在のウィンドウ内の過去のメッセージの累積合計 `$1` 。
`running_avg($1)`	過去のメッセージ全体にわたる`$1`の累積平均。
`running_min($1)`	以前のメッセージで確認された `$1` の最小値。最初のメッセージでは `$1` を返します（1 つの要素の最小値はその要素自体です）。
`running_max($1)`	前のメッセージで見られた最大 `$1` 値。最初のメッセージで返 `$1` (ある要素の最大値はそのままです)。
`running_count($1)`	`$1` が含まれるメッセージ数。
`running_count()`	総メッセージ数(フィールドフィルターなし)。
`first($1)`	現在のウィンドウ内での `$1` の最初の非空の値。最初のメッセージで `$1` 返事。
`changed($1)`	`true` `$1`が前のメッセージの値と異なる場合。 `false` ウィンドウの最初のメッセージ(比較対象の前の値がない)に。
`prev($1)`	現在のウィンドウ内の以前のメッセージにおける `$1` の直近の非空の値。 `$1`が空だったメッセージはスキップされます(保存値は上書きされません)。ウィンドウの最初のメッセージで `$1` 返します。

注

running_sum($1) 同様の関数は、以前に処理されたメッセージから値を返します。現在のメッセージには $1を使ってください。

トリガールールの例

以下の例を使って、完全なトリガー構成オブジェクトにおける共通の inputs および trigger パターンを確認します:

この例は正則トリガー表現を示しています。現在の temperature が80を超えるとウィンドウが閉じます。

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "$1 > 80",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

この例は、 running_sum($1) + $1 を用いて現在のウィンドウ内の以前のメッセージと現在のメッセージを組み合わせ、閾値を超えるとウィンドウを閉じるトリガー式を示しています。

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "running_sum($1) + $1 > 100",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

この例では、 temperature ?? 0と境界メッセージを次のウィンドウに配置するための messageInNext を使ったnull-safe入力処理を示しています。

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature ?? 0"],
      "trigger": "running_avg($1) > 80",
      "boundaryMessage": "messageInNext"
    }
  ]
}

この例は、 $metadata.topicの特定のトピック値に対してウィンドウが閉じるメタデータベースのトリガーを示しています。

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["$metadata.topic"],
      "trigger": "$1 == \"telemetry/high-priority\"",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

この例は、データセットエンリッチメントを用いたトリガールールを示しています。メッセージfactoryIdを状態ストア行にマッチさせ、shiftIdから$context(factory).shiftId読み取り、シフト値が変化したとき(changed($1))ウィンドウを閉じます。

{
  "type": "expression",
  "datasets": [
    {
      "key": "factory",
      "inputs": ["$source.factoryId", "$context.factoryId"],
      "expression": "$1 == $2"
    }
  ],
  "rules": [
    {
      "inputs": ["$context(factory).shiftId"],
      "trigger": "changed($1)",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

この例では、 factory で表される状態ストアデータセットには factoryId や shiftIdのようなフィールドが含まれていることが期待されます。

境界挙動

boundaryMessage設定は、カウント、メモリ、またはトリガーに基づくウィンドウが閉じる原因となったメッセージの処理方法を決定します。

messageInCurrent: クロージングウィンドウに境界メッセージを含める。
messageInNextまず現在のウィンドウを閉じ、次に境界メッセージ付きの次のウィンドウを起動します。

messageInNext が新しいウィンドウ内の最初のメッセージで発生した場合、空のウィンドウが出力されないように、ウィンドウを閉じる処理は抑止されます。

注

持続時間ベースのウィンドウは boundaryMessageを使いません。持続時間の境界はメッセージではなく時間ベースなので、現在や次のウィンドウに境界メッセージを置く必要はありません。

終了条件を結合する

持続時間、カウント、メモリ、トリガー条件を同じグラフにまとめることができます。

持続時間は時間駆動され、タイマーによって評価されます。
各受信メッセージに対して、メッセージ駆動条件は次の順序で評価されます: Memory > Count > Trigger。
triggers.rules以内にルールが順次評価され、最初に一致するルールが勝ちます。

メッセージ駆動条件における評価順序は、メッセージが同時に複数の条件を満たす場合に蓄積結果に影響します。例えば、 memory が messageInCurrent を使い、 count が messageInNextを使う場合、両方の条件を満たすメッセージがメモリ構成に従っています。メッセージは現在のウィンドウに留まり、そのウィンドウの累積出力に寄与します。

累積ルールを定義する

各累積ルールは、メッセージのウィンドウを単一の出力値に減らす方法を指定します。構成キーは rules。

ウィンドウ変換構成では、入力 temperature、出力 avgTemperature、集計関数 average($1)を含む蓄積ルールを追加します。

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

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

{
  "accumulate": [
    {
      "inputs": ["temperature"],
      "output": "avgTemperature",
      "expression": "average($1)"
    }
  ]
}

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

{
  "key": "rules",
  "value": "{\"accumulate\":[{\"inputs\":[\"temperature\"],\"output\":\"avgTemperature\",\"expression\":\"average($1)\"}]}"
}

{
  key: 'rules'
  value: '{"accumulate":[{"inputs":["temperature"],"output":"avgTemperature","expression":"average($1)"}]}'
}

Important

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

{
  "accumulate": [
    {
      "inputs": ["temperature"],
      "output": "avgTemperature",
      "expression": "average($1)"
    }
  ]
}

財産	必須	説明
`inputs`	はい	各受信メッセージから読み取るフィールドパスの一覧。
`output`	はい	集計結果のフィールドパス。各ルールには一意の出力が必要です。
`expression`	はい	ウィンドウ全体の入力値を 1 つのスカラーに減らす数式。少なくとも 1 つの集計関数を含む必要があります。
`description`	いいえ	人間が判読できる説明。

マップルールとは異なり、すべての累積ルールに expression が必要です。 1 つのスカラーではなく値のコレクションを参照するため、 $1 単独では使用できません。 average($1)などの集計関数でラップする必要があります。

集計関数

Function	返品	空のウィンドウの動作
`average`	数値の平均	エラー
`sum`	数値の合計	0.0
`min`	最小数値	エラー
`max`	最大数値	エラー
`count`	フィールドが存在するメッセージの数	0
`first`	ウィンドウの最初の値	エラー
`last`	ウィンドウの最後の値	エラー

各関数は、引数として 1 つの位置変数を受け取ります (最初の入力の$1 、2 番目の入力の $2 など)。

数値以外の値: average、 sum、 min、および max 関数は、数値以外の値を自動的にスキップします。

プレゼンスベースの関数: count、 first、および last は、値の種類に関係なくフィールドプレゼンスで動作します。

集計を結合する

複数の集約関数を1つの式に組み合わせます:

入力 temperature と humidity、および式の average($1) + max($2)を含むルールを追加します。

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

{
  "inputs": [
    "temperature",
    "humidity"
  ],
  "output": "tempHumidityIndex",
  "expression": "average($1) + max($2)"
}

{
  inputs: [ 'temperature', 'humidity' ]
  output: 'tempHumidityIndex'
  expression: 'average($1) + max($2)'
}

Important

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

- inputs:
    - temperature     # $1
    - humidity        # $2
  output: tempHumidityIndex
  expression: "average($1) + max($2)"

集計値を変換するには、集計の外部に変換関数を適用します。たとえば、 cToF(average($1)) は平均温度を華氏に変換します。

各集計関数は、1 つの位置変数を直接参照する必要があります。 average($1) + max($2) は有効ですが、 average($1 + $2) は有効ではありません。

マップルールとの違い

能力	マップルール	累積ルール
式が必要です	いいえ	はい
ワイルドカード入力	サポートされている	サポートしていません
`$metadata` アクセス	サポートされている	サポートしていません
`$context` エンリッチメント	サポートされている	サポートしていません
`? $last` ディレクティブ	サポートされている	サポートしていません
出力コンテンツタイプ	入力と一致します	常に `application/json`

完全な構成の例

この例は、30秒、5メッセージ、1,048,576バイトバッファード、または running_sum($1) + $1 > 100後にウィンドウを閉じる完全なウィンドウ構成を示しています。例では、最後の3つの条件で boundaryMessage 値を messageInCurrent に設定し、ウィンドウが閉じると温度統計が計算されます。

どの条件がウィンドウを閉じるかは、メッセージのタイミング、カウント、ペイロードサイズ、内容によって異なります。以下の例は、各閉鎖条件に対する結果の出力を示しています。

期間終了

他の条件が先に満たされず、ウィンドウが以下の 3 つのメッセージを受信してから 30 秒に達した場合:

{ "temperature": 21.5 }
{ "temperature": 23.0 }
{ "temperature": 19.8 }

出力メッセージは次のとおりです。

{
  "avgTemperature": 21.433333333333334,
  "minTemperature": 19.8,
  "maxTemperature": 23.0,
  "readingCount": 3,
  "tempRange": 3.2
}

カウント終了

ウィンドウが他の条件が発生する前に次の5つのメッセージを受信した場合:

{ "temperature": 20.0 }
{ "temperature": 22.0 }
{ "temperature": 21.0 }
{ "temperature": 24.0 }
{ "temperature": 23.0 }

出力メッセージは次のとおりです。

{
  "avgTemperature": 22.0,
  "minTemperature": 20.0,
  "maxTemperature": 24.0,
  "readingCount": 5,
  "tempRange": 4.0
}

記憶は閉じる

たとえば、これら 2 つの大きなメッセージの後など、他の条件が満たされる前にバッファリングされたペイロードのサイズが 1,048,576 バイトに達した場合:

{ "temperature": 21.0, "payloadPad": "<large string>" }
{ "temperature": 22.5, "payloadPad": "<large string>" }

出力メッセージは次のとおりです。

{
  "avgTemperature": 21.75,
  "minTemperature": 21.0,
  "maxTemperature": 22.5,
  "readingCount": 2,
  "tempRange": 1.5
}

トリガーが閉じる

トリガー式 running_sum($1) + $1 > 100 が他のどの条件よりも先に発動する場合。たとえば、以下の3つのメッセージの後です:

{ "temperature": 40.0 }
{ "temperature": 35.0 }
{ "temperature": 30.0 }

出力メッセージは次のとおりです。

{
  "avgTemperature": 35.0,
  "minTemperature": 30.0,
  "maxTemperature": 40.0,
  "readingCount": 3,
  "tempRange": 10.0
}

操作エクスペリエンスで、ウィンドウ変換を含むデータフローグラフを作成します。

から読み取るtelemetry/temperatureを追加します。
ウィンドウ変換を追加します。 30秒の持続時間ウィンドウ、5メッセージカウント制限、1,048,576バイトのバッファサイズ制限、 temperature のトリガールールと式 running_sum($1) + $1 > 100を設定しましょう。カウント、メモリ、トリガー条件については、境界メッセージの挙動を messageInCurrentに設定します。 temperatureフィールドに平均、最小、最大、カウント、範囲の累積ルールを追加してください。
宛先を追加してtelemetry/aggregatedに送信します。

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

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "aggregate",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-window:1.1.0",
        "configuration": [
          {
            "key": "delay",
            "value": "{\"type\":\"duration\",\"delaySeconds\":30}"
          },
          {
            "key": "count",
            "value": "{\"type\":\"messageCount\",\"maxMessageCount\":5,\"boundaryMessage\":\"messageInCurrent\"}"
          },
          {
            "key": "memory",
            "value": "{\"type\":\"bufferSize\",\"maxBufferBytes\":1048576,\"boundaryMessage\":\"messageInCurrent\"}"
          },
          {
            "key": "triggers",
            "value": "{\"type\":\"expression\",\"rules\":[{\"inputs\":[\"temperature\"],\"trigger\":\"running_sum($1) + $1 > 100\",\"boundaryMessage\":\"messageInCurrent\"}]}"
          },
          {
            "key": "rules",
            "value": "{\"accumulate\":[{\"inputs\":[\"temperature\"],\"output\":\"avgTemperature\",\"expression\":\"average($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"minTemperature\",\"expression\":\"min($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"maxTemperature\",\"expression\":\"max($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"readingCount\",\"expression\":\"count($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"tempRange\",\"expression\":\"max($1) - min($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/aggregated"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "aggregate"
      }
    },
    {
      "from": {
        "name": "aggregate"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'temperature-window'
  parent: dataflowProfile
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'aggregate'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-window:1.1.0'
          configuration: [
            {
              key: 'delay'
              value: '{"type":"duration","delaySeconds":30}'
            }
            {
              key: 'count'
              value: '{"type":"messageCount","maxMessageCount":5,"boundaryMessage":"messageInCurrent"}'
            }
            {
              key: 'memory'
              value: '{"type":"bufferSize","maxBufferBytes":1048576,"boundaryMessage":"messageInCurrent"}'
            }
            {
              key: 'triggers'
              value: '{"type":"expression","rules":[{"inputs":["temperature"],"trigger":"running_sum($1) + $1 > 100","boundaryMessage":"messageInCurrent"}]}'
            }
            {
              key: 'rules'
              value: '{"accumulate":[{"inputs":["temperature"],"output":"avgTemperature","expression":"average($1)"},{"inputs":["temperature"],"output":"minTemperature","expression":"min($1)"},{"inputs":["temperature"],"output":"maxTemperature","expression":"max($1)"},{"inputs":["temperature"],"output":"readingCount","expression":"count($1)"},{"inputs":["temperature"],"output":"tempRange","expression":"max($1) - min($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/aggregated'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'aggregate' } }
      { from: { name: 'aggregate' }, to: { name: 'output' } }
    ]
  }
}

Important

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

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

    - nodeType: Graph
      name: aggregate
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-window:1.1.0
        configuration:
          - key: delay
            value: |
              {
                "type": "duration",
                "delaySeconds": 30
              }
          - key: count
            value: |
              {
                "type": "messageCount",
                "maxMessageCount": 5,
                "boundaryMessage": "messageInCurrent"
              }
          - key: memory
            value: |
              {
                "type": "bufferSize",
                "maxBufferBytes": 1048576,
                "boundaryMessage": "messageInCurrent"
              }
          - key: triggers
            value: |
              {
                "type": "expression",
                "rules": [
                  {
                    "inputs": ["temperature"],
                    "trigger": "running_sum($1) + $1 > 100",
                    "boundaryMessage": "messageInCurrent"
                  }
                ]
              }
          - key: rules
            value: |
              {
                "accumulate": [
                  {
                    "inputs": ["temperature"],
                    "output": "avgTemperature",
                    "expression": "average($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "minTemperature",
                    "expression": "min($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "maxTemperature",
                    "expression": "max($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "readingCount",
                    "expression": "count($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "tempRange",
                    "expression": "max($1) - min($1)"
                  }
                ]
              }

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

  nodeConnections:
    - from: { name: sensors }
      to: { name: aggregate }
    - from: { name: aggregate }
      to: { name: output }

次のステップ

フィードバック

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

Last updated on 2026-08-14

データフローグラフにおけるウィンドウ変換を用いた集約データ

前提条件

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

ウィンドウ変換を使用するタイミング

ウィンドウ変換のしくみ

ウィンドウクローズ条件の設定

持続時間に基づくウィンドウ

カウントベースのウィンドウ

メモリベースウィンドウ

トリガーベースのウィンドウ

トリガールールの例

境界挙動

終了条件を結合する

累積ルールを定義する

集計関数

集計を結合する

マップ ルールとの違い

完全な構成の例

期間終了

カウント終了

記憶は閉じる

トリガーが閉じる

次のステップ

フィードバック

その他のリソース

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

マップルールとの違い