feat(bundle): make the bare model name a variable (base_model_name)

README.md

@@ -456,7 +456,9 @@ base_parameters:
 ```
 
 and the model name follows the same schema:
-`${var.catalog_name}.${resources.schemas.ml_workspace.name}.fraud_detection`.
+`${var.catalog_name}.${resources.schemas.ml_workspace.name}.${var.model_name}`
+(where `model_name` is the bare name, default `fraud_detection`). The notebooks prepend
+the catalog and `ml_schema` to it to build the three-level `uc_model_name`.
 
 > **Why reference the resource, not a plain variable?** Development mode prefixes the schema
 > *resource* but not a hand-written variable, so a variable like `${short_name}_fraud` would

databricks.yml

@@ -15,8 +15,10 @@ variables:
     description: MLflow experiment for model training runs (per-user path).
     default: /Users/${workspace.current_user.userName}/${bundle.target}-mlops-workshop-fraud
   model_name:
-    description: Unity Catalog registered model name (in the per-user workspace schema).
-    default: ${var.catalog_name}.${var.ml_schema}.fraud_detection
+    description: >-
+      Bare (unqualified) registered model name. Notebooks prepend catalog_name + ml_schema
+      to form the three-level Unity Catalog name (uc_model_name).
+    default: fraud_detection
   catalog_name:
     description: Unity Catalog catalog for the environment (dev/staging/prod).
     default: adoption_workshop

resources/deployment-job-workflow.yml

@@ -24,7 +24,7 @@ resources:
         enabled: true
       parameters:
         - name: model_name
-          default: ${var.model_name}
+          default: ${var.catalog_name}.${var.ml_schema}.${var.model_name}
         - name: model_version
           default: ""
       tasks:
@@ -35,6 +35,7 @@ resources:
             notebook_path: ../${var.code_root}/deployment/model_deployment/notebooks/evaluation.py
             base_parameters:
               catalog_name: ${var.catalog_name}
+              ml_schema: ${var.ml_schema}
               metric: roc_auc
               baseline: "0.65"
 

solution/deployment/batch_inference/notebooks/batch_inference.py

@@ -26,7 +26,7 @@
 dbutils.widgets.text("catalog_name", "adoption_workshop")
 dbutils.widgets.text("gold_schema", "fraud_gold")
 dbutils.widgets.text("ml_schema", "")
-dbutils.widgets.text("model_name", "")
+dbutils.widgets.text("model_name", "fraud_detection")
 dbutils.widgets.text("model_alias", "champion")
 
 catalog_name = dbutils.widgets.get("catalog_name")
@@ -43,14 +43,17 @@
     ml_schema = f"dev_{_short}_fraud"
 
 model_alias = dbutils.widgets.get("model_alias")
-# The batch-inference job passes the full three-level model name; fall back for interactive runs.
-model_name = dbutils.widgets.get("model_name") or f"{catalog_name}.{ml_schema}.fraud_detection"
+model_name = dbutils.widgets.get("model_name")
+# Accept either a bare name (prepend catalog + ml_schema) or an already-qualified 3-level name.
+uc_model_name = (
+    model_name if model_name.count(".") == 2 else f"{catalog_name}.{ml_schema}.{model_name}"
+)
 
 # Read the transactions to score from the shared gold source; write predictions to the personal schema.
 source_table = f"{catalog_name}.{gold_schema}.transactions_enriched"
 predictions_table = f"{catalog_name}.{ml_schema}.fraud_predictions"
 
-print(f"Model:       {model_name}@{model_alias}")
+print(f"Model:       {uc_model_name}@{model_alias}")
 print(f"Score from:  {source_table}")
 print(f"Write to:    {predictions_table}")
 
@@ -77,7 +80,7 @@
 # Resolve the champion alias to a concrete version, to stamp every scored row with the model
 # version that produced it (the monitor's model_id_col groups metrics by this).
 client = MlflowClient()
-champion = client.get_model_version_by_alias(model_name, model_alias)
+champion = client.get_model_version_by_alias(uc_model_name, model_alias)
 model_version = str(champion.version)
 print(f"{model_alias} -> version {model_version}")
 
@@ -121,10 +124,10 @@
 # COMMAND ----------
 
 # TODO-BEGIN: score the batch with the champion model
-# HINT: fe.score_batch(model_uri=f"models:/{model_name}@{model_alias}", df=to_score).
+# HINT: fe.score_batch(model_uri=f"models:/{uc_model_name}@{model_alias}", df=to_score).
 # HINT: the result has the lookup key plus a "prediction" column (the monitor's prediction_col).
 scored = fe.score_batch(
-    model_uri=f"models:/{model_name}@{model_alias}",
+    model_uri=f"models:/{uc_model_name}@{model_alias}",
     df=to_score,
 )
 # TODO-END

solution/deployment/model_deployment/notebooks/approval.py

@@ -20,22 +20,59 @@
 
 # COMMAND ----------
 
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` it is gating. Both are empty when this notebook is
+# run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
+dbutils.widgets.text("catalog_name", "adoption_workshop")
+dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("approval_tag_name", "")
 dbutils.widgets.text("auto_approve", "false")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
