Azure IoT Operations のData flow graphs

データフローグラフは、Azure IoT Operationsを通過するデータを処理する設定可能なパイプラインです。 標準 的なデータフロー は固定されたエンリッチ、フィルタ、マップのシーケンスに従いますが、データフローグラフは任意の順序で変換を合成したり、並列パスに分岐したり、時間ウィンドウにデータを集約したりできます。

DataflowGraphKubernetesのカスタムリソースはデータフローグラフを定義しています。 リソース内で、ソース、変換、目的地を接続して、シナリオに合った処理パイプラインを構築します。

データ フローとデータ フロー グラフ

Azure IoT Operations には、パイプライン内のデータを処理する 2 つの方法が用意されています。

サポートされているエンドポイントの種類を使用する新しいプロジェクトでは、データ フロー グラフをお勧めします。 データ フローはすべてのシナリオで完全にサポートされ、さまざまな種類のエンドポイントをサポートします。

使用可能な変換

各変換はルールで設定し、 DataflowGraph リソース内の他の変換と連鎖する既成処理ステップです。

すべての変換は、演算子、関数、およびフィールド参照の 式言語 を共有します。 マップ、フィルター、ブランチ変換の状態ストアからの外部データを使用してメッセージを エンリッチ することもできます。

データフローグラフにおける変換の合成方法

変換は、 DataflowGraph リソース内で順番に接続します。 ソース > 変換 A > 変換 B > ... > 変換先。

分岐変換によってフローは並列パスに分割され、連結変換によって再びマージされます。

任意の数の変換を任意の順序でチェーンできます。 1 つのマップ変換を持つパイプラインは、フィルター処理や分岐を行い、各パスを異なる方法でマッピングし、マージして時間窓で集計を行うパイプラインと同様に有効です。

データフローグラフの構成方法

データフローグラフ内の各変換は、コンテナレジストリから取得したあらかじめ構築されたアーティファクトを参照しています。 変換を構成するには、グラフ リソースの configuration セクションを通じて JSON としてルールを渡します。

Azure IoT Operationsをデプロイすると、自動的にmcr.microsoft.comを指すデフォルトのレジストリエンドポイントdefault作成されます。 組み込みの変換では、このエンドポイントを使用して Microsoft Container Registry から成果物をプルします。 追加のレジストリ設定は必要ありません。

データフローグラフリソースは、ソース、1つ以上の変換(それぞれ nodeType: Graph付き)、宛先の3種類の要素、そしてそれら間のデータの流れを説明する一連の nodeConnections を定義します。 各変換の configuration は、 rules キーの下にJSON文字列としてルールを渡します。

温度データを読み取り、摂氏を華氏にマップ変換で変換し、結果を公開する完全で実行可能な例(Operations、Azure CLI、Bicep、Kubernetes)については、データフローグラフの作成を参照してください。 以下のハウツー記事では、変換ルール自体に焦点を当てています。

ノード接続上でスキーマを設定する

データフローグラフはスキーマをデータフローとは異なる方法で扱います。 ソースまたは変換でスキーマを設定する代わりに、グラフ内のノード間の ノード接続 でスキーマを構成します。 分岐変換やフィルター変換は、ノード接続に付随するスキーマに対して実行時データを検証することも可能です。

nodeConnections配列の各エントリには、接続のfrom側にschemaを含めることができます。 このスキーマは、これら2つのノード間を流れるデータの期待されるフォーマットを記述しています:

nodeConnections: [
  {
    from: {
      name: 'source'
      schema: {
        schemaRef: 'aio-sr://my-namespace/sensor-data:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'transform'
    }
  }
]

nodeConnections:
  - from:
      name: source
      schema:
        schemaRef: "aio-sr://my-namespace/sensor-data:1"
        serializationFormat: Json
    to:
      name: transform

schemaRef値はフォーマットaio-sr://<namespace>/<name>:<version>を使用し、スキーマレジストリに保存されたスキーマを指しています。 データフローグラフはMQTT、Kafka、OpenTelemetryの各エンドポイントのみをサポートするため、対応されるシリアライズ形式は Jsonです。

以下の表は、データフローとデータフローグラフ間でスキーマ構成の違いをまとめたものです。

メッセージスキーマの定義、フォーマット、アップロード方法については「 メッセージスキーマの理解」を参照してください。

組み込みの変換と WASM 変換

データ フロー グラフでは、次の 2 種類の変換がサポートされています。

• 組み込みの変換はMicrosoftによって既に構築されています(map、filter、branch、concatenate、window、throttle)。 ルールを使用して構成します。 コーディングは必要ありません。
• WASM 変換 は、開発者がビルドしてデプロイするカスタム WebAssembly モジュールです。 組み込みの変換でカバーされないロジックが必要な場合に使用します。

両方の変換は同じ DataflowGraph リソース内で動作し、1つのパイプライン内で混ぜることができます。 カスタム変換の構築とデプロイの詳細については、「 データ フロー グラフでの WASM 変換の使用」を参照してください。

データフローグラフにおけるエラー処理

変換がメッセージ処理中にエラー(例:フィールドの欠落や無効な式)に遭遇すると、変換はメッセージを除外しエラーを記録します。 パイプラインは後続のメッセージの処理を続行します。

処理エラーの一般的な原因:

• ルールの inputs で参照されているフィールドがメッセージに存在しません。
• フィルター式または分岐式は、ブール以外の値を返します。
• 式は互換性のないデータ型を参照します(例えば、算術におけるJSONオブジェクト)。
• エンリッチメントに使用する状態ストアにアクセスできません。

エラーの処理を監視するには、データ フロー グラフのポッド ログを確認するか、メトリック エンドポイントを使用します。 詳細については、「 監視と監視の構成」を参照してください。

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

データフローグラフのパフォーマンスガイダンス

パイプライン内の各変換によって、処理オーバーヘッドが増加します。 次のガイドラインに留意してください。

• より多くのルールを使用して、変換の数を減らします。 同じ構造で動作する変換ルールが多数ある場合は、ルールごとに個別の変換を作成するのではなく、1 つのマップ変換に配置します。
• ロジックが異なる場合は、複数の変換を使用します。 異なる処理手順 (フィルター処理とマッピングと集計) が根本的に異なる場合は、個別の変換が理にかなっています。
• 関連するルールを一緒に保持します。 1 つのマップ変換で、フィールドの名前変更、再構築、計算フィールド、メタデータ変換をすべて一度に処理できます。

関連するコンテンツ

• データ フローとデータ フロー グラフ
• データ フロー グラフを作成する
• マップを使用してデータを変換する
• データのフィルター処理とルーティング
• 時間の経過に伴うデータの集計
• 外部データを使用してエンリッチする
• スロットルデータ
• さまざまなトピックにメッセージをルーティングする
• 式リファレンス
• データ フロー グラフで WASM 変換を使用する