Azure IoT Operations でデータフローを構成する

[アーティクル]
02/26/2025

重要

このページには、プレビュー段階にある Kubernetes デプロイマニフェストを使用して Azure IoT Operations コンポーネントを管理する手順が含まれます。この機能はいくつかの制限を設けて提供されており、運用環境のワークロードには使用しないでください。

ベータ版、プレビュー版、または一般提供としてまだリリースされていない Azure の機能に適用される法律条項については、「Microsoft Azure プレビューの追加使用条件」を参照してください。

データフローとは、データがソースから宛先までたどるパスであり、必要に応じて変換が行われます。データフローを構成するには、Dataflow カスタムリソースを作成するか、Azure IoT Operations Studio ポータルを使用します。データフローは次の 3 つの部分で構成されます: ソース、変換、宛先。

ソースから変換、次に宛先までのフローを示すデータフローのダイアグラム。

ソースと宛先を定義するには、データフローエンドポイントを構成する必要があります。変換は省略可能であり、データのエンリッチメント、データのフィルター処理、データの別のフィールドへのマッピングなどの操作を含めることができます。

重要

各データフローでは、ソースまたは宛先の "いずれか" として、Azure IoT Operations ローカル MQTT ブローカーの既定のエンドポイントを使用する必要があります。

Azure IoT Operations の操作エクスペリエンスを使用して、データフローを作成できます。操作エクスペリエンスには、データフローを構成するためのビジュアルインターフェイスが用意されています。 Bicep を使って Bicep テンプレートファイルを使用するデータフローを作成したり、Kubernetes を使って YAML ファイルを使用するデータフローを作成したりすることもできます。

ソース、変換、宛先を構成する方法については、引き続きお読みください。

前提条件

既定のデータフロープロファイルとエンドポイントを使用して Azure IoT Operations のインスタンスが作成されたら、すぐにデータフローを展開できます。ただし、データフロープロファイルとエンドポイントを構成してデータフローをカスタマイズしたい場合があります。

データフロープロファイル

データフローに複数の異なるスケーリング設定が必要ない場合は、Azure IoT Operations が提供する既定のデータフロープロファイルを使用します。データフロープロファイルを構成する方法については、「データフロープロファイルを構成する」を参照してください。

データフローエンドポイント

データフローのソースと宛先を構成するには、データフローエンドポイントが必要です。すぐに開始するには、ローカル MQTT ブローカーの既定のデータフローエンドポイントを使用できます。 Kafka、Event Hubs、Azure Data Lake Storage などの他の種類のデータフローエンドポイントを作成することもできます。各種のデータフローエンドポイントを構成する方法については、「データフローエンドポイントを構成する」を参照してください。

作業の開始

前提条件を満たしたら、データフローの作成を開始できます。

操作エクスペリエンスでデータフローを作成するには、[データフロー]>[データフローを作成する] を選択します。次に、データフローのソース、変換、宛先を構成できるページが表示されます。

操作エクスペリエンスを使用してデータフローを作成するスクリーンショット。

Bicep .bicep ファイルを作成して、データフローの作成を開始します。この例では、ソース、変換、宛先の構成を含むデータフローの構造を示します。

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

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

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

resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent data flow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // See source configuration section
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // See transformation configuration section
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // See destination configuration section
        }
      }
    ]
  }
}

Kubernetes マニフェスト .yaml ファイルを作成して、データフローの作成を開始します。この例では、ソース、変換、宛先の構成を含むデータフローの構造を示します。

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow profile
  # This field is required when configuring via Kubernetes YAML
  # The syntax is different when using Bicep
  profileRef: default 
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
      # Transformation optional
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

データフローの操作の種類を構成する方法については、以下のセクションを参照してください。

ソース

データフローのソースを構成するには、エンドポイント参照とエンドポイントのデータソースの一覧を指定します。データフローのソースとして、次のいずれかのオプションを選択します。

既定のエンドポイントがソースとして使用されていない場合は、宛先として使用する必要があります。詳細については、「データフローではローカル MQTT ブローカーエンドポイントの使用が必須」を参照してください。

オプション 1: 既定のメッセージブローカーエンドポイントをソースとして使用する

[ソースの詳細] で、[メッセージブローカー] を選択します。

メッセージブローカーソースについて次の設定を入力します。

設定	説明
データフローエンドポイント	既定の MQTT メッセージブローカーエンドポイントを使用するには、"既定値" を選択します。
トピック	受信メッセージをサブスクライブするための MQTT トピックフィルター。 [トピック]>[行の追加] を使用して、複数のトピックを追加します。トピックの詳細については、「MQTT または Kafka トピックの構成」を参照してください。
メッセージスキーマ	受信メッセージの逆シリアル化に使用するスキーマ。「データを逆シリアル化するスキーマを指定する」を参照してください。

