Unity カタログにデータを提供するカスタムモデルを保持する

Important

この機能はベータ版です。すべての顧客に対して自動的に有効になるわけではありません。機能は変更される可能性があります。アクセスを要求するには、Azure Databricks アカウントチームにお問い合わせください。

エンドポイントを提供するカスタムモデルから Unity カタログテーブルに OpenTelemetry ログ、トレース、メトリックを保持するようにエンドポイントテレメトリを構成する方法について説明します。永続化されたテレメトリデータを使用して、根本原因分析を実行し、エンドポイントの正常性を監視し、標準の SQL クエリでコンプライアンス要件を満たします。

必要条件

ワークスペースは、Unity Catalog に対して有効にする必要があります。既定のストレージ (Arclight) はサポートされていません。
ログが格納される対象の Unity カタログカタログとスキーマに対する USE CATALOG、 USE SCHEMA、 CREATE TABLE、および MODIFY のアクセス許可が必要です。
既存のカスタムモデル提供用エンドポイント、またはそれを作成するための権限。
ワークスペースは、サポートされているリージョンに存在する必要があります。
- canadacentral
- westus
- westus2
- southcentralus
- eastus
- eastus2
- centralus
- northcentralus
- swedencentral
- westeurope
- northeurope
- uksouth
- australiaeast
- southeastasia

手順 1: モデルコードをインストルメント化する

モデルコードにインストルメンテーションを追加してテレメトリをキャプチャします。

アプリケーションログをモデルに追加します。エンドポイントテレメトリは、標準の Python logging 出力を自動的にキャプチャします。基本的なログ記録には OpenTelemetry SDK インストルメンテーションは必要ありません。

import logging

class MyCustomModel(mlflow.pyfunc.PythonModel):
    def predict(self, context, model_input):
        # This log will be persisted to the <prefix>_otel_logs table
        logging.warning("Received inference request")

        try:
            # Your model logic here
            result = model_input * 2
            return result
        except Exception as e:
            # Error logs are also captured with severity 'ERROR'
            logging.error(f"Inference failed: {e}")
            raise e

ルートログレベルは WARNING に設定されます。ログレベルを変更するには、「トラブルシューティング」を参照してください。

(省略可能)OpenTelemetry を使用してカスタムメトリックとトレースをインストルメント化します。基本的なログ記録以外のカスタムメトリックとトレースをキャプチャするには、OpenTelemetry SDK インストルメンテーションをモデルに追加します。カウンターの作成、スパンの記録、カスタム属性のアタッチ方法を示す完全な例については、次のセクションを展開してください。

例: OpenTelemetry を使用したカスタムメトリック、スパン、モデルのログ記録

注

モデルのシリアル化の制限により、 %%writefile return_input_model.pyを使用して次に示すように、エラーを回避するために、ログ記録の前にモデルを別のファイルに書き込む必要があります。

%%writefile return_input_model.py
import os

import mlflow
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.metrics import get_meter, set_meter_provider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import get_tracer, set_tracer_provider

# highlight-start
# ---- OTel initialization (per-worker) ----
resource = Resource.create({
    "worker.pid": str(os.getpid()),
})

otlp_trace_exporter = OTLPSpanExporter()
tracer_provider = TracerProvider(resource=resource)
tracer_provider.add_span_processor(BatchSpanProcessor(otlp_trace_exporter))
set_tracer_provider(tracer_provider)

otlp_metric_exporter = OTLPMetricExporter()
metric_reader = PeriodicExportingMetricReader(otlp_metric_exporter)
meter_provider = MeterProvider(metric_readers=[metric_reader], resource=resource)
set_meter_provider(meter_provider)

_tracer = get_tracer(__name__)
_meter = get_meter(__name__)
_prediction_counter = _meter.create_counter(
    name="prediction_count",
    description="Number of predictions made",
    unit="1"
)
# highlight-end


class ReturnInputModel(mlflow.pyfunc.PythonModel):
    def load_context(self, context):
        # highlight-start
        self.tracer = _tracer
        self.prediction_counter = _prediction_counter
        # highlight-end

    def predict(self, context, model_input):
        # highlight-next-line
        with self.tracer.start_as_current_span("ReturnInputModel.predict") as span:
            # highlight-next-line
            span.set_attribute("input_shape", str(model_input.shape))
            # highlight-next-line
            span.set_attribute("input_columns", str(list(model_input.columns)))
            # highlight-next-line
            self.prediction_counter.add(1)
            return model_input

mlflow.models.set_model(ReturnInputModel())

モデルをログに記録して登録します。

import pandas as pd
import mlflow
from mlflow.models import infer_signature

# Prepare tabular input/output for signature (pyfunc expects DataFrame)
input_df = pd.DataFrame({"inputs": ["hello world"]})
output_df = input_df.copy()  # model returns input unchanged

# Log the model with OpenTelemetry dependencies (using code-based logging to avoid serialization issues)
with mlflow.start_run():
    signature = infer_signature(input_df, output_df)

    model_info = mlflow.pyfunc.log_model(
        name="model",
        python_model="return_input_model.py",
        signature=signature,
        input_example=input_df,
        pip_requirements=[
            "mlflow==3.1",
            # highlight-next-line
            "opentelemetry-sdk",
            # highlight-next-line
            "opentelemetry-exporter-otlp-proto-http",
        ],
    )

# Register with serverless optimized deployment environment packing
# Use Unity Catalog name: catalog.schema.model_name
registered = mlflow.register_model(
    model_info.model_uri,
    MODEL_NAME,
    env_pack="databricks_model_serving"
)

