feature: handler for stopping transform job

Author: imujjwal96

src/sagemaker/session.py

@@ -902,6 +902,27 @@ def wait_for_transform_job(self, job, poll=5):
         self._check_job_status(job, desc, 'TransformJobStatus')
         return desc
 
+    def stop_transform_job(self, name):
+        """Stop the Amazon SageMaker hyperparameter tuning job with the specified name.
+
+        Args:
+            name (str): Name of the Amazon SageMaker batch transform job.
+
+        Raises:
+            ClientError: If an error occurs while trying to stop the batch transform job.
+        """
+        try:
+            LOGGER.info('Stopping transform job: {}'.format(name))
+            self.sagemaker_client.stop_transform_job(TransformJobName=name)
+        except ClientError as e:
+            error_code = e.response['Error']['Code']
+            # allow to pass if the job already stopped
+            if error_code == 'ValidationException':
+                LOGGER.info('Transform job: {} is already stopped or not running.'.format(name))
+            else:
+                LOGGER.error('Error occurred while attempting to stop transform job: {}.'.format(name))
+                raise
+
     def _check_job_status(self, job, desc, status_key_name):
         """Check to see if the job completed successfully and, if not, construct and
         raise a ValueError.

src/sagemaker/transformer.py

@@ -156,6 +156,12 @@ def wait(self):
         self._ensure_last_transform_job()
         self.latest_transform_job.wait()
 
+    def stop_transform_job(self):
+        """Stop latest running batch transform job.
+        """
+        self._ensure_last_transform_job()
+        self.latest_transform_job.stop()
+
     def _ensure_last_transform_job(self):
         if self.latest_transform_job is None:
             raise ValueError('No transform job available')
@@ -230,6 +236,9 @@ def start_new(cls, transformer, data, data_type, content_type, compression_type,
     def wait(self):
         self.sagemaker_session.wait_for_transform_job(self.job_name)
 
+    def stop(self):
+        self.sagemaker_session.stop_transform_job(name=self.job_name)
+
     @staticmethod
     def _load_config(data, data_type, content_type, compression_type, split_type, transformer):
         input_config = _TransformJob._format_inputs_to_input_config(data, data_type, content_type,