適用を選択します。

メッセージブローカーエンドポイントは、Bicep テンプレートファイルで構成されます。たとえば、次のエンドポイントはデータフローのソースです。

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/telemetry/temperature/#'
    'humidifiers/+/telemetry/humidity/#'
  ]
}

ここでは、エンドポイント構成を変更せずに dataSources で複数の MQTT または Kafka トピックを指定できます。この柔軟性により、トピックが異なる場合でも、複数のデータフロー間で同じエンドポイントを再利用できます。詳細については、「データソースを構成する」を参照してください。

たとえば、1 つのメッセージブローカーエンドポイントと 2 つのトピックフィルターを使ってソースを構成するには、次の構成を使用します。

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/humidity/#

dataSources ではエンドポイント構成を変更せずに MQTT または Kafka トピックを指定できるため、トピックが異なる場合でも複数のデータフローに対してエンドポイントを再利用できます。詳細については、「データソースを構成する」を参照してください。

オプション 2: 資産をソースとして使用する

資産をデータフローのソースとして使用できます。資産をソースとして使用することは、操作エクスペリエンスでのみ使用できます。

[ソースの詳細] で、[資産] を選択します。
ソースエンドポイントとして使用する資産を選択します。
[続行] を選択します。

選択した資産のデータポイントの一覧が表示されます。
[適用] を選択して、資産をソースエンドポイントとして使用します。

資産をソースとして使用する場合、資産定義はデータフローのスキーマを推論するために使用されます。資産定義には、資産のデータポイントのスキーマが含まれます。詳細については、「資産の構成をリモートで管理する」を参照してください。

構成が完了すると、資産からのデータは、ローカル MQTT ブローカーを介してデータフローに到達します。そのため、ソースとして資産を使用する場合、データフローはローカル MQTT ブローカーの既定のエンドポイントを実際のソースとして使用します。

オプション 3: カスタム MQTT または Kafka データフローエンドポイントをソースとして使用する

カスタム MQTT または Kafka データフローエンドポイントを作成した場合 (たとえば、Event Grid または Event Hubs で使用する場合)、データフローのソースとして使用できます。 Data Lake や Fabric OneLake などのストレージタイプのエンドポイントは、ソースとして使用できないことに注意してください。

[ソースの詳細] で、[メッセージブローカー] を選択します。

メッセージブローカーソースについて次の設定を入力します。

設定	説明
データフローエンドポイント	カスタムの MQTT または Kafka データフローエンドポイントを選択するには、[再選択] ボタンを使用します。詳細については、「MQTT データフローエンドポイントを構成する」または「Azure Event Hubs と Kafka データフローエンドポイントを構成する」を参照してください。
トピック	受信メッセージをサブスクライブするための MQTT トピックフィルター。 [トピック]>[行の追加] を使用して、複数のトピックを追加します。トピックの詳細については、「MQTT または Kafka トピックの構成」を参照してください。
メッセージスキーマ	受信メッセージの逆シリアル化に使用するスキーマ。「データを逆シリアル化するスキーマを指定する」を参照してください。

適用を選択します。

プレースホルダーの値をカスタムエンドポイントの名前とトピックに置き換えます。

sourceSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_1>'
    '<TOPIC_2>'
    // See section on configuring MQTT or Kafka topics for more information
  ]
}

プレースホルダーの値をカスタムエンドポイントの名前とトピックに置き換えます。

sourceSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataSources:
  - <TOPIC_1>
  - <TOPIC_2>
  # See section on configuring MQTT or Kafka topics for more information

データソースを構成する (MQTT または Kafka のトピック)

データフローエンドポイントの構成を変更しなくても、ソースに複数の MQTT または Kafka トピックを指定できます。この柔軟性により、トピックが異なる場合でも、複数のデータフロー間で同じエンドポイントを再利用できます。詳細については、データフローエンドポイントの再利用に関するセクションを参照してください。

MQTT のトピック

ソースが MQTT (Event Grid を含む) エンドポイントの場合は、MQTT トピックフィルターを使用して受信メッセージをサブスクライブできます。トピックフィルターには、複数のトピックをサブスクライブするためのワイルドカードを含めることができます。たとえば、thermostats/+/telemetry/temperature/# はサーモスタットからのすべての温度テレメトリメッセージをサブスクライブします。 MQTT トピックフィルターを構成するには:

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

