データフローを作成する

データフローとは、データがソースから宛先までたどるパスであり、必要に応じて変換が行われます。 データフローを構成するには、Azure IoT Operations ポータルを使用するか、Dataflow カスタム リソースを作成します。 データフローを作成する前に、データ ソースと宛先のデータフロー エンドポイントを構成する必要があります。

MQTT ソース エンドポイント、変換、Kafka 宛先エンドポイントを含むデータフロー構成の例を次に示します。

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
spec:
  profileRef: my-dataflow-profile
  mode: enabled
  operations:
    - operationType: source
      name: my-source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
        serializationFormat: json
    - operationType: builtInTransformation
      name: my-transformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
        serializationFormat: json
    - operationType: destination
      name: my-destination
      destinationSettings:
        endpointRef: kafka
        dataDestination: factory

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

ソースの構成

データフローのソースを構成するには、エンドポイント参照とデータ ソースを指定します。 エンドポイントのデータ ソースの一覧を指定できます。 たとえば、MQTT や Kafka トピックです。 ソース エンドポイントとデータ ソースを含むデータフロー構成の例を次に示します。

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: mq-to-kafka
  namespace: azure-iot-operations
spec:
  profileRef: example-dataflow
  operations:
    - operationType: source
      sourceSettings:
        endpointRef: mq-source
        dataSources:
        - azure-iot-operations/data/thermostat

変換を構成する

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

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    builtInTransformationSettings:
      datasets:
        # ...
      filter:
        # ...
      map:
        # ...

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

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

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

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    builtInTransformationSettings:
      datasets:
      - key: assetDataset
        inputs:
          - $source.deviceId # ------------- $1
          - $context(assetDataset).asset # - $2
        expression: $1 == $2

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

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

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

DSS セット ツール サンプルを使用して、サンプル データを分散状態ストア (DSS) に読み込むことができます。

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

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

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

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

spec:
  operations:
  - operationType: builtInTransformation
    name: transform1
    builtInTransformationSettings:
      filter:
        - inputs:
          - temperature ? $last # - $1
          expression: "$1 > 20"

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

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

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

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

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

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

コピー先を構成する

データフローの宛先を構成するには、宛先のエンドポイントとパス (トピックまたはテーブル) を指定する必要があります。

宛先エンドポイントの参照を構成する

宛先エンドポイントを構成するには、ID とエンドポイントの参照を指定する必要があります。

spec:
  operations:
  - operationType: destination
    name: destination1
    destinationSettings:
      endpointRef: eventgrid

宛先のパスを構成する

エンドポイントを取得したら、宛先のパスを構成できます。 宛先が MQTT または Kafka エンドポイントの場合は、パスを使用してトピックを指定します。

- operationType: destination
  destinationSettings:
    endpointRef: eventgrid
    dataDestination: factory

Microsoft Fabric などのストレージ エンドポイントの場合は、パスを使用してテーブル名を指定します。

- operationType: destination
  destinationSettings:
    endpointRef: adls
    dataDestination: telemetryTable