change: Update s3 path of scheduling analysis config on ClarifyCheckStep

* change: Update s3 path of scheduling analysis config in ClarifyCheckStep

Co-authored-by: Dewen Qi <[email protected]>

Author: 2 people

src/sagemaker/session.py

@@ -4186,7 +4186,8 @@ def get_model_package_args(
         approval_status (str): Model Approval Status, values can be "Approved", "Rejected",
             or "PendingManualApproval" (default: "PendingManualApproval").
         description (str): Model Package description (default: None).
-        tags (List[dict[str, str]]): A list of dictionaries containing key-value pairs.
+        tags (List[dict[str, str]]): A list of dictionaries containing key-value pairs
+            (default: None).
         container_def_list (list): A list of container defintiions (default: None).
         drift_check_baselines (DriftCheckBaselines): DriftCheckBaselines object (default: None).
     Returns:
@@ -4267,7 +4268,8 @@ def get_create_model_package_request(
         approval_status (str): Model Approval Status, values can be "Approved", "Rejected",
             or "PendingManualApproval" (default: "PendingManualApproval").
         description (str): Model Package description (default: None).
-        tags (List[dict[str, str]]): A list of dictionaries containing key-value pairs.
+        tags (List[dict[str, str]]): A list of dictionaries containing key-value pairs
+            (default: None).
         drift_check_baselines (DriftCheckBaselines): DriftCheckBaselines object (default: None).
     """
 

src/sagemaker/workflow/check_job_config.py

@@ -32,10 +32,8 @@ class CheckJobConfig:
     def __init__(
         self,
         role,
-        image_uri=None,
         instance_count=1,
         instance_type="ml.m5.xlarge",
-        entrypoint=None,
         volume_size_in_gb=30,
         volume_kms_key=None,
         output_kms_key=None,
@@ -50,14 +48,9 @@ def __init__(
 
         Args:
             role (str): An AWS IAM role. The Amazon SageMaker jobs use this role.
-            image_uri (str): The uri of the image to use for the jobs
-                started by the QualityCheckStep/ClarifyCheckStep (default: None).
-                If not specified, the default auto-generated image_uri will be used.
             instance_count (int): The number of instances to run the jobs with (default: 1).
             instance_type (str): Type of EC2 instance to use for the job
                 (default: 'ml.m5.xlarge').
-            entrypoint ([str]): The entrypoint for the job (default: None).
-                Only the QualityCheckStep will take this input.
             volume_size_in_gb (int): Size in GB of the EBS volume
                 to use for storing data during processing (default: 30).
             volume_kms_key (str): A KMS key for the processing volume (default: None).
@@ -77,12 +70,11 @@ def __init__(
             network_config (sagemaker.network.NetworkConfig): A NetworkConfig
                 object that configures network isolation, encryption of
                 inter-container traffic, security group IDs, and subnets (default: None).
+
         """
         self.role = role
-        self.image_uri = image_uri
         self.instance_count = instance_count
         self.instance_type = instance_type
-        self.entrypoint = entrypoint
         self.volume_size_in_gb = volume_size_in_gb
         self.volume_kms_key = volume_kms_key
         self.output_kms_key = output_kms_key
@@ -174,7 +166,4 @@ def _generate_model_monitor(self, mm_type: str) -> Optional[ModelMonitor]:
                 '"ModelBiasMonitor", "ModelExplainabilityMonitor"'
             )
             return None
-
-        monitor.image_uri = self.image_uri or monitor.image_uri
-        monitor.entrypoint = self.entrypoint or monitor.entrypoint
         return monitor

src/sagemaker/workflow/clarify_check_step.py

@@ -22,6 +22,7 @@
 
 import attr
 
+from sagemaker import s3
 from sagemaker.clarify import (
     DataConfig,
     BiasConfig,
@@ -33,6 +34,7 @@
     _set,
 )
 from sagemaker.model_monitor import BiasAnalysisConfig, ExplainabilityAnalysisConfig
+from sagemaker.model_monitor.model_monitoring import _MODEL_MONITOR_S3_PATH
 from sagemaker.processing import ProcessingInput, ProcessingOutput, ProcessingJob
 from sagemaker.utils import name_from_base
 from sagemaker.workflow import PipelineNonPrimitiveInputTypes, ExecutionVariable, Parameter
@@ -44,8 +46,8 @@
 _DATA_BIAS_TYPE = "DATA_BIAS"
 _MODEL_BIAS_TYPE = "MODEL_BIAS"
 _MODEL_EXPLAINABILITY_TYPE = "MODEL_EXPLAINABILITY"
-_BIAS_JOB_DEFINITION_BASE_NAME = "model-bias-job-definition"
-_EXPLAINABILITY_JOB_DEFINITION_BASE_NAME = "model-explainability-job-definition"
+_BIAS_MONITORING_CFG_BASE_NAME = "bias-monitoring"
+_EXPLAINABILITY_MONITORING_CFG_BASE_NAME = "model-explainability-monitoring"
 
 
 @attr.s
@@ -248,10 +250,14 @@ def __init__(
     @property
     def arguments(self) -> RequestType:
         """The arguments dict that is used to define the ClarifyCheck step."""
+        normalized_inputs, normalized_outputs = self._baselining_processor._normalize_args(
+            inputs=[self._processing_params["config_input"], self._processing_params["data_input"]],
+            outputs=[self._processing_params["result_output"]],
+        )
         process_args = ProcessingJob._get_process_args(
             self._baselining_processor,
-            [self._processing_params["config_input"], self._processing_params["data_input"]],
-            [self._processing_params["result_output"]],
+            normalized_inputs,
+            normalized_outputs,
             experiment_config=dict(),
         )
         request_dict = self._baselining_processor.sagemaker_session._get_process_request(
@@ -392,10 +398,8 @@ def _upload_monitoring_analysis_config(self) -> str:
         Returns:
             str: The S3 uri of the uploaded monitoring schedule analysis config
         """
-        monitor_schedule_name = self._model_monitor._generate_monitoring_schedule_name()
-        output_s3_uri = self._model_monitor._normalize_monitoring_output(
-            monitor_schedule_name
-        ).destination
+
+        output_s3_uri = self._get_s3_base_uri_for_monitoring_analysis_config()
 
         if isinstance(self.clarify_check_config, ModelExplainabilityCheckConfig):
             # Explainability analysis doesn't need label
@@ -410,7 +414,9 @@ def _upload_monitoring_analysis_config(self) -> str:
             analysis_config = explainability_analysis_config._to_dict()
             if "predictor" in analysis_config and "model_name" in analysis_config["predictor"]:
                 analysis_config["predictor"].pop("model_name")
-            job_definition_name = name_from_base(_EXPLAINABILITY_JOB_DEFINITION_BASE_NAME)
+            job_definition_name = name_from_base(
+                f"{_EXPLAINABILITY_MONITORING_CFG_BASE_NAME}-config"
+            )
 
         else:
             bias_analysis_config = BiasAnalysisConfig(
@@ -419,8 +425,33 @@ def _upload_monitoring_analysis_config(self) -> str:
                 label=self.clarify_check_config.data_config.label,
             )
             analysis_config = bias_analysis_config._to_dict()
-            job_definition_name = name_from_base(_BIAS_JOB_DEFINITION_BASE_NAME)
+            job_definition_name = name_from_base(f"{_BIAS_MONITORING_CFG_BASE_NAME}-config")
 
         return self._model_monitor._upload_analysis_config(
             analysis_config, output_s3_uri, job_definition_name
         )
+
+    def _get_s3_base_uri_for_monitoring_analysis_config(self) -> str:
+        """Generate s3 base uri for monitoring schedule analysis config
+
+        Returns:
+            str: The S3 base uri of the monitoring schedule analysis config
+        """
+        s3_analysis_config_output_path = (
+            self.clarify_check_config.data_config.s3_analysis_config_output_path
+        )
+        monitoring_cfg_base_name = f"{_BIAS_MONITORING_CFG_BASE_NAME}-configuration"
+        if isinstance(self.clarify_check_config, ModelExplainabilityCheckConfig):
+            monitoring_cfg_base_name = f"{_EXPLAINABILITY_MONITORING_CFG_BASE_NAME}-configuration"
+
+        if s3_analysis_config_output_path:
+            return s3.s3_path_join(
+                s3_analysis_config_output_path,
+                monitoring_cfg_base_name,
+            )
+        return s3.s3_path_join(
+            "s3://",
+            self._model_monitor.sagemaker_session.default_bucket(),
+            _MODEL_MONITOR_S3_PATH,
+            monitoring_cfg_base_name,
+        )

src/sagemaker/workflow/quality_check_step.py

@@ -222,16 +222,21 @@ def __init__(
     @property
     def arguments(self) -> RequestType:
         """The arguments dict that is used to define the QualityCheck step."""
+        normalized_inputs, normalized_outputs = self._baselining_processor._normalize_args(
+            inputs=self._baseline_job_inputs,
+            outputs=[self._baseline_output],
+        )
         process_args = ProcessingJob._get_process_args(
             self._baselining_processor,
-            self._baseline_job_inputs,
-            [self._baseline_output],
+            normalized_inputs,
+            normalized_outputs,
             experiment_config=dict(),
         )
         request_dict = self._baselining_processor.sagemaker_session._get_process_request(
             **process_args
         )
-        request_dict.pop("ProcessingJobName")
+        if "ProcessingJobName" in request_dict:
+            request_dict.pop("ProcessingJobName")
 
         return request_dict