sourceSettings: {
  endpointRef: '<MESSAGE_BROKER_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_FILTER_1>'
    '<TOPIC_FILTER_2>'
    // Add more topic filters as needed
  ]
}

ワイルドカードを含む複数の MQTT トピックフィルターの例:

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/telemetry/temperature/#'
    'humidifiers/+/telemetry/humidity/#'
  ]
}

ここでは、ワイルドカード + を使用して、thermostats と humidifiers トピックの下にあるすべてのデバイスを選択します。 # ワイルドカードは、temperature と humidity トピックのすべてのサブトピックのすべてのテレメトリメッセージを選択するために使用されます。

sourceSettings:
  endpointRef: <ENDPOINT_NAME>
  dataSources:
    - <TOPIC_FILTER_1>
    - <TOPIC_FILTER_2>
    # Add more topic filters as needed

ワイルドカードを含む複数のトピックフィルターの例:

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/humidity/#

共有サブスクリプション

メッセージブローカーソースで共有サブスクリプションを使用するには、$shared/<GROUP_NAME>/<TOPIC_FILTER> の形式で共有サブスクリプショントピックを指定します。

操作エクスペリエンスのデータフローの [ソースの詳細] で、[メッセージブローカー] を選択し、[トピック] フィールドを使用して共有サブスクリプショングループとトピックを指定します。

sourceSettings: {
  dataSources: [
    '$shared/<GROUP_NAME>/<TOPIC_FILTER>'
  ]
}

sourceSettings:
  dataSources:
    - $shared/<GROUP_NAME>/<TOPIC_FILTER>

データフロープロファイルのインスタンス数が 1 より大きい場合は、メッセージブローカーソースを使用するすべてのデータフローに対して共有サブスクリプションが自動で有効になります。この場合、$shared プレフィックスが追加され、共有サブスクリプショングループ名が自動的に生成されます。たとえば、インスタンス数が 3 のデータフロープロファイルがあり、データフローがトピック topic1 と topic2 で構成されたソースとしてメッセージブローカーエンドポイントを使用している場合、それらは自動的に共有サブスクリプションに $shared/<GENERATED_GROUP_NAME>/topic1 および $shared/<GENERATED_GROUP_NAME>/topic2 として変換されます。

構成内に $shared/mygroup/topic という名前のトピックを明示的に作成できます。ただし、$shared プレフィックスは必要に応じて自動的に追加されるため、$shared トピックを明示的に追加することはお勧めしません。データフローが設定されていない場合は、グループ名を使用して最適化を行うことができます。たとえば、$share は設定されておらず、データフローはトピック名に対してのみ動作する必要がある場合があります。

重要

インスタンス数が 1 つ以上のときに共有サブスクリプションを必要とするデータフローは、Event Grid MQTT ブローカーをソースとして使用する場合に、共有サブスクリプションをサポートしていないため重要となります。メッセージの欠落を回避するため、Event Grid MQTT ブローカーをソースとして使用するときは、データフロープロファイルインスタンス数を 1 に設定します。これは、データフローがサブスクライバーで、クラウドからメッセージを受信する場合です。

Kafka トピック

ソースが Kafka (Event Hubs を含む) エンドポイントの場合は、受信メッセージをサブスクライブする個々の Kafka トピックを指定します。ワイルドカードはサポートされていないため、各トピックを静的に指定する必要があります。

Note

Kafka エンドポイント経由で Event Hubs を使用する場合、名前空間内の個々のイベントハブは Kafka トピックです。たとえば、thermostats と humidifiers の 2 つのイベントハブを含む Event Hubs 名前空間がある場合、各イベントハブを Kafka トピックとして指定できます。

Kafka トピックを構成するには:

操作エクスペリエンスのデータフローの [ソースの詳細] で [メッセージブローカー] を選択してから、[トピック] フィールドを使用して、受信メッセージをサブスクライブするための Kafka トピックフィルターを指定します。

Note

操作エクスペリエンスで指定できるトピックフィルターは 1 つだけです。複数のトピックフィルターを使用するには、Bicep または Kubernetes を使用してください。

sourceSettings: {
  endpointRef: '<KAFKA_ENDPOINT_NAME>'
  dataSources: [
    '<KAFKA_TOPIC_1>'
    '<KAFKA_TOPIC_2>'
    // Add more Kafka topics as needed
  ]
}

sourceSettings:
  endpointRef: <KAFKA_ENDPOINT_NAME>
  dataSources:
    - <KAFKA_TOPIC_1>
    - <KAFKA_TOPIC_2>
    # Add more Kafka topics as needed

ソーススキーマを指定する

