use model version for batch scoring

Author: j-so

.pipelines/diabetes_regression-batchscoring-ci.yml

@@ -80,5 +80,5 @@ stages:
         azureSubscription: '$(WORKSPACE_SVC_CONNECTION)'
         PipelineId: '$(pipeline_id)'
         ExperimentName: '$(EXPERIMENT_NAME)'
-        PipelineParameters: '"ParameterAssignments": {"model_name": "$(MODEL_NAME)"}'
+        PipelineParameters: '"ParameterAssignments": {"model_name": "$(MODEL_NAME)", "model_version": "$(MODEL_VERSION)"}'
 

diabetes_regression/evaluate/evaluate_model.py

@@ -109,7 +109,10 @@
     tag_name = 'experiment_name'
 
     model = get_latest_model(
-        model_name, tag_name, exp.name, ws)
+                model_name=model_name,
+                tag_name=tag_name,
+                tag_value=exp.name,
+                aml_workspace=ws)
 
     if (model is not None):
         production_model_mse = 10000

diabetes_regression/scoring/parallel_batchscore.py

@@ -29,7 +29,7 @@
 import joblib
 import sys
 from typing import List
-from util.model_helper import get_latest_model
+from util.model_helper import get_model
 
 model = None
 
@@ -59,6 +59,19 @@ def parse_args() -> List[str]:
 
     model_name = model_name_param[0][1]
 
+    model_version_param = [
+    (sys.argv[idx], sys.argv[idx + 1])
+    for idx, itm in enumerate(sys.argv)
+    if itm == "--model_version"
+    ]
+
+    if len(model_version_param) == 0:
+        raise ValueError(
+            "Model name is required but no model name parameter was passed to the script"  # NOQA: E501
+        )
+
+    model_version = model_version_param[0][1]
+
     model_tag_name_param = [
         (sys.argv[idx], sys.argv[idx + 1])
         for idx, itm in enumerate(sys.argv)
@@ -83,7 +96,7 @@ def parse_args() -> List[str]:
         else model_tag_value_param[0][1]
     )
 
-    return [model_name, model_tag_name, model_tag_value]
+    return [model_name, model_version, model_tag_name, model_tag_value]
 
 
 def init():
@@ -95,12 +108,14 @@ def init():
         print("Initializing batch scoring script...")
 
         model_filter = parse_args()
-        amlmodel = get_latest_model(
-            model_filter[0], model_filter[1], model_filter[2]
-        )  # NOQA: E501
+        amlmodel = get_model(
+            model_name=env.model_filter[0],
+            model_version=model_filter[1],
+            tag_name=model_filter[2],
+            tag_value=model_filter[3])
 
         global model
-        modelpath = amlmodel.get_model_path(model_name=model_filter[0])
+        modelpath = Model.get_model_path(model_name=model_filter[0])
         model = joblib.load(modelpath)
         print("Loaded model {}".format(model_filter[0]))
     except Exception as ex:

diabetes_regression/util/model_helper.py

@@ -22,8 +22,9 @@ def get_current_workspace() -> Workspace:
     return experiment.workspace
 
 
