fix: Fix Pipeline variables related customer issues

Author: Dewen Qi

src/sagemaker/estimator.py

@@ -1869,7 +1869,9 @@ def _add_spot_checkpoint_args(cls, local_mode, estimator, train_args):
         if estimator.use_spot_instances:
             if local_mode:
                 raise ValueError("Spot training is not supported in local mode.")
-            train_args["use_spot_instances"] = True
+            # estimator.use_spot_instances may be a Pipeline ParameterBoolean object
+            # which is parsed during the Pipeline execution runtime
+            train_args["use_spot_instances"] = estimator.use_spot_instances
 
         if estimator.checkpoint_s3_uri:
             if local_mode:

src/sagemaker/model.py

@@ -37,6 +37,7 @@
 from sagemaker.utils import unique_name_from_base
 from sagemaker.async_inference import AsyncInferenceConfig
 from sagemaker.predictor_async import AsyncPredictor
+from sagemaker.workflow import is_pipeline_variable
 
 LOGGER = logging.getLogger("sagemaker")
 
@@ -443,7 +444,7 @@ def _upload_code(self, key_prefix: str, repack: bool = False) -> None:
             )
 
         if repack and self.model_data is not None and self.entry_point is not None:
-            if isinstance(self.model_data, sagemaker.workflow.properties.Properties):
+            if is_pipeline_variable(self.model_data):
                 # model is not yet there, defer repacking to later during pipeline execution
                 return
 

src/sagemaker/parameter.py

@@ -14,9 +14,8 @@
 from __future__ import absolute_import
 
 import json
-from sagemaker.workflow.parameters import Parameter as PipelineParameter
-from sagemaker.workflow.functions import JsonGet as PipelineJsonGet
-from sagemaker.workflow.functions import Join as PipelineJoin
+
+from sagemaker.workflow import is_pipeline_variable
 
 
 class ParameterRange(object):
