次の方法で共有

チュートリアル: OPC UA サーバーから Azure Data Lake Storage Gen 2 にデータを送信する

  • [アーティクル]

クイックスタートでは、データを Azure IoT Operations から Event Hubs に送信し、さらに EventStreams 経由で Microsoft Fabric に送信するデータフローを作成しました。

ただし、Event Hubs を使用せずにストレージ エンドポイントにデータを直接送信することもできます。 この方法では、データを表す Delta Lake スキーマを作成し、そのスキーマを Azure IoT Operations にアップロードしてから、OPC UA サーバーからデータを読み取ってストレージ エンドポイントに書き込むデータフローを作成する必要があります。

このチュートリアルは、クイックスタートのセットアップに基づいて、データを Azure Data Lake Storage Gen 2 に分岐させる方法を示します。 この方法を使用すると、スケーラブルで安全なデータ レイクにデータを直接格納でき、さらに分析と処理に使用できます。

前提条件

クイックスタート の 2 番目の手順 (OPC UA サーバーから Azure IoT Operations MQTT ブローカーにデータを送信) を完了します。 Event Hubs にデータが表示されていることを確認します。

Data Lake Storage 機能を備えたストレージ アカウントを作成する

まず、手順に従って、Data Lake Storage Gen 2 機能を持つストレージ アカウントを作成します。

  • ストレージ アカウントのために覚えやすい一意の名前を選択します。これは次の手順で必要になるためです。
  • 最良の結果を得るには、Azure IoT Operations が実行されている Kubernetes クラスターに近い場所を使用します。
  • 作成プロセス中に、[階層型名前空間] の設定を有効にします。 Azure IoT Operations がストレージ アカウントに書き込みを行うには、この設定が必要です。
  • その他の設定は、既定値のままにしておきます。

[確認] ステップで設定を確認し、[作成] を選んでストレージ アカウントを作成します。

階層型名前空間が有効になっている、ストレージ アカウントの作成の [確認] ページを示すスクリーンショット。

Azure IoT Operations の拡張機能名を取得する

Azure portal で、クイックスタートで作成した Azure IoT Operations インスタンスを探します。 [概要] ブレードで、[Arc 拡張機能] セクションを探し、拡張機能の名前を確認します。 azure-iot-operations-xxxxx のようになります。

拡張機能名を見つける場所を示す Azure portal のスクリーンショット。

この拡張機能名は、次の手順でストレージ アカウントにアクセス許可を割り当てるために使用されます。

ストレージ アカウントに書き込むためのアクセス許可を Azure IoT Operations に割り当てる

まず、ストレージ アカウントで、[アクセス制御 (IAM)] ブレードに移動し、[+ ロールの割り当ての追加] を選択します。 [ロールの割り当ての追加] ブレードで、[ストレージ BLOB データ共同作成者] ロールを探して選択します。

ストレージ BLOB データ共同作成者ロールを選択する方法を示すスクリーンショット。

その後、[次へ] を選択して、[メンバー] セクションに移動します。

次に、[メンバーの選択] を選択し、[選択] ボックスで、azure-iot-operations-xxxxx という名前の Azure IoT Operations Arc 拡張機能のマネージド ID を検索して選択します。

Arc 拡張機能名を検索し、IAM で選択する方法を示すスクリーンショット。

[確認と割り当て] で割り当てを完了します。

ストレージ アカウントにコンテナーを作成する

ストレージ アカウントの [コンテナー] ブレードに移動し、[+ コンテナー] を選択します。 このチュートリアルでは、コンテナーに aiotutorialという名前を付けます。 [作成] を選択して、コンテナーを作成します。

Azure portal でのストレージ コンテナーの作成方法を示すスクリーンショット。

スキーマ レジストリの名前と名前空間を取得する

Azure IoT Operations にスキーマをアップロードするには、スキーマ レジストリの名前と名前空間を知っている必要があります。 この情報は、Azure CLI を使って取得できます。

次のコマンドを実行して、スキーマ レジストリの名前と名前空間を取得します。 プレースホルダーを実際の値に置き換えます。

az iot ops schema registry list -g <RESOURCE_GROUP> --query "[0].{name: name, namespace: properties.namespace}" -o tsv

出力は次のように表示されます。

<REGISTRY_NAME>    <SCHEMA_NAMESPACE>

次の手順用に値を保存します。

Azure IoT Operations にスキーマをアップロードする

クイックスタートでは、オーブン資産から取得されるデータは次のようになります。

{
  "Temperature": {
    "SourceTimestamp": "2024-11-15T21:40:28.5062427Z",
    "Value": 6416
  },
  "FillWeight": {
    "SourceTimestamp": "2024-11-15T21:40:28.5063811Z",
    "Value": 6416
  },
  "EnergyUse": {
    "SourceTimestamp": "2024-11-15T21:40:28.506383Z",
    "Value": 6416
  }
}

Delta Lake に必要なスキーマ形式は、Delta Lake スキーマのシリアル化形式に従う JSON オブジェクトです。 スキーマによって、各フィールドの型やプロパティなど、データの構造を定義する必要があります。 スキーマ形式の詳細については、Delta Lake スキーマのシリアル化形式のドキュメントを参照してください。

ヒント

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

このチュートリアルでは、データのスキーマは次のようになります。