+catalog_name = dbutils.widgets.get("catalog_name")
+ml_schema = dbutils.widgets.get("ml_schema")
 # The deployment job passes the task name here (e.g. "Approval_Check") as the tag key.
 tag_name = dbutils.widgets.get("approval_tag_name") or "Approval_Check"
 auto_approve = dbutils.widgets.get("auto_approve").strip().lower() == "true"
 
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
+
+from pyspark.sql import functions as F
+
+# Interactive fallback: rebuild the personal schema and the full model name, and default to
+# the latest version, so the notebook can be run by hand (outside the deployment job).
+if not ml_schema:
+    _user = spark.range(1).select(F.current_user()).first()[0]
+    _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
+    ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
 print(
-    f"Approval check for {model_name} v{model_version} | tag='{tag_name}' | auto_approve={auto_approve}"
+    f"Approval check for {uc_model_name} v{model_version} | tag='{tag_name}' | auto_approve={auto_approve}"
 )
 
 # COMMAND ----------
@@ -47,10 +84,10 @@
 # In dev, auto-approve: set the tag so this run (and the UI) reflect an approved state, then
 # pass. In staging/prod, a human must have set the tag to "Approved".
 if auto_approve:
-    client.set_model_version_tag(model_name, model_version, tag_name, "Approved")
+    client.set_model_version_tag(uc_model_name, model_version, tag_name, "Approved")
     print(f"Auto-approved (dev): set {tag_name}=Approved on v{model_version}.")
 else:
-    mv = client.get_model_version(model_name, model_version)
+    mv = client.get_model_version(uc_model_name, model_version)
     status = (mv.tags or {}).get(tag_name)
     print(f"Current approval tag {tag_name}={status!r}")
     if status != "Approved":

solution/deployment/model_deployment/notebooks/deployment.py

@@ -18,35 +18,59 @@
 
 # COMMAND ----------
 