MQTT または Kafka をソースとして使用する場合は、スキーマを指定して、操作エクスペリエンスポータルにデータポイントの一覧を表示できます。受信メッセージの逆シリアル化および検証のためのスキーマの使用は、現在サポートされていません。

ソースが資産の場合、スキーマは資産定義から自動的に推論されます。

ヒント

サンプルデータファイルからスキーマを生成するには、Schema Gen Helper を使用します。

ソースからの受信メッセージを逆シリアル化するために使用するスキーマを構成するには:

操作エクスペリエンスのデータフローの [ソースの詳細] で [メッセージブローカー] を選択し、[メッセージスキーマ] フィールドを使用してスキーマを指定します。 [アップロード] ボタンを使用して、最初にスキーマファイルをアップロードできます。詳細については、「メッセージスキーマを理解する」を参照してください。

スキーマレジストリを使用してスキーマを格納したら、それをデータフロー構成で参照できます。

sourceSettings: {
  serializationFormat: 'Json'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
}

スキーマレジストリを使用してスキーマを格納したら、それをデータフロー構成で参照できます。

sourceSettings:
  serializationFormat: Json
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'

詳細については、「メッセージスキーマを理解する」を参照してください。

変換

変換操作では、宛先に送信する前にソースからのデータを変換できます。変換は省略可能です。データを変更する必要がない場合は、データフロー構成に変換操作を含めないでください。複数の変換は、構成で指定した順序に関係なく、段階的に連結されます。ステージの順序は常に次のようになります。

エンリッチ: 一致するデータセットと条件を指定して、ソースデータにデータを追加します。
フィルター: 条件に基づいてデータをフィルター処理します。
[マップ]、[コンピューティング]、[名前の変更]、または [新しいプロパティ] の追加: 省略可能な変換を使用して、あるフィールドから別のフィールドにデータを移動します。

このセクションでは、データフロー変換の概要について説明します。詳細については、「データフローを使用してデータをマップする」、「データフロー変換を使用してデータを変換する」、「データフローを使用してデータをエンリッチする」を参照してください。

操作エクスペリエンスで、[データフロー]>[変換の追加 (省略可能)] を選びます。

操作エクスペリエンスを使用して変換をデータフローに追加するスクリーンショット。

builtInTransformationSettings: {
  datasets: [
    // See section on enriching data
  ]
  filter: [
    // See section on filtering data
  ]
  map: [
    // See section on mapping data
  ]
}

builtInTransformationSettings:
  datasets:
    # See section on enriching data
  filter:
    # See section on filtering data
  map:
    # See section on mapping data

エンリッチ: 参照データを追加する

データをエンリッチするには、まず Azure IoT Operations の状態ストアに参照データセットを追加します。データセットは、条件に基づいてソースデータにさらにデータを追加するために使用されます。条件は、データセット内のフィールドと一致するソースデータ内のフィールドとして指定されます。

状態ストア CLI を使用して、状態ストアにサンプルデータを読み込むことができます。状態ストア内のキー名は、データフロー構成のデータセットに対応します。

現在、"エンリッチ" ステージは操作エクスペリエンスではサポートされていません。

この例に、ソースデータの deviceId フィールドを使用して、データセットの asset フィールドと一致させる方法を示しています。

builtInTransformationSettings: {
  datasets: [
    {
      key: 'assetDataset'
      inputs: [
        '$source.deviceId' // ---------------- $1
        '$context(assetDataset).asset' // ---- $2
      ]
      expression: '$1 == $2'
    }
  ]
}

たとえば、ソースデータの deviceId フィールドを使用して、データセットの asset フィールドと一致させることができます。

builtInTransformationSettings:
  datasets:
  - key: assetDataset
    inputs:
      - $source.deviceId # ------------- $1
      - $context(assetDataset).asset # - $2
    expression: $1 == $2

データセットに asset フィールドを持つレコードがある場合は、次のようになります。

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

thermostat1 と一致する deviceId フィールドを持つソースのデータには、フィルターとマップのステージで使用できる location と manufacturer のフィールドがあります。

条件構文の詳細については、「データフローを使用してデータをエンリッチする」と「データフローを使用したデータの変換」に関する記事を参照してください。

フィルター: 条件に基づいてデータをフィルター処理する

条件に基づいてデータをフィルター処理するには、filter ステージを使用できます。条件は、値と一致するソースデータ内のフィールドとして指定されます。

[変換 (省略可能)] で、[フィルター]>[追加] を選びます。

必要な設定を入力します。

設定	説明
フィルターの条件	ソースデータのフィールドに基づいてデータをフィルター処理する条件。
説明	フィルター条件の説明を入力します。

