Merge branch 'master' into update-hf-pt-2.0

Author: JingyaHuang

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## v2.162.0 (2023-06-06)
+
+### Features
+
+ * Add tagging support for create ir job
+ * Selective Step Execution feature in Pipelines
+ * Add Neuronx Image uri - Transformers 4.28 - PyTorch 1.13
+
+### Bug Fixes and Other Changes
+
+ * skip pipelines abalone notebook test
+ * Update neo multiversion support to include edge devices
+
+### Documentation Changes
+
+ * JumpStart Utility Doc Update
+
 ## v2.161.0 (2023-06-01)
 
 ### Features

VERSION

@@ -1 +1 @@
-2.161.1.dev0
+2.162.1.dev0

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

@@ -110,6 +110,12 @@ Pipeline Experiment Config
 
 .. autoclass:: sagemaker.workflow.pipeline_experiment_config.PipelineExperimentConfigProperty
 
+Selective Execution Config
+--------------------------
+
+.. autoclass:: sagemaker.workflow.selective_execution_config.SelectiveExecutionConfig
+
+
 Properties
 ----------
 

src/sagemaker/local/local_session.py

@@ -505,6 +505,8 @@ def start_pipeline_execution(self, PipelineName, **kwargs):
         """
         if "ParallelismConfiguration" in kwargs:
             logger.warning("Parallelism configuration is not supported in local mode.")
+        if "SelectiveExecutionConfig" in kwargs:
+            raise ValueError("SelectiveExecutionConfig is not supported in local mode.")
         if PipelineName not in LocalSagemakerClient._pipelines:
             error_response = {
                 "Error": {

src/sagemaker/session.py

@@ -5313,6 +5313,7 @@ def _create_inference_recommendations_job_request(
         framework: str,
         sample_payload_url: str,
         supported_content_types: List[str],
+        tags: Dict[str, str],
         model_name: str = None,
         model_package_version_arn: str = None,
         job_duration_in_seconds: int = None,
@@ -5348,6 +5349,8 @@ def _create_inference_recommendations_job_request(
                 benchmarked by Amazon SageMaker Inference Recommender that matches your model.
             supported_instance_types (List[str]): A list of the instance types that are used
                 to generate inferences in real-time.
+            tags (Dict[str, str]): Tags used to identify where the Inference Recommendatons Call
+                was made from.
             endpoint_configurations (List[Dict[str, any]]): Specifies the endpoint configurations
                 to use for a job. Will be used for `Advanced` jobs.
             traffic_pattern (Dict[str, any]): Specifies the traffic pattern for the job.
@@ -5386,6 +5389,7 @@ def _create_inference_recommendations_job_request(
             "InputConfig": {
                 "ContainerConfig": containerConfig,
             },
+            "Tags": tags,
         }
 
         request.get("InputConfig").update(
@@ -5477,6 +5481,8 @@ def create_inference_recommendations_job(
             job_name = "SMPYTHONSDK-" + str(unique_tail)
         job_description = "#python-sdk-create"
 
+        tags = [{"Key": "ClientType", "Value": "PythonSDK-RightSize"}]
+
         create_inference_recommendations_job_request = (
             self._create_inference_recommendations_job_request(
                 role=role,
@@ -5496,6 +5502,7 @@ def create_inference_recommendations_job(
                 traffic_pattern=traffic_pattern,
                 stopping_conditions=stopping_conditions,
                 resource_limit=resource_limit,
+                tags=tags,
             )
         )
 

src/sagemaker/workflow/pipeline.py

@@ -40,6 +40,7 @@
 from sagemaker.workflow.pipeline_experiment_config import PipelineExperimentConfig
 from sagemaker.workflow.parallelism_config import ParallelismConfiguration
 from sagemaker.workflow.properties import Properties
+from sagemaker.workflow.selective_execution_config import SelectiveExecutionConfig
 from sagemaker.workflow.steps import Step, StepTypeEnum
 from sagemaker.workflow.step_collections import StepCollection
 from sagemaker.workflow.condition_step import ConditionStep
@@ -312,6 +313,7 @@ def start(
         execution_display_name: str = None,
         execution_description: str = None,
         parallelism_config: ParallelismConfiguration = None,
+        selective_execution_config: SelectiveExecutionConfig = None,
     ):
         """Starts a Pipeline execution in the Workflow service.
 