@@ -73,10 +72,10 @@ def as_tuning_range(self, name):
         return {
             "Name": name,
             "MinValue": str(self.min_value)
-            if not isinstance(self.min_value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
+            if not is_pipeline_variable(self.min_value)
             else self.min_value,
             "MaxValue": str(self.max_value)
-            if not isinstance(self.max_value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
+            if not is_pipeline_variable(self.max_value)
             else self.max_value,
             "ScalingType": self.scaling_type,
         }
@@ -111,10 +110,7 @@ def __init__(self, values):  # pylint: disable=super-init-not-called
                 This input will be converted into a list of strings.
         """
         values = values if isinstance(values, list) else [values]
-        self.values = [
-            str(v) if not isinstance(v, (PipelineParameter, PipelineJsonGet, PipelineJoin)) else v
-            for v in values
-        ]
+        self.values = [str(v) if not is_pipeline_variable(v) else v for v in values]
 
     def as_tuning_range(self, name):
         """Represent the parameter range as a dictionary.

src/sagemaker/session.py

@@ -763,6 +763,8 @@ def _get_train_request(  # noqa: C901
             train_request["EnableInterContainerTrafficEncryption"] = encrypt_inter_container_traffic
 
         if use_spot_instances:
+            # estimator.use_spot_instances may be a Pipeline ParameterBoolean object
+            # which is parsed during the Pipeline execution runtime
             train_request["EnableManagedSpotTraining"] = use_spot_instances
 
         if checkpoint_s3_uri:
@@ -2338,13 +2340,17 @@ def _map_training_config(
             training_job_definition["VpcConfig"] = vpc_config
 
         if enable_network_isolation:
-            training_job_definition["EnableNetworkIsolation"] = True
+            training_job_definition["EnableNetworkIsolation"] = enable_network_isolation
 
         if encrypt_inter_container_traffic:
-            training_job_definition["EnableInterContainerTrafficEncryption"] = True
+            training_job_definition[
+                "EnableInterContainerTrafficEncryption"
+            ] = encrypt_inter_container_traffic
 
         if use_spot_instances:
-            training_job_definition["EnableManagedSpotTraining"] = True
+            # use_spot_instances may be a Pipeline ParameterBoolean object
+            # which is parsed during the Pipeline execution runtime
+            training_job_definition["EnableManagedSpotTraining"] = use_spot_instances
 
         if checkpoint_s3_uri:
             checkpoint_config = {"S3Uri": checkpoint_s3_uri}

src/sagemaker/tensorflow/model.py

@@ -21,6 +21,7 @@
 from sagemaker.deprecations import removed_kwargs
 from sagemaker.predictor import Predictor
 from sagemaker.serializers import JSONSerializer
+from sagemaker.workflow import is_pipeline_variable
 
 
 class TensorFlowPredictor(Predictor):
@@ -326,7 +327,9 @@ def prepare_container_def(self, instance_type=None, accelerator_type=None):
         image_uri = self._get_image_uri(instance_type, accelerator_type)
         env = self._get_container_env()
 
-        if self.entry_point:
+        # If self.model_data is pipeline variable, model is not yet there.
+        # So defer repacking to later during pipeline execution
+        if self.entry_point and not is_pipeline_variable(self.model_data):
             key_prefix = sagemaker.fw_utils.model_code_key_prefix(
                 self.key_prefix, self.name, image_uri
             )

src/sagemaker/tuner.py

@@ -38,9 +38,7 @@
     IntegerParameter,
     ParameterRange,
 )
-from sagemaker.workflow.parameters import Parameter as PipelineParameter
-from sagemaker.workflow.functions import JsonGet as PipelineJsonGet
-from sagemaker.workflow.functions import Join as PipelineJoin
+from sagemaker.workflow import is_pipeline_variable
 
 from sagemaker.session import Session
 from sagemaker.utils import base_from_name, base_name_from_image, name_from_base
@@ -63,18 +61,6 @@
 logger = logging.getLogger(__name__)
 
 
-def is_pipeline_parameters(value):
-    """Determine if a value is a pipeline parameter or function representation
-
-    Args:
-        value (float or int): The value to be verified.
-
-    Returns:
-        bool: True if it is, False otherwise.
-    """
-    return isinstance(value, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-
-
 class WarmStartTypes(Enum):
     """Warm Start Configuration type.
 
@@ -376,9 +362,7 @@ def _prepare_static_hyperparameters(
         """Prepare static hyperparameters for one estimator before tuning."""
         # Remove any hyperparameter that will be tuned
         static_hyperparameters = {
-            str(k): str(v)
-            if not isinstance(v, (PipelineParameter, PipelineJsonGet, PipelineJoin))
-            else v
+            str(k): str(v) if not is_pipeline_variable(v) else v
             for (k, v) in estimator.hyperparameters().items()
         }
         for hyperparameter_name in hyperparameter_ranges.keys():

src/sagemaker/workflow/__init__.py

@@ -20,3 +20,15 @@
 from sagemaker.workflow.properties import Properties
 
 PipelineNonPrimitiveInputTypes = Union[ExecutionVariable, Expression, Parameter, Properties]
+
+
+def is_pipeline_variable(var: object) -> bool:
+    """Check if the variable is a pipeline Parameter/Properties/Expression/ExecutionVariable
+
+    Args:
+        var (object): The variable to be verified.
+
+    Returns:
+         bool: True if it is, False otherwise.
+    """
+    return isinstance(var, (Properties, Parameter, Expression, ExecutionVariable))

src/sagemaker/workflow/_repack_model.py

@@ -39,14 +39,14 @@ def repack(inference_script, model_archive, dependencies=None, source_dir=None):
 
     Args:
         inference_script (str): The path to the custom entry point.
-        model_archive (str): The name of the model TAR archive.
+        model_archive (str): The name or path (e.g. s3 uri) of the model TAR archive.
         dependencies (str): A space-delimited string of paths to custom dependencies.
         source_dir (str): The path to a custom source directory.
     """
 
     # the data directory contains a model archive generated by a previous training job
     data_directory = "/opt/ml/input/data/training"
-    model_path = os.path.join(data_directory, model_archive)
+    model_path = os.path.join(data_directory, model_archive.split("/")[-1])
 
     # create a temporary directory
     with tempfile.TemporaryDirectory() as tmp:

src/sagemaker/workflow/_utils.py

@@ -137,12 +137,6 @@ def __init__(
         self._model_data = model_data
         self.sagemaker_session = sagemaker_session
         self.role = role
-        if isinstance(model_data, Properties):
-            self._model_prefix = model_data
-            self._model_archive = "model.tar.gz"
-        else:
-            self._model_prefix = "/".join(self._model_data.split("/")[:-1])
-            self._model_archive = self._model_data.split("/")[-1]
         self._entry_point = entry_point
         self._entry_point_basename = os.path.basename(self._entry_point)
         self._source_dir = source_dir
@@ -164,7 +158,7 @@ def __init__(
             role=self.role,
             hyperparameters={
                 "inference_script": self._entry_point_basename,
-                "model_archive": self._model_archive,
+                "model_archive": self._model_data,
                 "dependencies": dependencies_hyperparameter,
                 "source_dir": self._source_dir,
             },
@@ -173,7 +167,7 @@ def __init__(
             **kwargs,
         )
         repacker.disable_profiler = True
-        inputs = TrainingInput(self._model_prefix)
+        inputs = TrainingInput(self._model_data)
 
         # super!
         super(_RepackModelStep, self).__init__(

src/sagemaker/workflow/airflow.py

@@ -184,7 +184,9 @@ def training_base_config(estimator, inputs=None, job_name=None, mini_batch_size=
         train_config["VpcConfig"] = job_config["vpc_config"]
 
     if estimator.use_spot_instances:
-        train_config["EnableManagedSpotTraining"] = True
+        # estimator.use_spot_instances may be a Pipeline ParameterBoolean object
+        # which is parsed during the Pipeline execution runtime
+        train_config["EnableManagedSpotTraining"] = estimator.use_spot_instances
 
     if estimator.hyperparameters() is not None:
         hyperparameters = {str(k): str(v) for (k, v) in estimator.hyperparameters().items()}

src/sagemaker/workflow/clarify_check_step.py

@@ -37,8 +37,8 @@
 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
-from sagemaker.workflow.entities import RequestType, Expression
+from sagemaker.workflow import PipelineNonPrimitiveInputTypes, is_pipeline_variable
+from sagemaker.workflow.entities import RequestType
 from sagemaker.workflow.properties import Properties
 from sagemaker.workflow.steps import Step, StepTypeEnum, CacheConfig
 from sagemaker.workflow.check_job_config import CheckJobConfig
@@ -193,18 +193,15 @@ def __init__(
                 + "DataBiasCheckConfig, ModelBiasCheckConfig or ModelExplainabilityCheckConfig"
             )
 
-        if isinstance(
-            clarify_check_config.data_config.s3_analysis_config_output_path,
-            (ExecutionVariable, Expression, Parameter, Properties),
-        ):
+        if is_pipeline_variable(clarify_check_config.data_config.s3_analysis_config_output_path):
             raise RuntimeError(
                 "s3_analysis_config_output_path cannot be of type "
                 + "ExecutionVariable/Expression/Parameter/Properties"
             )
 
-        if not clarify_check_config.data_config.s3_analysis_config_output_path and isinstance(
-            clarify_check_config.data_config.s3_output_path,
-            (ExecutionVariable, Expression, Parameter, Properties),
+        if (
+            not clarify_check_config.data_config.s3_analysis_config_output_path
+            and is_pipeline_variable(clarify_check_config.data_config.s3_output_path)
         ):
             raise RuntimeError(
                 "`s3_output_path` cannot be of type ExecutionVariable/Expression/Parameter"

src/sagemaker/workflow/conditions.py

@@ -22,6 +22,7 @@
 
 import attr
 
+from sagemaker.workflow import is_pipeline_variable
 from sagemaker.workflow.entities import (
     DefaultEnumMeta,
     Entity,
@@ -261,6 +262,6 @@ def primitive_or_expr(
     Returns:
         Either the expression of the value or the primitive value.
     """
-    if isinstance(value, (ExecutionVariable, Expression, Parameter, Properties)):
+    if is_pipeline_variable(value):
         return value.expr
     return value

src/sagemaker/workflow/entities.py

@@ -57,3 +57,16 @@ class Expression(abc.ABC):
     @abc.abstractmethod
     def expr(self) -> RequestType:
         """Get the expression structure for workflow service calls."""
+
+    def startswith(self, prefix, start=None, end=None) -> bool:  # pylint: disable=unused-argument
+        """Simulate the Python string's built-in method: startswith
+
+        Args:
+            prefix (str, tuple): The (tuple of) string to be checked.
+            start (int): To set the start index of the matching boundary (default: None).
+            end (int): To set the end index of the matching boundary (default: None).
+
+        Return:
+            bool: always return False as Pipeline variables are parsed during execution runtime
+        """
+        return False

src/sagemaker/workflow/quality_check_step.py

@@ -22,9 +22,9 @@
 from sagemaker import s3
 from sagemaker.model_monitor import ModelMonitor
 from sagemaker.processing import ProcessingOutput, ProcessingJob, Processor, ProcessingInput
-from sagemaker.workflow import PipelineNonPrimitiveInputTypes, ExecutionVariable, Parameter
+from sagemaker.workflow import PipelineNonPrimitiveInputTypes, is_pipeline_variable
 
-from sagemaker.workflow.entities import RequestType, Expression
+from sagemaker.workflow.entities import RequestType
 from sagemaker.workflow.properties import (
     Properties,
 )
@@ -279,7 +279,7 @@ def _generate_baseline_job_inputs(self):
                 _CONTAINER_BASE_PATH, _CONTAINER_INPUT_PATH, _BASELINE_DATASET_INPUT_NAME
             )
         )
-        if isinstance(baseline_dataset, (ExecutionVariable, Expression, Parameter, Properties)):
+        if is_pipeline_variable(baseline_dataset):
             baseline_dataset_input = ProcessingInput(
                 source=self.quality_check_config.baseline_dataset,
                 destination=baseline_dataset_des,