-# The deployment job injects model_name + model_version as JOB-level parameters.
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` that triggered it. Both are empty when this notebook
+# is run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
 dbutils.widgets.text("catalog_name", "adoption_workshop")
 dbutils.widgets.text("ml_schema", "")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
 catalog_name = dbutils.widgets.get("catalog_name")
 ml_schema = dbutils.widgets.get("ml_schema")
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
+
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
 
 from pyspark.sql import functions as F
 
-# Interactive fallback: jobs pass the resolved personal schema; running standalone derives
-# the same per-user name the bundle uses (dev_<short>_fraud) so runs stay isolated.
+# Interactive fallback: rebuild the personal schema and the full model name, and default to
+# the latest version, so the notebook can be run by hand (outside the deployment job).
 if not ml_schema:
     _user = spark.range(1).select(F.current_user()).first()[0]
     _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
     ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
 
 endpoint_name = f"{ml_schema}_fraud"
 online_store_name = "fraud-workshop-online"
 card_feature_table = f"{catalog_name}.{ml_schema}.card_features"
 client_feature_table = f"{catalog_name}.{ml_schema}.client_features"
 
-print(f"Deploying {model_name} v{model_version} -> endpoint {endpoint_name}")
+print(f"Deploying {uc_model_name} v{model_version} -> endpoint {endpoint_name}")
 
 # COMMAND ----------
 
@@ -63,9 +87,9 @@
 mlflow.set_registry_uri("databricks-uc")
 client = MlflowClient()
 
-client.set_registered_model_alias(model_name, "champion", model_version)
+client.set_registered_model_alias(uc_model_name, "champion", model_version)
 try:
-    client.delete_registered_model_alias(model_name, "challenger")
+    client.delete_registered_model_alias(uc_model_name, "challenger")
 except Exception:
     pass  # there may be no challenger alias (e.g. first deployment)
 print(f"@champion -> version {model_version}")
@@ -131,7 +155,7 @@ def wait_online_store(name, minutes=20):
 
 served_entities = [
     ServedEntityInput(
-        entity_name=model_name,
+        entity_name=uc_model_name,
         entity_version=model_version,
         scale_to_zero_enabled=True,
         workload_size="Small",

solution/deployment/model_deployment/notebooks/evaluation.py

@@ -21,23 +21,58 @@
 
 # COMMAND ----------
 
-# The deployment job injects model_name + model_version as JOB-level parameters.
+# The deployment job injects the FULL three-level model name (catalog.schema.model) into
+# `model_name`, plus the `model_version` that triggered it. Both are empty when this notebook
+# is run interactively, so the fallback below rebuilds them.
 dbutils.widgets.text("model_name", "")
 dbutils.widgets.text("model_version", "")
 dbutils.widgets.text("catalog_name", "adoption_workshop")
+dbutils.widgets.text("ml_schema", "")
 dbutils.widgets.text("metric", "roc_auc")
 dbutils.widgets.text("baseline", "0.65")
 
-model_name = dbutils.widgets.get("model_name")
-model_version = dbutils.widgets.get("model_version")
+uc_model_name = dbutils.widgets.get("model_name").strip()
+model_version = dbutils.widgets.get("model_version").strip()
 catalog_name = dbutils.widgets.get("catalog_name")
+ml_schema = dbutils.widgets.get("ml_schema")
 metric = dbutils.widgets.get("metric")
 baseline = float(dbutils.widgets.get("baseline"))
 
-assert model_name and model_version, (
-    "model_name and model_version are injected by the deployment job."
-)
-print(f"Evaluating {model_name} version {model_version} on '{metric}' (floor {baseline}).")
+# Fail fast on a half-configured deployment job: model_name and model_version are injected
+# together. Both present = job; both empty = interactive run (derived below). Exactly one
+# present is a misconfiguration that would silently gate the wrong model or version.
+if bool(uc_model_name) != bool(model_version):
+    raise ValueError(
+        "Only one of model_name / model_version was provided. Pass both (deployment job) "
+        "or neither (interactive run)."
+    )
+
+from pyspark.sql import functions as F
+
+# Interactive fallback: rebuild the personal schema and the full model name the same way the
+# training/batch notebooks do, and default to the latest version, so this runs by hand too.
+if not ml_schema:
+    _user = spark.range(1).select(F.current_user()).first()[0]
+    _short = "".join(c if c.isalnum() else "_" for c in _user.split("@")[0])
+    ml_schema = f"dev_{_short}_fraud"
+if not uc_model_name:
+    uc_model_name = f"{catalog_name}.{ml_schema}.fraud_detection"
+if not model_version:
+    import mlflow
+    from mlflow.tracking import MlflowClient
+
+    mlflow.set_registry_uri("databricks-uc")
+    _versions = [
+        int(mv.version)
+        for mv in MlflowClient().search_model_versions(f"name='{uc_model_name}'")
+    ]
+    if not _versions:
+        raise ValueError(
+            f"No model versions found for {uc_model_name!r}. Provide model_version explicitly."
+        )
+    model_version = str(max(_versions))
+
+print(f"Evaluating {uc_model_name} version {model_version} on '{metric}' (floor {baseline}).")
 
 # COMMAND ----------
 
@@ -79,7 +114,7 @@
 
 def score_holdout(version):
     """Score a model version on the holdout; return a pandas DF of is_fraud + fraud score."""
-    scored = fe.score_batch(model_uri=f"models:/{model_name}/{version}", df=eval_spine)
+    scored = fe.score_batch(model_uri=f"models:/{uc_model_name}/{version}", df=eval_spine)
     return scored.select("is_fraud", F.col("prediction").cast("double").alias("score")).toPandas()
 
 
@@ -89,7 +124,7 @@ def score_holdout(version):
 # Fair comparison: re-score the current champion on the SAME holdout. If there is no champion
 # yet (first model), gate against the metric floor instead.
 try:
-    champion = client.get_model_version_by_alias(model_name, "champion")
+    champion = client.get_model_version_by_alias(uc_model_name, "champion")
     champ_pdf = score_holdout(champion.version)
     bar = float(roc_auc_score(champ_pdf["is_fraud"], champ_pdf["score"]))
     bar_label = f"champion (v{champion.version})"
@@ -99,9 +134,9 @@ def score_holdout(version):
 print(f"candidate {metric}={candidate_score:.4f}  vs  {bar_label}={bar:.4f}")
 
 # Surface the decision inputs on the model version so the approver sees them in the UI.
-client.set_model_version_tag(model_name, model_version, "eval_metric", metric)
-client.set_model_version_tag(model_name, model_version, "eval_score", f"{candidate_score:.4f}")
-client.set_model_version_tag(model_name, model_version, "eval_bar", f"{bar:.4f}")
+client.set_model_version_tag(uc_model_name, model_version, "eval_metric", metric)
+client.set_model_version_tag(uc_model_name, model_version, "eval_score", f"{candidate_score:.4f}")
+client.set_model_version_tag(uc_model_name, model_version, "eval_bar", f"{bar:.4f}")
 
 # COMMAND ----------
 
@@ -125,7 +160,7 @@ def score_holdout(version):
 ConfusionMatrixDisplay(cm, display_labels=["legit", "fraud"]).plot(
     cmap="Blues", ax=ax, colorbar=False
 )
-ax.set_title(f"{model_name.split('.')[-1]} v{model_version} (holdout, threshold 0.5)")
+ax.set_title(f"{uc_model_name.split('.')[-1]} v{model_version} (holdout, threshold 0.5)")
 fig.tight_layout()
 with mlflow.start_run(run_name=f"evaluation_v{model_version}") as ev_run:
     mlflow.log_figure(fig, "confusion_matrix.png")