@@ -323,16 +325,26 @@ def start(
             parallelism_config (Optional[ParallelismConfiguration]): Parallelism configuration
                 that is applied to each of the executions of the pipeline. It takes precedence
                 over the parallelism configuration of the parent pipeline.
+            selective_execution_config (Optional[SelectiveExecutionConfig]): The configuration for
+                selective step execution.
 
         Returns:
             A `_PipelineExecution` instance, if successful.
         """
+        if selective_execution_config is not None:
+            if selective_execution_config.source_pipeline_execution_arn is None:
+                selective_execution_config.source_pipeline_execution_arn = (
+                    self._get_latest_execution_arn()
+                )
+            selective_execution_config = selective_execution_config.to_request()
+
         kwargs = dict(PipelineName=self.name)
         update_args(
             kwargs,
             PipelineExecutionDescription=execution_description,
             PipelineExecutionDisplayName=execution_display_name,
             ParallelismConfiguration=parallelism_config,
+            SelectiveExecutionConfig=selective_execution_config,
         )
         if self.sagemaker_session.local_mode:
             update_args(kwargs, PipelineParameters=parameters)
@@ -388,6 +400,57 @@ def _interpolate_step_collection_name_in_depends_on(self, step_requests: list):
                 )
                 self._interpolate_step_collection_name_in_depends_on(sub_step_requests)
 
+    def list_executions(
+        self,
+        sort_by: str = None,
+        sort_order: str = None,
+        max_results: int = None,
+        next_token: str = None,
+    ) -> Dict[str, Any]:
+        """Lists a pipeline's executions.
+
+        Args:
+            sort_by (str): The field by which to sort results(CreationTime/PipelineExecutionArn).
+            sort_order (str): The sort order for results (Ascending/Descending).
+            max_results (int): The maximum number of pipeline executions to return in the response.
+            next_token (str):  If the result of the previous ListPipelineExecutions request was
+                truncated, the response includes a NextToken. To retrieve the next set of pipeline
+                executions, use the token in the next request.
+
+        Returns:
+            List of Pipeline Execution Summaries. See
+            boto3 client list_pipeline_executions
+            https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sagemaker.html#SageMaker.Client.list_pipeline_executions
+        """
+        kwargs = dict(PipelineName=self.name)
+        update_args(
+            kwargs,
+            SortBy=sort_by,
+            SortOrder=sort_order,
+            NextToken=next_token,
+            MaxResults=max_results,
+        )
+        response = self.sagemaker_session.sagemaker_client.list_pipeline_executions(**kwargs)
+
+        # Return only PipelineExecutionSummaries and NextToken from the list_pipeline_executions
+        # response
+        return {
+            key: response[key]
+            for key in ["PipelineExecutionSummaries", "NextToken"]
+            if key in response
+        }
+
+    def _get_latest_execution_arn(self):
+        """Retrieves the latest execution of this pipeline"""
+        response = self.list_executions(
+            sort_by="CreationTime",
+            sort_order="Descending",
+            max_results=1,
+        )
+        if response["PipelineExecutionSummaries"]:
+            return response["PipelineExecutionSummaries"][0]["PipelineExecutionArn"]
+        return None
+
 
 def format_start_parameters(parameters: Dict[str, Any]) -> List[Dict[str, Any]]:
     """Formats start parameter overrides as a list of dicts.

src/sagemaker/workflow/selective_execution_config.py

@@ -0,0 +1,56 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You
+# may not use this file except in compliance with the License. A copy of
+# the License is located at
+#
+#     http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file is
+# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific
+# language governing permissions and limitations under the License.
+"""Pipeline Parallelism Configuration"""
+from __future__ import absolute_import
+from typing import List
+from sagemaker.workflow.entities import RequestType
+
+
+class SelectiveExecutionConfig:
+    """The selective execution configuration, which defines a subset of pipeline steps to run in
+
+    another SageMaker pipeline run.
+    """
+
+    def __init__(self, selected_steps: List[str], source_pipeline_execution_arn: str = None):
+        """Create a `SelectiveExecutionConfig`.
+
+        Args:
+            source_pipeline_execution_arn (str): The ARN from a reference execution of the
+                current pipeline. Used to copy input collaterals needed for the selected
+                steps to run. The execution status of the pipeline can be `Stopped`, `Failed`, or
+                `Succeeded`.
+            selected_steps (List[str]): A list of pipeline steps to run. All step(s) in all
+                path(s) between two selected steps should be included.
+        """
+        self.source_pipeline_execution_arn = source_pipeline_execution_arn
+        self.selected_steps = selected_steps
+
+    def _build_selected_steps_from_list(self) -> RequestType:
+        """Get the request structure for the list of selected steps."""
+        selected_step_list = []
+        for selected_step in self.selected_steps:
+            selected_step_list.append(dict(StepName=selected_step))
+        return selected_step_list
+
+    def to_request(self) -> RequestType:
+        """Convert `SelectiveExecutionConfig` object to request dict."""
+        request = {}
+
+        if self.source_pipeline_execution_arn is not None:
+            request["SourcePipelineExecutionArn"] = self.source_pipeline_execution_arn
+
+        if self.selected_steps is not None:
+            request["SelectedSteps"] = self._build_selected_steps_from_list()
+
+        return request