Merge branch 'master' into fix/remove-unnecessary-get-caller-identity-call

Author: evakravi

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## v2.171.0 (2023-07-06)
+
+### Features
+
+ * Add PipelineDefinitionConfig to pipelines to toggle custom job …
+
+### Bug Fixes and Other Changes
+
+ * Upgrade DJL deepspeed versions
+ * Remove unused dependency `protobuf3-to-dict`
+ * skip intelligent volume_size allocation based on instance type if it is a pipeline parameter
+
 ## v2.170.0 (2023-07-05)
 
 ### Features

VERSION

@@ -1 +1 @@
-2.170.1.dev0
+2.171.1.dev0

doc/workflows/pipelines/sagemaker.workflow.pipelines.rst

@@ -103,6 +103,11 @@ Parallelism Configuration
 .. autoclass:: sagemaker.workflow.parallelism_config.ParallelismConfiguration
     :members:
 
+Pipeline Definition Config
+--------------------------
+
+.. autoclass:: sagemaker.workflow.pipeline_definition_config.PipelineDefinitionConfig
+
 Pipeline Experiment Config
 --------------------------
 

setup.py

@@ -53,7 +53,6 @@ def read_requirements(filename):
     "google-pasta",
     "numpy>=1.9.0,<2.0",
     "protobuf>=3.1,<4.0",
-    "protobuf3-to-dict>=0.1.5,<1.0",
     "smdebug_rulesconfig==1.0.1",
     "importlib-metadata>=1.4.0,<5.0",
     "packaging>=20.0",

src/sagemaker/clarify.py

@@ -267,6 +267,7 @@
                     },
                 },
                 SchemaOptional("seed"): int,
+                SchemaOptional("features_to_explain"): [Or(int, str)],
             },
             SchemaOptional("pre_training_bias"): {"methods": Or(str, [str])},
             SchemaOptional("post_training_bias"): {"methods": Or(str, [str])},