フィルター条件フィールドに「@」と入力するか、Ctrl + Space キーを押して、ドロップダウンからデータポイントを選択します。

MQTT メタデータプロパティは、形式 @$metadata.user_properties.<property> または @$metadata.topic を使用して入力できます。 @$metadata.<header> 形式を使用して、$metadata ヘッダーを入力することもできます。 $metadata 構文は、メッセージヘッダーの一部である MQTT プロパティにのみ必要です。詳細については、「フィールド参照」を参照してください。

条件では、ソースデータのフィールドを使用できます。たとえば、@temperature > 20 のようなフィルター条件を使用して、温度フィールドに基づいて 20 以下のデータをフィルター処理できます。

適用を選択します。

たとえば、ソースデータの temperature フィールドを使用してデータをフィルター処理できます。

builtInTransformationSettings: {
  filter: [
    {
      inputs: [
        'temperature ? $last'
      ]
      expression: '$1 > 20'
    }
  ]
}

temperature フィールドが 20 より大きい場合、データは次のステージに渡されます。 temperature フィールドが 20 以下の場合、データはフィルター処理されます。

たとえば、ソースデータの temperature フィールドを使用してデータをフィルター処理できます。

builtInTransformationSettings:
  filter:
    - inputs:
      - temperature ? $last # - $1
      expression: "$1 > 20"

マップ: あるフィールドから別のフィールドにデータを移動する

省略可能な変換を使用してデータを別のフィールドにマップするには、map 操作を使用できます。変換は、ソースデータのフィールドを使用する数式として指定されます。

操作エクスペリエンスでは、マッピングは現在 [コンピューティング]、[名前変更]、[新しいプロパティ] の各変換を使用してサポートされています。

Compute

[コンピューティング] 変換を使用して、ソースデータに数式を適用できます。この操作は、ソースデータに数式を適用し、結果フィールドを格納するために使用されます。

[変換 (省略可能)] で、[コンピューティング]>[追加] を選びます。

必要な設定を入力します。

設定	説明
数式を選択する	ドロップダウンから既存の数式を選択するか、[カスタム] を選択して手動で数式を入力します。
出力	結果の出力表示名を指定します。
式	ソースデータに適用する数式を入力します。
説明	変換の説明を入力します。
最後の既知の値	必要に応じて、現在の値が使用できない場合は、最後の既知の値を使用します。

[数式] フィールドに数式を入力するか、既存の数式を編集できます。数式には、ソースデータ内のフィールドを使用できます。「@」と入力するか、Ctrl + Space キーを押して、ドロップダウンからデータポイントを選択します。組み込みの数式の場合は、<dataflow> プレースホルダーを選択して、使用可能なデータポイントの一覧を表示します。

数式には、ソースデータ内のフィールドを使用できます。たとえば、ソースデータの temperature フィールドを使用して温度を摂氏に変換し、それを temperatureCelsius 出力フィールドに格納できます。

適用を選択します。

名前の変更

[名前の変更] 変換を使用して、データポイントの名前を変更できます。この操作は、ソースデータ内のデータポイントの名前を新しい名前に変更するために使用されます。新しい名前は、データフローの後続のステージで使用できます。

[変換 (省略可能)] で、[名前の変更]>[追加] を選びます。

必要な設定を入力します。

設定	説明
データポイント	ドロップダウンからデータポイントを選択するか、$metadata ヘッダーを入力します。
新しいデータポイント名	データポイントの新しい名前を入力します。
説明	変換の説明を入力します。

適用を選択します。

新しいプロパティ

[新しいプロパティ] 変換を使用して、ソースデータに新しいプロパティを追加できます。この操作は、ソースデータに新しいプロパティを追加するために使用されます。新しいプロパティは、データフローの後続のステージで使用できます。

[変換 (省略可能)] で、[新しいプロパティ]>[追加] を選びます。

必要な設定を入力します。

設定	説明
プロパティキー	新しいプロパティのキーを入力します。
プロパティ値	新しいプロパティの値を入力します。
説明	新しいプロパティの説明を入力します。

適用を選択します。

MQTT メタデータプロパティには、形式 $metadata.user_properties.<property> または $metadata.topic を使用してアクセスできます。 $metadata.<header> 形式を使用して、$metadata ヘッダーを入力することもできます。詳細については、「フィールド参照」を参照してください。

たとえば、ソースデータの temperature フィールドを使用して温度を摂氏に変換し、それを temperatureCelsius フィールドに格納できます。コンテキスト化データセットの location フィールドを使用してソースデータをエンリッチすることもできます。

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        'temperature'
      ]
      output: 'temperatureCelsius'
      expression: '($1 - 32) * 5/9'
    }
    {
      inputs: [
        '$context(assetDataset).location'
      ]
      output: 'location'
    }
  ]
}

