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

操作エクスペリエンス ポータルでデータフローを作成するには、[データフロー]>[データフローを作成する] を選択します。

データフロー構成の全体的な構造は、次のとおりです。

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

ソースとして資産を使用する

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

1. [ソースの詳細] で、[資産] を選択します。

2. ソース エンドポイントとして使用する資産を選択します。

3. [続行] を選択します。

   選択した資産のデータポイントの一覧が表示されます。

   

4. [適用] を選択して、資産をソース エンドポイントとして使用します。

資産のソースとしての構成は、操作エクスペリエンス ポータルでのみ可能です。

1. [ソースの詳細] で、[MQTT] を選択します。

2. 受信メッセージをリッスンする MQTT トピックを入力します。

3. ドロップダウン リストからメッセージ スキーマを選択するか、新しいスキーマをアップロードします。 ソース データに省略可能なフィールドや、さまざまな型を持つフィールドがある場合は、一貫性を保つために逆シリアル化スキーマを指定します。 たとえば、データに、すべてのメッセージに存在するわけではないフィールドがあるとします。 スキーマがないと、これらのフィールドは空の値を持つため、変換ではこれらを処理できません。 スキーマがあれば、既定値を指定することも、フィールドを無視することもできます。

   

4. 適用を選択します。

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

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

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

データを逆シリアル化するスキーマを指定する

ソース データに省略可能なフィールドや、さまざまな型を持つフィールドがある場合は、一貫性を保つために逆シリアル化スキーマを指定します。 たとえば、データに、すべてのメッセージに存在するわけではないフィールドがあるとします。 スキーマがないと、これらのフィールドは空の値を持つため、変換ではこれらを処理できません。 スキーマがあれば、既定値を指定することも、フィールドを無視することもできます。

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

スキーマを指定するには、ファイルを作成し、スキーマ レジストリに格納します。

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

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

共有サブスクリプション

MQTT ソースで共有サブスクリプションを使用するには、$shared/<subscription-group>/<topic> の形式で共有サブスクリプション トピックを指定します。

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

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

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

現在、エンリッチ操作は、操作エクスペリエンス ポータルでは使用できません。

たとえば、ソース データの 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 フィールドを持つソースのデータには、filter と map のステージで使用できる location と manufacturer のフィールドがあります。

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

2. データセットに含めるデータポイントを選択します。

3. フィルター条件と説明を追加します。

   

4. 適用を選択します。

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

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

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

操作エクスペリエンス ポータルでは、現在、コンピューティング変換を使用したマッピングがサポートされています。

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

2. 必須フィールドと式を入力します。

   

3. 適用を選択します。

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

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

宛先データフロー エンドポイントを追加するときに、出力スキーマを指定します。

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

スキーマを指定するには、スキーマ定義を使用して Schema カスタム リソースを作成します。

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

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

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

   

2. [続行] を選択して、宛先を構成します。

3. 宛先の種類に基づいてマッピングの詳細を追加します。

たとえば、静的 MQTT トピックと先ほど作成した MQTT エンドポイントを使用して宛先を構成するには、次の構成を使用します。

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Microsoft Fabric などのストレージ エンドポイントを作成済みの場合は、データ宛先フィールドを使用してテーブルまたはコンテナー名を指定します。

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

例

次の例は、ソースと宛先に MQTT エンドポイントを使用したデータフロー構成です。 ソースは、MQTT トピック thermostats/+/telemetry/temperature/# および humidifiers/+/telemetry/humidity/# からのデータをフィルターします。 変換によって温度が華氏に変換され、温度が 100000 未満のデータにフィルターされます。 宛先が MQTT トピック factory にデータを送信します。

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      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'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

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

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