-def get_latest_model(
+def get_model(
     model_name: str,
+    model_version: int = None, # If none, return latest model
     tag_name: str = None,
     tag_value: str = None,
     aml_workspace: Workspace = None
@@ -35,53 +36,25 @@ def get_latest_model(
     Parameters:
     aml_workspace (Workspace): aml.core Workspace that the model lives.
     model_name (str): name of the model we are looking for
+    (optional) model_version (str): version of the model. Returns latest if not provided.
     (optional) tag (str): the tag value & name the model was registered under.
 
     Return:
     A single aml model from the workspace that matches the name and tag.
     """
-    try:
-        # Validate params. cannot be None.
-        if model_name is None:
-            raise ValueError("model_name[:str] is required")
-
-        if aml_workspace is None:
+    if aml_workspace is None:
             print("No workspace defined - using current experiment workspace.")
             aml_workspace = get_current_workspace()
 
-        model_list = None
-        tag_ext = ""
-
-        # Get lastest model
-        # True: by name and tags
-        if tag_name is not None and tag_value is not None:
-            model_list = AMLModel.list(
-                aml_workspace, name=model_name,
-                tags=[[tag_name, tag_value]], latest=True
-            )
-            tag_ext = f"tag_name: {tag_name}, tag_value: {tag_value}."
-        # False: Only by name
-        else:
-            model_list = AMLModel.list(
-                aml_workspace, name=model_name, latest=True)
-
-        # latest should only return 1 model, but if it does,
-        # then maybe sdk or source code changed.
-
-        # define the error messages
-        too_many_model_message = ("Found more than one latest model. "
-                                  f"Models found: {model_list}. "
-                                  f"{tag_ext}")
-
-        no_model_found_message = (f"No Model found with name: {model_name}. "
-                                  f"{tag_ext}")
-
-        if len(model_list) > 1:
-            raise ValueError(too_many_model_message)
-        if len(model_list) == 1:
-            return model_list[0]
-        else:
-            print(no_model_found_message)
-            return None
-    except Exception:
-        raise
+    if tagname is not None and tagvalue is not None:
+        model = Model(aml_workspace, name=model_name, version=model_version, tags=[[tag_name, tag_value]])
+    elif (tagname is None and tagvalue is not None) or (
+        tagvalue is None and tagname is not None
+    ):
+        raise ValueError(
+            "model_tag_name and model_tag_value should both be supplied"
+            + "or excluded"  # NOQA: E501
+        )
+    else:
+        model = Model(aml_workspace, name=env.model_name, version=env.model_version)
+    return model

ml_service/pipelines/diabetes_regression_build_parallel_batchscore_pipeline.py

@@ -61,37 +61,6 @@ def parse_args() -> Namespace:
     args = parser.parse_args()
     return args
 
-
-def get_model(
-    ws: Workspace, env: Env, tagname: str = None, tagvalue: str = None
-) -> Model:
-    """
-    Gets a model from the models registered with the AML workspace.
-    If a tag/value pair is supplied, uses it to filter.
-
-    :param ws: Current AML workspace
-    :param env: Environment variables
-    :param tagname: Optional tag name, default is None
-    :param tagvalue: Optional tag value, default is None
-
-    :returns: Model
-
-    :raises: ValueError
-    """
-    if tagname is not None and tagvalue is not None:
-        model = Model(ws, name=env.model_name, tags=[[tagname, tagvalue]])
-    elif (tagname is None and tagvalue is not None) or (
-        tagvalue is None and tagname is not None
-    ):
-        raise ValueError(
-            "model_tag_name and model_tag_value should both be supplied"
-            + "or excluded"  # NOQA: E501
-        )
-    else:
-        model = Model(ws, name=env.model_name)
-    return model
-
-
 def get_or_create_datastore(
     datastorename: str, ws: Workspace, env: Env, input: bool = True
 ) -> Datastore:
@@ -331,7 +300,6 @@ def get_run_configs(
 
 
 def get_scoring_pipeline(
-    model: Model,
     scoring_dataset: Dataset,
     output_loc: PipelineData,
     score_run_config: ParallelRunConfig,
@@ -362,6 +330,9 @@ def get_scoring_pipeline(
     model_name_param = PipelineParameter(
         "model_name", default_value=env.model_name
     )  # NOQA: E501
+    model_version_param = PipelineParameter(
+        "model_version", default_value=env.model_version
+    )  # NOQA: E501
     model_tag_name_param = PipelineParameter(
         "model_tag_name", default_value=" "
     )  # NOQA: E501
@@ -376,6 +347,8 @@ def get_scoring_pipeline(
         arguments=[
             "--model_name",
             model_name_param,
+            "--model_version",
+            model_version_param,
             "--model_tag_name",
             model_tag_name_param,
             "--model_tag_value",
@@ -450,12 +423,7 @@ def build_batchscore_pipeline():
             aml_workspace, aml_compute_score, env
         )
 
-        trained_model = get_model(
-            aml_workspace, env, args.model_tag_name, args.model_tag_value
-        )
-
         scoring_pipeline = get_scoring_pipeline(
-            trained_model,
             input_dataset,
             output_location,
             scoring_runconfig,

ml_service/pipelines/diabetes_regression_verify_train_pipeline.py

@@ -3,7 +3,7 @@
 import os
 from azureml.core import Run, Experiment, Workspace
 from ml_service.util.env_variables import Env
-from diabetes_regression.util.model_helper import get_latest_model
+from diabetes_regression.util.model_helper import get_model
 
 
 def main():
@@ -53,8 +53,8 @@ def main():
 
     try:
         tag_name = 'BuildId'
-        model = get_latest_model(
-            model_name, tag_name, build_id, exp.workspace)
+        model = get_model(
+            model_name=model_name, tag_name=tag_name, tag_value=build_id, aml_workspace=exp.workspace)
         if (model is not None):
             print("Model was registered for this build.")
         if (model is None):

ml_service/pipelines/run_parallel_batchscore_pipeline.py

@@ -115,6 +115,7 @@ def run_batchscore_pipeline():
             scoringpipeline,
             pipeline_parameters={
                 "model_name": env.model_name,
+                "model_version": env.model_version,
                 "model_tag_name": " ",
                 "model_tag_value": " ",
             },