builtInTransformationSettings:
  map:
    - inputs:
      - temperature # - $1
      expression: "($1 - 32) * 5/9"
      output: temperatureCelsius
    - inputs:
      - $context(assetDataset).location  
      output: location

詳細については、「データフローを使用してデータをマッピングする」と「データフローを使用したデータの変換」に関する記事を参照してください。

スキーマに従ってデータをシリアル化する

データを宛先に送信する前にシリアル化する場合は、スキーマとシリアル化形式を指定する必要があります。それ以外の場合は、データは推論された型を使用して JSON でシリアル化されます。 Microsoft Fabric や Azure Data Lake などのストレージエンドポイントには、データの一貫性を確保するためにスキーマが必要です。サポートされているシリアル化形式は、Parquet と Delta です。

ヒント

サンプルデータファイルからスキーマを生成するには、Schema Gen Helper を使用します。

操作エクスペリエンスの場合は、データフローエンドポイントの詳細でスキーマとシリアル化の形式を指定します。シリアル化形式をサポートするエンドポイントは、Microsoft Fabric OneLake、Azure Data Lake Storage Gen 2、Azure Data Explorer、およびローカルストレージです。たとえば、差分形式でデータをシリアル化するには、スキーマをスキーマレジストリにアップロードし、データフロー変換先エンドポイント構成で参照する必要があります。

操作エクスペリエンスを使用してデータフロー変換先エンドポイントのシリアル化を設定するスクリーンショット。

スキーマレジストリにスキーマをアップロードしたら、データフロー構成でスキーマを参照できます。

builtInTransformationSettings: {
  serializationFormat: 'Delta'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'
}

スキーマレジストリにスキーマをアップロードしたら、データフロー構成でスキーマを参照できます。

builtInTransformationSettings:
  serializationFormat: Delta
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'

スキーマレジストリの詳細については、「メッセージスキーマについて」を参照してください。

宛先

データフローの宛先を構成するには、エンドポイント参照とデータ宛先を指定します。エンドポイントのデータの宛先の一覧を指定できます。

ローカル MQTT ブローカー以外の宛先にデータを送信するには、データフローエンドポイントを作成します。その方法については、データフローエンドポイントの構成に関するページを参照してください。宛先がローカル MQTT ブローカーでない場合は、ソースとして使用する必要があります。詳細については、「データフローではローカル MQTT ブローカーエンドポイントの使用が必須」を参照してください。

重要

ストレージエンドポイントには、シリアル化にスキーマが必要です。 Microsoft Fabric OneLake、Azure Data Lake Storage、Azure Data Explorer、またはローカル記憶域でデータフローを使用するには、スキーマ参照を指定する必要があります。

宛先として使用するデータフローエンドポイントを選択します。

ストレージエンドポイントには、シリアル化にスキーマが必要です。 Microsoft Fabric OneLake、Azure Data Lake Storage、Azure Data Explorer、またはローカルストレージの宛先エンドポイントを選択する場合は、スキーマ参照を指定する必要があります。たとえば、データを Delta 形式で Microsoft Fabric エンドポイントにシリアル化するには、スキーマをスキーマレジストリにアップロードし、データフローの宛先エンドポイントの構成内でそれを参照する必要があります。
[続行] を選択して、宛先を構成します。
データの送信先となるトピックやテーブルなど、宛先に必要な設定を入力します。詳細については、「データの宛先の構成 (トピック、コンテナー、またはテーブル)」を参照してください。

destinationSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataDestination: '<TOPIC_OR_TABLE>' // See section on configuring data destination
}

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE> # See section on configuring data destination

データの宛先 (トピック、コンテナー、またはテーブル) を構成する

データソースと同様に、データの宛先は、複数のデータフロー間でデータフローエンドポイントを再利用可能に保つために使用される概念です。基本的には、データフローエンドポイント構成のサブディレクトリを表します。たとえば、データフローエンドポイントがストレージエンドポイントの場合、データの宛先はストレージアカウント内のテーブルです。データフローエンドポイントが Kafka エンドポイントの場合、データの宛先は Kafka トピックです。