@@ -1308,6 +1309,7 @@ def __init__(
         num_clusters: Optional[int] = None,
         text_config: Optional[TextConfig] = None,
         image_config: Optional[ImageConfig] = None,
+        features_to_explain: Optional[List[Union[str, int]]] = None,
     ):
         """Initializes config for SHAP analysis.
 
@@ -1343,6 +1345,14 @@ def __init__(
                 text features. Default is None.
             image_config (:class:`~sagemaker.clarify.ImageConfig`): Config for handling image
                 features. Default is None.
+            features_to_explain: A list of names or indices of dataset features to compute SHAP
+                values for. If not provided, SHAP values are computed for all features by default.
+                Currently only supported for tabular datasets.
+
+        Raises:
+            ValueError: when ``agg_method`` is invalid, ``baseline`` and ``num_clusters`` are provided
+                together, or ``features_to_explain`` is specified when ``text_config`` or
+                ``image_config`` is provided
         """  # noqa E501  # pylint: disable=c0301
         if agg_method is not None and agg_method not in [
             "mean_abs",
@@ -1376,6 +1386,13 @@ def __init__(
                 )
         if image_config:
             _set(image_config.get_image_config(), "image_config", self.shap_config)
+        if features_to_explain is not None and (
+            text_config is not None or image_config is not None
+        ):
+            raise ValueError(
+                "`features_to_explain` is not supported for datasets containing text features or images."
+            )
+        _set(features_to_explain, "features_to_explain", self.shap_config)
 
     def get_explainability_config(self):
         """Returns a shap config dictionary."""

src/sagemaker/djl_inference/model.py

@@ -196,7 +196,7 @@ def _create_estimator(
     image_uri: str,
     role: str,
     sagemaker_session: Optional[Session],
-    volume_size: int = 30,
+    volume_size: int,
     vpc_config: Optional[
         Dict[
             str,
@@ -453,7 +453,9 @@ def partition(
         self,
         instance_type: str,
         s3_output_uri: str = None,
+        s3_output_prefix: str = "aot-partitioned-checkpoints",
         job_name: Optional[str] = None,
+        volume_size: int = 30,
         volume_kms_key: Optional[str] = None,
         output_kms_key: Optional[str] = None,
         use_spot_instances: bool = False,
@@ -469,8 +471,13 @@ def partition(
                     artifacts and output files). If not specified, results are
                     stored to a default bucket. If the bucket with the specific name
                     does not exist, it will be created.
+            s3_output_prefix (str): Name of the prefix where all the partitioned
+                    checkpoints to be uploaded. If not provided, the default value is
+                    aot-partitioned-checkpoints.
             job_name (str): Training job name. If not specified, a unique training job
                         name will be created.
+            volume_size (int): Size in GB of the storage volume to use for
+                storing input and output data during training (default: 30).
             volume_kms_key (str): Optional. KMS key ID for encrypting EBS
                 volume attached to the training instance (default: None).
             output_kms_key (str): Optional. KMS key ID for encrypting the
@@ -499,20 +506,19 @@ def partition(
             region_name = self.sagemaker_session.boto_session.region_name
             self.image_uri = self.serving_image_uri(region_name)
 
-        deploy_key_prefix = fw_utils.model_code_key_prefix(
-            self.key_prefix, self.name, self.image_uri
-        )
         if s3_output_uri is None:
+            deploy_key_prefix = fw_utils.model_code_key_prefix(
+                self.key_prefix, self.name, self.image_uri
+            )
+
             bucket, deploy_key_prefix = s3.determine_bucket_and_prefix(
                 bucket=self.bucket,
                 key_prefix=deploy_key_prefix,
                 sagemaker_session=self.sagemaker_session,
             )
             s3_output_uri = s3_path_join("s3://", bucket, deploy_key_prefix)
-        else:
-            s3_output_uri = s3_path_join(s3_output_uri, deploy_key_prefix)
 
-        self.save_mp_checkpoint_path = s3_path_join(s3_output_uri, "aot-partitioned-checkpoints")
+        self.save_mp_checkpoint_path = s3_path_join(s3_output_uri, s3_output_prefix)
 
         container_def = self._upload_model_to_s3(upload_as_tar=False)
         estimator = _create_estimator(
@@ -521,6 +527,7 @@ def partition(
             image_uri=self.image_uri,
             role=self.role,
             sagemaker_session=self.sagemaker_session,
+            volume_size=volume_size,
             vpc_config=self.vpc_config,
             volume_kms_key=volume_kms_key,
             output_kms_key=output_kms_key,
@@ -924,7 +931,9 @@ def partition(
         self,
         instance_type: str,
         s3_output_uri: str = None,
+        s3_output_prefix: str = "aot-partitioned-checkpoints",
         job_name: Optional[str] = None,
+        volume_size: int = 30,
         volume_kms_key: Optional[str] = None,
         output_kms_key: Optional[str] = None,
         use_spot_instances: bool = False,
@@ -940,8 +949,13 @@ def partition(
                     artifacts and output files). If not specified, results are
                     stored to a default bucket. If the bucket with the specific name
                     does not exist, it will be created.
+            s3_output_prefix (str): Name of the prefix where all the partitioned
+                    checkpoints to be uploaded. If not provided, the default value is
+                    aot-partitioned-checkpoints.
             job_name (str): Training job name. If not specified, a unique training job
                         name will be created.
+            volume_size (int): Size in GB of the storage volume to use for
+                storing input and output data during training (default: 30).
             volume_kms_key (str): Optional. KMS key ID for encrypting EBS
                 volume attached to the training instance (default: None).
             output_kms_key (str): Optional. KMS key ID for encrypting the
@@ -969,7 +983,9 @@ def partition(
         super(DeepSpeedModel, self).partition(
             instance_type,
             s3_output_uri,
-            job_name,
+            s3_output_prefix=s3_output_prefix,
+            job_name=job_name,
+            volume_size=volume_size,
             volume_kms_key=volume_kms_key,
             output_kms_key=output_kms_key,
             use_spot_instances=use_spot_instances,
@@ -1096,7 +1112,9 @@ def partition(
         self,
         instance_type: str,
         s3_output_uri: str = None,
+        s3_output_prefix: str = "aot-partitioned-checkpoints",
         job_name: Optional[str] = None,
+        volume_size: int = 30,
         volume_kms_key: Optional[str] = None,
         output_kms_key: Optional[str] = None,
         use_spot_instances: bool = False,
@@ -1112,8 +1130,13 @@ def partition(
                     artifacts and output files). If not specified, results are
                     stored to a default bucket. If the bucket with the specific name
                     does not exist, it will be created.
+            s3_output_prefix (str): Name of the prefix where all the partitioned
+                    checkpoints to be uploaded. If not provided, the default value is
+                    aot-partitioned-checkpoints.
             job_name (str): Training job name. If not specified, a unique training job
                         name will be created.
+            volume_size (int): Size in GB of the storage volume to use for
+                storing input and output data during training (default: 30).
             volume_kms_key (str): Optional. KMS key ID for encrypting EBS
                 volume attached to the training instance (default: None).
             output_kms_key (str): Optional. KMS key ID for encrypting the

src/sagemaker/image_uri_config/djl-deepspeed.json

@@ -30,7 +30,7 @@
                 "us-west-2": "763104351884"
             },
             "repository": "djl-inference",
-            "tag_prefix": "0.22.1-deepspeed0.8.3-cu118"
+            "tag_prefix": "0.22.1-deepspeed0.9.2-cu118"
         },
         "0.21.0": {
             "registries": {

src/sagemaker/model_monitor/clarify_model_monitoring.py

@@ -175,6 +175,7 @@ def _create_baselining_processor(self):
             network_config=self.network_config,
         )
         baselining_processor.image_uri = self.image_uri
+        baselining_processor.base_job_name = self.base_job_name
         return baselining_processor
 
     def _upload_analysis_config(self, analysis_config, output_s3_uri, job_definition_name):

src/sagemaker/workflow/_utils.py

@@ -34,6 +34,7 @@
 )
 from sagemaker.utils import _save_model, download_file_from_url
 from sagemaker.workflow.retry import RetryPolicy
+from sagemaker.workflow.utilities import trim_request_dict
 
 if TYPE_CHECKING:
     from sagemaker.workflow.step_collections import StepCollection
@@ -412,6 +413,8 @@ def __init__(
     @property
     def arguments(self) -> RequestType:
         """The arguments dict that are used to call `create_model_package`."""
+        from sagemaker.workflow.utilities import _pipeline_config
+
         model_name = self.name
 
         if self.step_args:
@@ -492,9 +495,9 @@ def arguments(self) -> RequestType:
         if "Description" in request_dict:
             request_dict.pop("Description")
             logger.warning(warn_msg_template, "Description")
-        if "ModelPackageName" in request_dict:
-            request_dict.pop("ModelPackageName")
-            logger.warning(warn_msg_template, "ModelPackageName")
+
+        # Continue to pop job name if not explicitly opted-in via config
+        request_dict = trim_request_dict(request_dict, "ModelPackageName", _pipeline_config)
 
         return request_dict
 

src/sagemaker/workflow/automl_step.py

@@ -23,7 +23,7 @@
 from sagemaker.workflow.properties import Properties
 from sagemaker.workflow.retry import RetryPolicy
 from sagemaker.workflow.steps import ConfigurableRetryStep, CacheConfig, Step, StepTypeEnum
-from sagemaker.workflow.utilities import validate_step_args_input
+from sagemaker.workflow.utilities import validate_step_args_input, trim_request_dict
 from sagemaker.workflow.step_collections import StepCollection
 
 
@@ -89,10 +89,11 @@ def arguments(self) -> RequestType:
         NOTE: The `CreateAutoMLJob` request is not quite the
             args list that workflow needs.
 
-        The `AutoMLJobName`, `ModelDeployConfig` and `GenerateCandidateDefinitionsOnly`
+        `ModelDeployConfig` and `GenerateCandidateDefinitionsOnly`
             attribute cannot be included.
         """
         from sagemaker.workflow.utilities import execute_job_functions
+        from sagemaker.workflow.utilities import _pipeline_config
 
         # execute fit function in AutoML with saved parameters,
         # and store args in PipelineSession's _context
@@ -114,7 +115,10 @@ def arguments(self) -> RequestType:
             request_dict.pop("ModelDeployConfig", None)
         if "GenerateCandidateDefinitionsOnly" in request_dict:
             request_dict.pop("GenerateCandidateDefinitionsOnly", None)
-        request_dict.pop("AutoMLJobName", None)
+        # Continue to pop job name if not explicitly opted-in via config
+        # AutoML Trims to AutoMLJo-2023-06-23-22-57-39-083
+        request_dict = trim_request_dict(request_dict, "AutoMLJobName", _pipeline_config)
+
         return request_dict
 
     @property

src/sagemaker/workflow/clarify_check_step.py

@@ -44,6 +44,7 @@
 from sagemaker.workflow.step_collections import StepCollection
 from sagemaker.workflow.steps import Step, StepTypeEnum, CacheConfig
 from sagemaker.workflow.check_job_config import CheckJobConfig
+from sagemaker.workflow.utilities import trim_request_dict
 
 _DATA_BIAS_TYPE = "DATA_BIAS"
 _MODEL_BIAS_TYPE = "MODEL_BIAS"
@@ -253,6 +254,8 @@ def __init__(
     @property
     def arguments(self) -> RequestType:
         """The arguments dict that is used to define the ClarifyCheck step."""
+        from sagemaker.workflow.utilities import _pipeline_config
+
         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"]],
@@ -266,8 +269,8 @@ def arguments(self) -> RequestType:
         request_dict = self._baselining_processor.sagemaker_session._get_process_request(
             **process_args
         )
-        if "ProcessingJobName" in request_dict:
-            request_dict.pop("ProcessingJobName")
+        # Continue to pop job name if not explicitly opted-in via config
+        request_dict = trim_request_dict(request_dict, "ProcessingJobName", _pipeline_config)
 
         return request_dict
 

src/sagemaker/workflow/pipeline.py

@@ -37,6 +37,7 @@
 )
 from sagemaker.workflow.execution_variables import ExecutionVariables
 from sagemaker.workflow.parameters import Parameter
+from sagemaker.workflow.pipeline_definition_config import PipelineDefinitionConfig
 from sagemaker.workflow.pipeline_experiment_config import PipelineExperimentConfig
 from sagemaker.workflow.parallelism_config import ParallelismConfiguration
 from sagemaker.workflow.properties import Properties
@@ -52,6 +53,8 @@
     ExecutionVariables.PIPELINE_NAME, ExecutionVariables.PIPELINE_EXECUTION_ID
 )
 
+_DEFAULT_DEFINITION_CFG = PipelineDefinitionConfig(use_custom_job_prefix=False)
+
 
 class Pipeline(Entity):
     """Pipeline for workflow."""
@@ -63,6 +66,7 @@ def __init__(
         pipeline_experiment_config: Optional[PipelineExperimentConfig] = _DEFAULT_EXPERIMENT_CFG,
         steps: Optional[Sequence[Union[Step, StepCollection]]] = None,
         sagemaker_session: Optional[Session] = None,
+        pipeline_definition_config: Optional[PipelineDefinitionConfig] = _DEFAULT_DEFINITION_CFG,
     ):
         """Initialize a Pipeline
 
@@ -84,12 +88,16 @@ def __init__(
             sagemaker_session (sagemaker.session.Session): Session object that manages interactions
                 with Amazon SageMaker APIs and any other AWS services needed. If not specified, the
                 pipeline creates one using the default AWS configuration chain.
+            pipeline_definition_config (Optional[PipelineDefinitionConfig]): If set,
+                the workflow customizes the pipeline definition using the configurations
+                specified. By default, custom job-prefixing is turned off.
         """
         self.name = name
         self.parameters = parameters if parameters else []
         self.pipeline_experiment_config = pipeline_experiment_config
         self.steps = steps if steps else []
         self.sagemaker_session = sagemaker_session if sagemaker_session else Session()
+        self.pipeline_definition_config = pipeline_definition_config
 
         self._version = "2020-12-01"
         self._metadata = dict()
@@ -105,7 +113,11 @@ def to_request(self) -> RequestType:
             "PipelineExperimentConfig": self.pipeline_experiment_config.to_request()
             if self.pipeline_experiment_config is not None
             else None,
-            "Steps": build_steps(self.steps, self.name),
+            "Steps": build_steps(
+                self.steps,
+                self.name,
+                self.pipeline_definition_config,
+            ),
         }
 
     def create(