Rendere persistente il modello personalizzato che gestisce i dati in Unity Catalog

Importante

Questa funzionalità è in versione beta. Non è abilitato automaticamente per tutti i clienti e le funzionalità sono soggette a modifiche. Per richiedere l'accesso, contattare il team dell'account di Azure Databricks.

Informazioni su come configurare i dati di telemetria degli endpoint per rendere persistenti i log, le tracce e le metriche di OpenTelemetry dal modello personalizzato che gestisce gli endpoint alle tabelle del catalogo Unity. Usare i dati di telemetria persistenti per eseguire l'analisi della causa radice, monitorare l'integrità degli endpoint e soddisfare i requisiti di conformità con query SQL standard.

Requisiti

L'area di lavoro deve essere abilitata per Unity Catalog. L'archiviazione predefinita (Arclight) non è supportata.
È necessario disporre delle autorizzazioni USE CATALOG, USE SCHEMA, CREATE TABLE e MODIFY sul catalogo e lo schema di destinazione del catalogo Unity in cui sono archiviati i log.
Un modello personalizzato esistente che gestisce l'endpoint o le autorizzazioni per crearne uno.
L'area di lavoro deve trovarsi in un'area supportata:
- canadacentral
- westus
- westus2
- southcentralus
- eastus
- eastus2
- centralus
- northcentralus
- swedencentral
- westeurope
- northeurope
- uksouth
- australiaeast
- southeastasia

Passaggio 1: Instrumentare il codice del modello

Aggiungere strumentazione al codice del modello per acquisire i dati di telemetria.

Aggiungere la registrazione delle applicazioni al modello. La telemetria degli endpoint acquisisce automaticamente l'output Python logging standard. Non è necessaria alcuna strumentazione di OpenTelemetry SDK per la registrazione di base.

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

Il livello di log del root è impostato su WARNING. Vedere Risoluzione dei problemi per modificare il livello di registrazione.

(Facoltativo) Instrumentare metriche e tracce personalizzate con OpenTelemetry. Per acquisire metriche e tracce personalizzate oltre alla registrazione di base, aggiungere la strumentazione di OpenTelemetry SDK al modello. Espandere la sezione seguente per un esempio completo che illustra come creare contatori, intervalli di record e associare attributi personalizzati.

Esempio: Metriche personalizzate, segmenti e log del modello con OpenTelemetry

Annotazioni

A causa delle limitazioni della serializzazione del modello, è necessario scrivere il modello in un file separato prima della registrazione per evitare errori, come illustrato di seguito usando %%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())

Annotare e iscrivere il modello.

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"
)

Passaggio 2: Preparare la destinazione del catalogo Unity

Prima di creare l'endpoint, assicurarsi di avere un catalogo e uno schema pronti per ricevere i dati di telemetria. Azure Databricks crea automaticamente le tabelle necessarie in questo schema, se non esistono già.

In Esplora cataloghi passare al catalogo e allo schema da usare, ad esempio my_catalog.observability.

Passaggio 3: Abilitare la telemetria degli endpoint

È possibile abilitare i dati di telemetria durante la creazione di un nuovo endpoint o aggiungerli a uno esistente.

Nuovo endpoint

Per abilitare i dati di telemetria nell'interfaccia utente:

Passare a Serving nella barra laterale sinistra.
Fare click su Crea l'endpoint di servizio.
Nella sezione Endpoint telemetry (contrassegnata come anteprima) espandere le opzioni di configurazione.
Percorso catalogo Unity: selezionare il catalogo di destinazione e lo schema preparati nel passaggio 2.
(Facoltativo) Prefisso tabella: immettere un prefisso per le tabelle generate. Se lasciato vuoto, non è presente alcun prefisso. Le tabelle sono denominate <prefix>_otel_logs, <prefix>_otel_spanse <prefix>_otel_metrics.
Completare il resto della configurazione dell'endpoint (selezione del modello, impostazioni di calcolo) e fare clic su Crea.

A tale scopo, usare l'API:

Abilitare la telemetria usando l'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"
      }
    }
  }
}'

Endpoint esistente

Annotazioni

L'aggiornamento attiva una nuova distribuzione. Le modifiche diventano effettive al termine della distribuzione.

Per abilitare i dati di telemetria nell'interfaccia utente:

Nella pagina di visualizzazione endpoint, dal pannello a destra, sotto la sezione Telemetria dell'endpoint, cliccare su Aggiungi.
Percorso catalogo Unity: selezionare il catalogo di destinazione e lo schema preparati nel passaggio 2.
(Facoltativo) Prefisso tabella: immettere un prefisso per le tabelle generate. Se lasciato vuoto, non è presente alcun prefisso. Le tabelle sono denominate <prefix>_otel_logs, <prefix>_otel_spanse <prefix>_otel_metrics.
Clicca su Aggiorna.

Passaggio 4: Verificare e interrogare i dati di telemetria

Dopo che l'endpoint riceve il traffico, i flussi di dati di telemetria vengono inviati alle tabelle del catalogo Unity configurate.

Vai a Esplora Cataloghi o Editor SQL.
Individuare la tabella denominata <prefix>_otel_logs nello schema configurato.

Esegui una query per verificare che i dati fluiscano.

SELECT * FROM <catalog>.<schema>.<prefix>_otel_logs
LIMIT 10;

Interrogazione dei dati di telemetria

Negli esempi seguenti vengono visualizzate query comuni.

Per visualizzare lo schema completo di qualsiasi tabella di telemetria, eseguire:

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

Usare queste colonne per filtrare e correlare i dati di telemetria:

timestamp
severity_text
body
trace_id
span_id
attributes — una mappa che contiene metadati specifici dell'evento.

Verificare la presenza di errori nell'ultima ora

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;

Risoluzione dei problemi

Log non visualizzati nella tabella: il livello di registrazione predefinito della radice è impostato su WARNING per ridurre il sovraccarico. Per acquisire i log con gravità inferiore, modificare il livello nel codice del modello:

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)

Limitazioni

I limiti seguenti si applicano ai dati di telemetria degli endpoint:

L'evoluzione dello schema nella tabella di destinazione non è supportata.
Sono supportate solo le tabelle Delta gestite. L'archiviazione esterna e l'archiviazione predefinita di Arclight non sono supportate.
La posizione della tabella deve trovarsi nella stessa area dell'area di lavoro.
Sono supportati solo i nomi di tabella con lettere ASCII, cifre e caratteri di sottolineatura.
La ricreazione di una tabella di destinazione non è supportata.
È supportata solo la durabilità della singola zona di disponibilità (single-az).
La consegna garantisce almeno un recapito. Una conferma dal server indica che il record è persistente e nella tabella Delta.
I record devono essere inferiori a 10 MB ciascuno.
Le richieste devono essere inferiori a 30 MB ciascuno.
Le righe di log devono essere inferiori a 1 MB ciascuno.
La latenza della telemetria si deteriora oltre 2500 QPS.
I log vengono visualizzati nella tabella del catalogo Unity pochi secondi dopo l'emissione.

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-03-15