エンドポイントの種類	データの宛先の意味	説明
MQTT (または Event Grid)	トピック	データが送信される MQTT トピック。静的トピックのみがサポートされ、ワイルドカードはサポートされません。
Kafka (または Event Hubs)	トピック	データが送信される Kafka トピック。静的トピックのみがサポートされ、ワイルドカードはサポートされません。エンドポイントが Event Hubs 名前空間の場合、データの宛先は名前空間内の個々のイベントハブです。
Azure Data Lake Storage	コンテナー	ストレージアカウントのコンテナー。テーブルではありません。
Microsoft Fabric OneLake	ファイルまたはフォルダー	構成済みのエンドポイントのパスの種類に対応します。
Azure Data Explorer	テーブル	Azure Data Explorer データベース内のテーブル。
ローカルストレージ	フォルダー	ローカルストレージの永続ボリュームマウント内のフォルダーまたはディレクトリ名。 Azure Arc クラウド取り込みエッジボリュームで有効な Azure コンテナーストレージを使用する場合、これは、作成したサブボリュームの `spec.path` パラメーターと一致する必要があります。

データの宛先を構成するには:

操作エクスペリエンスを使用する場合、データの宛先フィールドはエンドポイントの種類に基づいて自動的に解釈されます。たとえば、データフローエンドポイントがストレージエンドポイントの場合、宛先の詳細ページでコンテナー名の入力が求められます。データフローエンドポイントが MQTT エンドポイントである場合、宛先の詳細ページでトピックの入力などが求められます。

エンドポイントの種類を指定して MQTT トピックを入力するようユーザーに求める操作エクスペリエンスを示すスクリーンショット。

構文は、すべてのデータフローエンドポイントで同じです:

destinationSettings: {
  endpointRef: "<CUSTOM_ENDPOINT_NAME>"
  dataDestination: '<TOPIC_OR_TABLE>'
}

たとえば、静的 MQTT トピックのローカル MQTT ブローカーにデータを送り返すには、次の構成を使用します。

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'example-topic'
}

または、カスタムイベントハブエンドポイントがある場合、構成は次のようになります。

destinationSettings: {
  endpointRef: 'my-eh-endpoint'
  dataDestination: 'individual-event-hub'
}

ストレージエンドポイントを宛先として使用する別の例:

destinationSettings: {
  endpointRef: 'my-adls-endpoint'
  dataDestination: 'my-container'
}

構文は、すべてのデータフローエンドポイントで同じです:

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE>

たとえば、静的 MQTT トピックのローカル MQTT ブローカーにデータを送り返すには、次の構成を使用します。

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

または、カスタムイベントハブエンドポイントがある場合、構成は次のようになります。

destinationSettings:
  endpointRef: my-eh-endpoint
  dataDestination: individual-event-hub

ストレージエンドポイントを宛先として使用する別の例:

destinationSettings:
  endpointRef: my-adls-endpoint
  dataDestination: my-container

例

次の例は、ソースと宛先に MQTT エンドポイントを使用したデータフロー構成です。ソースは、MQTT トピック azure-iot-operations/data/thermostat からのデータをフィルターします。変換によって温度が華氏に変換され、温度に湿度を掛けた値が 100,000 未満のデータにフィルターされます。宛先が MQTT トピック factory にデータを送信します。

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

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

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

