搭配 Databricks Lakehouse 監視使用自訂計量
此頁面說明如何在 Databricks Lakehouse 監視中建立自訂計量。 除了自動計算的分析與漂移統計數據之外，您還可以建立自訂計量。 例如，您可能想要追蹤擷取商務邏輯某些層面的加權平均值，或使用自訂模型品質分數。 您也可以建立自定義漂移計量，以追蹤主要數據表中值的變更（相較於基準或上一個時間範圍）。
如需更多如何使用 MonitorMetric API 的詳細資訊，請參閱 API 參照。
自訂計量的類型
Databricks Lakehouse 監視包括下列類型的自訂計量：
- 匯總指標，這是根據主表中的列計算。 匯總計量會儲存在配置檔計量數據表中。
- 衍生指標是根據先前計算的彙總指標計算得到的，並且不會直接使用來自主表的數據。 衍生的度量會儲存在分析度量表中。
- 漂移計量，其會比較先前計算的匯總或衍生計量來自兩個不同的時間範圍，或主要數據表與基準數據表之間的計量。 漂移計量會儲存在漂移計量數據表中。
盡可能使用衍生和漂移計量，將完整主數據表的重新計算降至最低。 僅匯總計量會從主數據表存取數據。 然後可以直接從匯總計量值計算衍生和漂移計量。
自定義計量參數
若要定義自定義計量，您可以為 SQL 資料行表示式建立 Jinja 範本。 本節中的數據表描述定義計量的參數，以及 Jinja 範本中使用的參數。
| 參數 | 描述 |
|---|---|
type |
MonitorMetricType.CUSTOM_METRIC_TYPE_AGGREGATE、MonitorMetricType.CUSTOM_METRIC_TYPE_DERIVED 或 MonitorMetricType.CUSTOM_METRIC_TYPE_DRIFT 的其中之一。 |
name |
計量數據表中自定義計量的數據行名稱。 |
input_columns |
輸入數據表中應該計算計量的欄位名稱清單。 若要指出計算中使用了一個以上的資料列，請使用 :table。 請參閱本文中的範例。 |
definition |
指定計算計量方法之 SQL 運算式 Jinja 範本。 請參閱建議定義。 |
output_data_type |
以 JSON 字串格式計量輸出的 Spark 資料類型。 |
建立 definition
參數 definition 必須為 Jinja 範本表單中的單一字串運算式。 它不能包含聯結或子查詢。
下表列出可用來建立 SQL Jinja 範本的參數，以指定如何計算計量。
| 參數 | 描述 |
|---|---|
{{input_column}} |
用來計算自定義指標的數據行。 |
{{prediction_col}} |
包含 ML 模型預測的欄。 與分析搭配 InferenceLog 使用。 |
{{label_col}} |
欄位持有 ML 模型真實值標籤。 與分析搭配 InferenceLog 使用。 |
{{current_df}} |
與前一個時間範圍相比的漂移。 上一個時間範圍的數據。 |
{{base_df}} |
相對於基準數據表的漂移。 基準資料。 |
彙總計量範例
下列範例會計算數據行中值的平方平均值，並套用至數據行 f1 和 f2。 輸出會儲存為設定檔計量表中的新數據列，並顯示在對應到數據列 f1 和 f2的分析數據行中。 適用的欄位名稱會取代 Jinja 參數 {{input_column}}。
from databricks.sdk.service.catalog import MonitorMetric, MonitorMetricType
from pyspark.sql import types as T
MonitorMetric(
type=MonitorMetricType.CUSTOM_METRIC_TYPE_AGGREGATE,
name="squared_avg",
input_columns=["f1", "f2"],
definition="avg(`{{input_column}}`*`{{input_column}}`)",
output_data_type=T.StructField("output", T.DoubleType()).json(),
)
下列程式代碼會定義自定義計量，以計算數據行 f1 與 f2之間的差異平均值。 這個範例示範在 input_columns 參數中使用 [":table"]，指出計算中使用數據表中的多個數據行。
from databricks.sdk.service.catalog import MonitorMetric, MonitorMetricType
from pyspark.sql import types as T
MonitorMetric(
type=MonitorMetricType.CUSTOM_METRIC_TYPE_AGGREGATE,
name="avg_diff_f1_f2",
input_columns=[":table"],
definition="avg(f1 - f2)",
output_data_type=T.StructField("output", T.DoubleType()).json(),
)
此範例會計算加權模型品質分數。 對於在 critical 欄位為 True的觀察中，當該列的預測值不匹配基準事實時，將會施加更重的懲罰。 因為它是在原始數據行上定義的 （prediction 和 label），所以它會定義為匯總計量。
:table 數據行表示此計量是從多個數據行計算而來。 Jinja 參數 {{prediction_col}} 和 {{label_col}} 會被替換為監視器的預測欄和真實標籤欄的名稱。
from databricks.sdk.service.catalog import MonitorMetric, MonitorMetricType
from pyspark.sql import types as T
MonitorMetric(
type=MonitorMetricType.CUSTOM_METRIC_TYPE_AGGREGATE,
name="weighted_error",
input_columns=[":table"],
definition="""avg(CASE
WHEN {{prediction_col}} = {{label_col}} THEN 0
WHEN {{prediction_col}} != {{label_col}} AND critical=TRUE THEN 2
ELSE 1 END)""",
output_data_type=T.StructField("output", T.DoubleType()).json(),
)
衍生的計量範例
下列代碼會定義自訂計量，用於計算本章節稍早定義之 squared_avg 計量的平方根。 因為這是衍生的計量，因此不會參考主數據表數據，而是以匯總計量 squared_avg 來定義。 輸出會儲存為配置檔計量數據表中的新欄位。
from databricks.sdk.service.catalog import MonitorMetric, MonitorMetricType
from pyspark.sql import types as T
MonitorMetric(
type=MonitorMetricType.CUSTOM_METRIC_TYPE_DERIVED,
name="root_mean_square",
input_columns=["f1", "f2"],
definition="sqrt(squared_avg)",
output_data_type=T.StructField("output", T.DoubleType()).json(),
)
漂移計量範例
下列代碼會定義漂移計量，用於追蹤本章節稍早定義的 weighted_error 計量變更。
{{current_df}} 和 {{base_df}} 參數可讓計量從目前視窗和比較窗口參考 weighted_error 值。 比較視窗可以是基準數據或上一個時間範圍中的資料。 漂移計量會儲存在漂移計量數據表中。
from databricks.sdk.service.catalog import MonitorMetric, MonitorMetricType
from pyspark.sql import types as T
MonitorMetric(
type=MonitorMetricType.CUSTOM_METRIC_TYPE_DRIFT,
name="error_rate_delta",
input_columns=[":table"],
definition="{{current_df}}.weighted_error - {{base_df}}.weighted_error",
output_data_type=T.StructField("output", T.DoubleType()).json(),
)