手順 2: Unity カタログの宛先を準備する

エンドポイントを作成する前に、テレメトリデータを受け取る準備ができているカタログとスキーマがあることを確認します。 Azure Databricks では、必要なテーブルがまだ存在しない場合は、このスキーマに自動的に作成されます。

カタログエクスプローラーで、使用するカタログとスキーマ (たとえば、 my_catalog.observability) に移動します。

手順 3: エンドポイントテレメトリを有効にする

新しいエンドポイントを作成するときにテレメトリを有効にするか、既存のエンドポイントに追加できます。

新規エンドポイント

UI でテレメトリを有効にするには:

左側のサイドバーで [ サービス ] に移動します。
[ サービスエンドポイントの作成] をクリックします。
[ エンドポイントテレメトリ ] セクション (プレビューとマークされています) で、構成オプションを展開します。
Unity カタログの場所: 手順 2 で準備した宛先 カタログ と スキーマ を選択します。
(省略可能) テーブルプレフィックス: 生成されたテーブルのプレフィックスを入力します。空白のままにした場合、プレフィックスはありません。テーブルには、 <prefix>_otel_logs、 <prefix>_otel_spans、および <prefix>_otel_metricsという名前が付けられます。
エンドポイント構成の残りの部分 (モデルの選択、コンピューティング設定) を完了し、[ 作成] をクリックします。

API を使用してこれを行うには:

API を使用してテレメトリを有効にする

curl -X POST -H "Authorization: Bearer <your-token>" \
https://<workspace-url>/api/2.0/serving-endpoints \
-d '{
  "name": "my-custom-logging-endpoint",
  "config": {
    "served_entities": [
      {
        "name": "my-model",
        "entity_name": "my-model",
        "entity_version": "1",
        "workload_size": "Small",
        "scale_to_zero_enabled": true
      }
    ],
    "telemetry_config": {
      "table_names": {
        "logs_table": "my_catalog.observability.custom_endpoint_logs",
        "metrics_table": "my_catalog.observability.custom_endpoint_metrics",
        "traces_table": "my_catalog.observability.custom_endpoint_spans"
      }
    }
  }
}'

既存のエンドポイント

注

更新すると、新しいデプロイがトリガーされます。変更は、デプロイが完了すると有効になります。

UI でテレメトリを有効にするには:

エンドポイントビューページの右側のパネルの [ エンドポイントテレメトリ ] セクションで、[ 追加] をクリックします。
Unity カタログの場所: 手順 2 で準備した宛先 カタログ と スキーマ を選択します。
(省略可能) テーブルプレフィックス: 生成されたテーブルのプレフィックスを入力します。空白のままにした場合、プレフィックスはありません。テーブルには、 <prefix>_otel_logs、 <prefix>_otel_spans、および <prefix>_otel_metricsという名前が付けられます。
[Update] をクリックします。

手順 4: テレメトリデータを確認してクエリを実行する

エンドポイントがトラフィックを受信すると、テレメトリデータは構成された Unity カタログテーブルにストリームされます。

カタログエクスプローラーまたは SQL エディターに移動します。
構成したスキーマで <prefix>_otel_logs という名前のテーブルを見つけます。
クエリを実行して、データが流れているかどうかを確認します。
```
SELECT * FROM <catalog>.<schema>.<prefix>_otel_logs
LIMIT 10;
```

テレメトリをクエリする

次の例は、一般的なクエリを示しています。

テレメトリテーブルの完全なスキーマを表示するには、次のコマンドを実行します。

DESCRIBE TABLE <catalog>.<schema>.<prefix>_otel_logs;

テレメトリデータをフィルター処理して関連付けるには、次の列を使用します。

timestamp
severity_text
body
trace_id
span_id
attributes — イベント固有のメタデータを含むマップ。

過去 1 時間のエラーを確認する

SELECT
  timestamp,
  severity_text,
  body,
  attributes
FROM <catalog>.<schema>.<prefix>_otel_logs
WHERE
  severity_text = 'ERROR'
  AND timestamp > current_timestamp() - INTERVAL 1 HOUR
ORDER BY timestamp DESC;

Troubleshooting

テーブルにログが表示されない: ルートログレベルは、オーバーヘッドを減らすために既定で WARNING されます。重要度の低いログをキャプチャするには、モデルコードのレベルを変更します。

class MyModel(mlflow.pyfunc.PythonModel):
    def load_context(self, context):
        root = logging.getLogger()
        root.setLevel(logging.DEBUG)
        for handler in root.handlers:
            handler.setLevel(logging.DEBUG)

制限事項

エンドポイントテレメトリには、次の制限が適用されます。

ターゲットテーブルでのスキーマの進化はサポートされていません。
マネージド Delta テーブルのみがサポートされています。外部ストレージと Arclight の既定のストレージはサポートされていません。
テーブルの場所は、ワークスペースと同じリージョンにある必要があります。
ASCII 文字、数字、およびアンダースコアを持つテーブル名のみがサポートされています。
ターゲットテーブルの再作成はサポートされていません。
単一可用性ゾーン (single-az) の持続性のみがサポートされています。
配信は少なくとも 1 回は保証されています。サーバーからの受信確認は、レコードが永続的であり、Delta テーブル内であることを意味します。
レコードはそれぞれ 10 MB 未満である必要があります。
要求はそれぞれ 30 MB 未満である必要があります。
ログ行はそれぞれ 1 MB 未満である必要があります。
テレメトリの待機時間は 2,500 QPS を超えて低下します。
ログは、生成されてから数秒後に Unity カタログテーブルに表示されます。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-03-15