// Pointer to the default data flow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent data flow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // Use the default MQTT endpoint as the source
          endpointRef: defaultDataflowEndpoint.name
          // Filter the data from the MQTT topic azure-iot-operations/data/thermostat
          dataSources: [
            'azure-iot-operations/data/thermostat'
          ]
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // Filter the data where temperature * "Tag 10" < 100000
          filter: [
            {
              inputs: [
                'temperature.Value'
                '"Tag 10".Value'
              ]
              expression: '$1 * $2 < 100000'
            }
          ]
          map: [
            // Passthrough all values by default
            {
              inputs: [
                '*'
              ]
              output: '*'
            }
            // Convert temperature to Fahrenheit and output it to TemperatureF
            {
              inputs: [
                'temperature.Value'
              ]
              output: 'TemperatureF'
              expression: 'cToF($1)'
            }
          // Extract the "Tag 10" value and output it to Humidity
            {
              inputs: [
                '"Tag 10".Value'
              ]
              output: 'Humidity'
            }
          ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // Use the default MQTT endpoint as the destination
          endpointRef: defaultDataflowEndpoint.name
          // Send the data to the MQTT topic factory
          dataDestination: 'factory'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow profile
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # Use the default MQTT endpoint as the source
        endpointRef: default
        # Filter the data from the MQTT topic azure-iot-operations/data/thermostat
        dataSources:
          - azure-iot-operations/data/thermostat
    # Transformation optional
    - operationType: builtInTransformation
      builtInTransformationSettings:
        # Filter the data where temperature * "Tag 10" < 100000
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: '$1 * $2 < 100000'
        map:
          # Passthrough all values by default
          - inputs:
              - '*'
            output: '*'
          # Convert temperature to Fahrenheit and output it to TemperatureF
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          # Extract the "Tag 10" value and output it to Humidity
          - inputs:
              - '"Tag 10".Value'
            output: 'Humidity'
    - operationType: Destination
      destinationSettings:
        # Use the default MQTT endpoint as the destination
        endpointRef: default
        # Send the data to the MQTT topic factory
        dataDestination: factory

データフロー構成のその他の例については、Azure REST API (データフロー) と Bicep のクイックスタートに関するページを参照してください。

データフローが機能していることを確認する

「チュートリアル: Azure Event Grid への双方向 MQTT ブリッジ」に従って、データフローが機能していることを確認します。

データフロー構成のエクスポート

データフロー構成をエクスポートするには、操作エクスペリエンスを使用するか、Dataflow カスタムリソースをエクスポートします。

エクスポートするデータフローを選択し、ツールバーから [エクスポート] を選びます。

操作エクスペリエンスを使用してデータフローをエクスポートするスクリーンショット。

kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml

適切なデータフロー構成

データフローが期待どおりに動作していることを確認するには、次のことを確かめします:

既定の MQTT データフローエンドポイントは、ソースまたは宛先のいずれかとして使用される必要があります。
データフロープロファイルが存在し、データフロー構成で参照されます。
ソースは、MQTT エンドポイント、Kafka エンドポイント、資産のいずれかです。ストレージの種類のエンドポイントは、ソースとして使用できません。
Event Grid をソースとして使用する場合、Event Grid MQTT ブローカーは共有サブスクリプションをサポートしていないため、データフロープロファイルインスタンス数は 1 に設定されます。
Event Hubs をソースとして使用する場合、名前空間内の各イベントハブは個別の Kafka トピックであり、データソースとして指定する必要があります。
変換を使用する場合は、特殊文字の適切なエスケープを含む、適切な構文で構成します。
ストレージの種類のエンドポイントを宛先として使用する場合は、スキーマを指定します。

次の方法で共有

Azure IoT Operations でデータフローを構成する

前提条件

データフロープロファイル

データフローエンドポイント

作業の開始

ソース

オプション 1: 既定のメッセージブローカーエンドポイントをソースとして使用する

オプション 2: 資産をソースとして使用する

オプション 3: カスタム MQTT または Kafka データフローエンドポイントをソースとして使用する

データソースを構成する (MQTT または Kafka のトピック)

MQTT のトピック

共有サブスクリプション

Kafka トピック

ソーススキーマを指定する

変換

エンリッチ: 参照データを追加する

フィルター: 条件に基づいてデータをフィルター処理する

マップ: あるフィールドから別のフィールドにデータを移動する

Compute

名前の変更

新しいプロパティ

スキーマに従ってデータをシリアル化する

宛先

データの宛先 (トピック、コンテナー、またはテーブル) を構成する

例

データフローが機能していることを確認する

データフロー構成のエクスポート

適切なデータフロー構成

次のステップ

フィードバック

その他のリソース

次の方法で共有

Azure IoT Operations でデータ フローを構成する

前提条件

データ フロー プロファイル

データ フロー エンドポイント

作業の開始

ソース

オプション 1: 既定のメッセージ ブローカー エンドポイントをソースとして使用する

オプション 2: 資産をソースとして使用する

オプション 3: カスタム MQTT または Kafka データ フロー エンドポイントをソースとして使用する

データ ソースを構成する (MQTT または Kafka のトピック)

MQTT のトピック

共有サブスクリプション

Kafka トピック

ソース スキーマを指定する

変換

エンリッチ: 参照データを追加する

フィルター: 条件に基づいてデータをフィルター処理する

マップ: あるフィールドから別のフィールドにデータを移動する

Compute

名前の変更

新しいプロパティ

スキーマに従ってデータをシリアル化する

宛先

データの宛先 (トピック、コンテナー、またはテーブル) を構成する

例

データ フローが機能していることを確認する

データ フロー構成のエクスポート

適切なデータ フロー構成

次のステップ

フィードバック

その他のリソース

Azure IoT Operations でデータフローを構成する

データフロープロファイル

データフローエンドポイント

オプション 1: 既定のメッセージブローカーエンドポイントをソースとして使用する

オプション 3: カスタム MQTT または Kafka データフローエンドポイントをソースとして使用する

データソースを構成する (MQTT または Kafka のトピック)

ソーススキーマを指定する

データフローが機能していることを確認する

データフロー構成のエクスポート

適切なデータフロー構成