データ プロセッサ パイプラインから Microsoft Fabric にデータを送信する

Fabric Lakehouse の変換先を使用して、Azure IoT Data Processor プレビュー パイプラインから Microsoft Fabric のレイクハウスにデータを書き込みます。 変換先ステージで Parquet ファイルが Lakehouse に書き込まれ、データをデルタ テーブルで閲覧できます。 変換先ステージは、メッセージを Microsoft Fabric に送信する前にバッチ処理します。

前提条件

Microsoft Fabric の変換先パイプライン ステージを構成して使用するには、次のものが必要です。

• データ プロセッサのデプロイされたインスタンス。
• Microsoft Fabric サブスクリプション。 または、無料の Microsoft Fabric 試用版キャパシティにサインアップします。
• Microsoft Fabric の Lakehouse。

Microsoft Fabric をセットアップする

データ パイプラインから Microsoft Fabric に書き込む前に、パイプラインからレイクハウスへのアクセス権を付与する必要があります。 サービス プリンシパルまたはマネージド ID を使用して、パイプラインを認証できます。 マネージド ID を使用する利点は、サービス プリンシパルのライフサイクルを管理する必要がないことです。 マネージド ID は Azure によって自動的に管理され、割り当てられているリソースのライフサイクルに関連付けられます。

レイクハウスへのサービス プリンシパルまたはマネージド ID アクセスを構成する前に、サービス プリンシパル認証を有効にします。

CLUSTER_NAME=<Your connected cluster name>
RESOURCE_GROUP=<The resource group where your connected cluster is installed>
SUBSCRIPTION=<The subscription where your connected cluster is installed>
EXTENSION_NAME=processor

OBJECT_ID=$(az k8s-extension show --name $EXTENSION_NAME --cluster-name $CLUSTER_NAME --resource-group $RESOURCE_GROUP --cluster-type connectedClusters --query identity.principalId -o tsv --subscription $SUBSCRIPTION)
echo "App ID:    " `az ad sp show --query appId --id $OBJECT_ID -o tsv`
echo "Tenant ID: " `az account show --query tenantId -o tsv`

変換先ステージを構成する

Fabric Lakehouse の変換先ステージの JSON 構成で、ステージの詳細を定義します。 ステージを作成するには、フォーム ベースの UI を使用するか、[詳細設定] タブで次の JSON 構成を指定します。

¹ファイル パス: Microsoft Fabric にファイルを書き込むには、ファイル パスが必要です。 テンプレート を使用して、ファイル パスを構成できます。 ファイル パスには、次のコンポーネントを任意の順序で含む必要があります。

• instanceId
• pipelineId
• partitionId
• YYYY
• MM
• DD
• HH
• mm
• fileNumber

ファイル名は、fileNumberで示される増分整数値です。 システムにファイルの種類を認識させる場合は、必ずファイル拡張子を含めるようにしてください。

²バッチ処理: データを Microsoft Fabric に書き込む場合はバッチ処理が必須です。 変換先ステージは、構成可能な時間間隔でメッセージをバッチ処理します。

バッチ処理間隔を構成しない場合、ステージでは既定として 60 秒が使用されます。

³型: データ プロセッサでは、デルタ形式を使用して Microsoft Fabric に書き込みます。 データ プロセッサは、decimal と timestamp without time zone を除くすべてのデルタ プリミティブ データ型をサポートします。

Microsoft Fabric ですべての日付と時刻が正しく表されるようにするには、プロパティの値が有効な RFC 3339 文字列であり、データ型が date または timestamp であることを確認します。

¹認証: 現在、変換先ステージでは、Microsoft Fabric に接続するときにサービス プリンシパル ベースの認証またはマネージド ID がサポートされています。

サービス プリンシパル ベースの認証

サービス プリンシパル ベースの認証を構成するには、次の値を指定します。 サービス プリンシパルを作成し、クラスターにシークレット参照を追加したときに、これらの値をメモしました。

サンプル構成

次の JSON 例は、データベースの quickstart テーブルにメッセージ全体を書き込む Microsoft Fabric Lakehouse の 変換先ステージの完全な構成を示しています。

{
    "displayName": "Fabric Lakehouse - 520f54",
    "type": "output/fabric@v1",
    "viewOptions": {
        "position": {
            "x": 0,
            "y": 784
        }
    },
    "workspace": "workspaceId",
    "lakehouse": "lakehouseId",
    "table": "quickstart",
    "columns": [
        {
            "name": "Timestamp",
            "type": "timestamp",
            "path": ".Timestamp"
        },
        {
            "name": "AssetName",
            "type": "string",
            "path": ".assetname"
        },
        {
            "name": "Customer",
            "type": "string",
            "path": ".Customer"
        },
        {
            "name": "Batch",
            "type": "integer",
            "path": ".Batch"
        },
        {
            "name": "CurrentTemperature",
            "type": "float",
            "path": ".CurrentTemperature"
        },
        {
            "name": "LastKnownTemperature",
            "type": "float",
            "path": ".LastKnownTemperature"
        },
        {
            "name": "Pressure",
            "type": "float",
            "path": ".Pressure"
        },
        {
            "name": "IsSpare",
            "type": "boolean",
            "path": ".IsSpare"
        }
    ],
    "authentication": {
        "type": "servicePrincipal",
        "tenantId": "tenantId",
        "clientId": "clientId",
        "clientSecret": "secretReference"
    },
    "batch": {
        "time": "5s",
        "path": ".payload"
    },
    "retry": {
        "type": "fixed",
        "interval": "20s",
        "maxRetries": 4
    }
}

この構成は次のことを定義します。

• メッセージは 5 秒間でバッチ処理されます。
• バッチ パス .payload を使用して、列のデータを検索します。

例

次の例は、Microsoft Fabric の変換先ステージへの入力メッセージのサンプルを示しています:

{
  "payload": {
    "Batch": 102,
    "CurrentTemperature": 7109,
    "Customer": "Contoso",
    "Equipment": "Boiler",
    "IsSpare": true,
    "LastKnownTemperature": 7109,
    "Location": "Seattle",
    "Pressure": 7109,
    "Timestamp": "2023-08-10T00:54:58.6572007Z",
    "assetName": "oven"
  },
  "qos": 0,
  "systemProperties": {
    "partitionId": 0,
    "partitionKey": "quickstart",
    "timestamp": "2023-11-06T23:42:51.004Z"
  },
  "topic": "quickstart"
}

関連するコンテンツ

• Azure Data Explorer にデータを送信する
• gRPC エンドポイントにデータを送信する
• HTTP エンドポイントにデータを送信する
• MQTT ブローカーにデータを発行する
• 参照データ ストアにデータを送信する