{
  "$schema": "Delta/1.0",
  "type": "object",
  "properties": {
    "type": "struct",
    "fields": [
      {
        "name": "Temperature",
        "type": {
          "type": "struct",
          "fields": [
            {
              "name": "SourceTimestamp",
              "type": "timestamp",
              "nullable": false,
              "metadata": {}
            },
            {
              "name": "Value",
              "type": "integer",
              "nullable": false,
              "metadata": {}
            }
          ]
        },
        "nullable": false,
        "metadata": {}
      },
      {
        "name": "FillWeight",
        "type": {
          "type": "struct",
          "fields": [
            {
              "name": "SourceTimestamp",
              "type": "timestamp",
              "nullable": false,
              "metadata": {}
            },
            {
              "name": "Value",
              "type": "integer",
              "nullable": false,
              "metadata": {}
            }
          ]
        },
        "nullable": false,
        "metadata": {}
      },
      {
        "name": "EnergyUse",
        "type": {
          "type": "struct",
          "fields": [
            {
              "name": "SourceTimestamp",
              "type": "timestamp",
              "nullable": false,
              "metadata": {}
            },
            {
              "name": "Value",
              "type": "integer",
              "nullable": false,
              "metadata": {}
            }
          ]
        },
        "nullable": false,
        "metadata": {}
      }
    ]
  }
}

これを opcua-schema.json という名前のファイルとして保存します。

次に、Azure CLI を使用して Azure IoT Operations にスキーマをアップロードします。 プレースホルダーを実際の値に置き換えます。

az iot ops schema create -n opcua-schema -g <RESOURCE_GROUP> --registry <REGISTRY_NAME> --format delta --type message --version-content opcua-schema.json --ver 1

これにより、バージョン 1 の Azure IoT Operations レジストリに opcua-schema という名前のスキーマが作成されます。

スキーマがアップロードされたことを確認するには、Azure CLI を使用してスキーマのバージョンを一覧表示します。

az iot ops schema version list -g <RESOURCE_GROUP> --schema opcua-schema --registry <REGISTRY_NAME>

データフロー エンドポイントを作成する

データフロー エンドポイントは、データが送信される宛先です。 この場合、データは Azure Data Lake Storage Gen 2 に送信されます。 認証方法はシステム割り当てマネージド ID です。ストレージ アカウントに書き込むための適切なアクセス許可をこれに設定します。

Bicep を使用してデータフロー エンドポイントを作成します。 プレースホルダーを実際の値に置き換えます。

// Replace with your values
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'

// Tutorial specific values
param endpointName string = 'adls-gen2-endpoint'
param host string = 'https://<ACCOUNT>.blob.core.windows.net'

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

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

resource adlsGen2Endpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
  parent: aioInstance
  name: endpointName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'DataLakeStorage'
    dataLakeStorageSettings: {
      host: host
      authentication: {
        method: 'SystemAssignedManagedIdentity'
        systemAssignedManagedIdentitySettings: {}
      }
    }
  }
}

ファイルを adls-gen2-endpoint.bicep として保存し、Azure CLI を使用してデプロイする

az deployment group create -g <RESOURCE_GROUP> --template-file adls-gen2-endpoint.bicep

データフローを作成する

Azure Data Lake Storage Gen 2 にデータを送信するには、OPC UA サーバーからデータを読み取り、ストレージ アカウントに書き込むデータフローを作成する必要があります。 この場合、変換は必要ないため、データはそのまま書き込まれます。

Bicep を使用してデータフローを作成します。 プレースホルダーを実際の値に置き換えます。

// Replace with your values
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param schemaNamespace string = '<SCHEMA_NAMESPACE>'

// Tutorial specific values
param schema string = 'opcua-schema'
param schemaVersion string = '1'
param dataflowName string = 'tutorial-adls-gen2'
param assetName string = 'oven'
param endpointName string = 'adls-gen2-endpoint'
param containerName string = 'aiotutorial'
param serialFormat string = 'Delta'

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 dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

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

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

resource asset 'Microsoft.DeviceRegistry/assets@2024-11-01' existing = {
  name: assetName
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent dataflow 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: {
          endpointRef: defaultDataflowEndpoint.name
          assetRef: asset.name
          dataSources: ['azure-iot-operations/data/${assetName}']
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          serializationFormat: serialFormat
          schemaRef: 'aio-sr://${schemaNamespace}/${schema}:${schemaVersion}'
          map: [
            {
              type: 'PassThrough'
              inputs: [
                '*'
              ]
              output: '*'
            }
         ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          endpointRef: adlsEndpoint.name
          dataDestination: containerName
        }
      }
    ]
  }
}

ファイルを adls-gen2-dataflow.bicep として保存し、Azure CLI を使用してデプロイする

az deployment group create -g <RESOURCE_GROUP> --template-file adls-gen2-dataflow.bicep

Azure Data Lake Storage Gen 2 でデータを確認する

ストレージ アカウントの [コンテナー] ブレードに移動し、作成したコンテナー aiotutorial を選択します。 aiotutorial という名前のフォルダーが表示されます。その中に、OPC UA サーバーのデータを含む Parquet ファイルが表示されます。 ファイル名の形式は、part-00001-44686130-347f-4c2c-81c8-eb891601ef98-c000.snappy.parquet です。

コンテナー内のファイルが表示されている Azure portal のスクリーンショット。

ファイルの内容を表示するには、各ファイルを選択し、[編集] を選択します。

parquet ファイルそのものが表示された Azure portal のスクリーンショット。

内容は Azure portal では正しくレンダリングされませんが、ファイルをダウンロードして、Parquet Viewer などのツールで開くことができます。

関連するコンテンツ