Merge branch 'zwei' into fix-pip-install-cmd

Author: laurenyu

src/sagemaker/chainer/estimator.py

@@ -216,7 +216,6 @@ def create_model(
             role or self.role,
             entry_point or self._model_entry_point(),
             source_dir=(source_dir or self._model_source_dir()),
-            enable_cloudwatch_metrics=self.enable_cloudwatch_metrics,
             container_log_level=self.container_log_level,
             code_location=self.code_location,
             py_version=self.py_version,

src/sagemaker/deserializers.py

@@ -22,6 +22,13 @@
 
 import numpy as np
 
+from sagemaker.utils import DeferredError
+
+try:
+    import pandas
+except ImportError as e:
+    pandas = DeferredError(e)
+
 
 class BaseDeserializer(abc.ABC):
     """Abstract base class for creation of new deserializers.
@@ -208,3 +215,31 @@ def deserialize(self, stream, content_type):
             return json.load(codecs.getreader("utf-8")(stream))
         finally:
             stream.close()
+
+
+class PandasDeserializer(BaseDeserializer):
+    """Deserialize CSV or JSON data from an inference endpoint into a pandas dataframe."""
+
+    ACCEPT = "text/csv"
+
+    def deserialize(self, stream, content_type):
+        """Deserialize CSV or JSON data from an inference endpoint into a pandas
+        dataframe.
+
+        If the data is JSON, the data should be formatted in the 'columns' orient.
+        See https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_json.html
+
+        Args:
+            stream (botocore.response.StreamingBody): Data to be deserialized.
+            content_type (str): The MIME type of the data.
+
+        Returns:
+            pandas.DataFrame: The data deserialized into a pandas DataFrame.
+        """
+        if content_type == "text/csv":
+            return pandas.read_csv(stream)
+
+        if content_type == "application/json":
+            return pandas.read_json(stream)
+
+        raise ValueError("%s cannot read content type %s." % (__class__.__name__, content_type))

src/sagemaker/estimator.py

@@ -17,7 +17,6 @@
 import logging
 import os
 import uuid
-import warnings
 from abc import ABCMeta
 from abc import abstractmethod
 
@@ -46,7 +45,6 @@
 from sagemaker.model import (
     SCRIPT_PARAM_NAME,
     DIR_PARAM_NAME,
-    CLOUDWATCH_METRICS_PARAM_NAME,
     CONTAINER_LOG_LEVEL_PARAM_NAME,
     JOB_NAME_PARAM_NAME,
     SAGEMAKER_REGION_PARAM_NAME,
@@ -1433,7 +1431,6 @@ def __init__(
         entry_point,
         source_dir=None,
         hyperparameters=None,
-        enable_cloudwatch_metrics=False,
         container_log_level=logging.INFO,
         code_location=None,
         image_uri=None,
@@ -1491,9 +1488,6 @@ def __init__(
                 SageMaker. For convenience, this accepts other types for keys
                 and values, but ``str()`` will be called to convert them before
                 training.
-            enable_cloudwatch_metrics (bool): [DEPRECATED] Now there are
-                cloudwatch metrics emitted by all SageMaker training jobs. This
-                will be ignored for now and removed in a further release.
             container_log_level (int): Log level to use within the container
                 (default: logging.INFO). Valid values are defined in the Python
                 logging module.
@@ -1624,12 +1618,6 @@ def __init__(
         self.dependencies = dependencies or []
         self.uploaded_code = None
 
-        if enable_cloudwatch_metrics:
-            warnings.warn(
-                "enable_cloudwatch_metrics is now deprecated and will be removed in the future.",
-                DeprecationWarning,
-            )
-        self.enable_cloudwatch_metrics = False
         self.container_log_level = container_log_level
         self.code_location = code_location
         self.image_uri = image_uri
@@ -1687,7 +1675,6 @@ def _prepare_for_training(self, job_name=None):
         # Modify hyperparameters in-place to point to the right code directory and script URIs
         self._hyperparameters[DIR_PARAM_NAME] = code_dir
         self._hyperparameters[SCRIPT_PARAM_NAME] = script
-        self._hyperparameters[CLOUDWATCH_METRICS_PARAM_NAME] = self.enable_cloudwatch_metrics
         self._hyperparameters[CONTAINER_LOG_LEVEL_PARAM_NAME] = self.container_log_level
         self._hyperparameters[JOB_NAME_PARAM_NAME] = self._current_job_name
         self._hyperparameters[SAGEMAKER_REGION_PARAM_NAME] = self.sagemaker_session.boto_region_name
@@ -1798,9 +1785,6 @@ class constructor
             init_params["hyperparameters"].get(SCRIPT_PARAM_NAME)
         )
         init_params["source_dir"] = json.loads(init_params["hyperparameters"].get(DIR_PARAM_NAME))
-        init_params["enable_cloudwatch_metrics"] = json.loads(
-            init_params["hyperparameters"].get(CLOUDWATCH_METRICS_PARAM_NAME)
-        )
         init_params["container_log_level"] = json.loads(
             init_params["hyperparameters"].get(CONTAINER_LOG_LEVEL_PARAM_NAME)
         )

src/sagemaker/model.py

@@ -599,7 +599,6 @@ def delete_model(self):
 
 SCRIPT_PARAM_NAME = "sagemaker_program"
 DIR_PARAM_NAME = "sagemaker_submit_directory"
-CLOUDWATCH_METRICS_PARAM_NAME = "sagemaker_enable_cloudwatch_metrics"
 CONTAINER_LOG_LEVEL_PARAM_NAME = "sagemaker_container_log_level"
 JOB_NAME_PARAM_NAME = "sagemaker_job_name"
 MODEL_SERVER_WORKERS_PARAM_NAME = "sagemaker_model_server_workers"
@@ -624,7 +623,6 @@ def __init__(
         predictor_cls=None,
         env=None,
         name=None,
-        enable_cloudwatch_metrics=False,
         container_log_level=logging.INFO,
         code_location=None,
         sagemaker_session=None,
@@ -682,9 +680,6 @@ def __init__(
                 when hosted in SageMaker (default: None).
             name (str): The model name. If None, a default model name will be
                 selected on each ``deploy``.
-            enable_cloudwatch_metrics (bool): Whether training and hosting
-                containers will generate CloudWatch metrics under the
-                AWS/SageMakerContainer namespace (default: False).
             container_log_level (int): Log level to use within the container
                 (default: logging.INFO). Valid values are defined in the Python
                 logging module.
@@ -792,7 +787,6 @@ def __init__(
         self.source_dir = source_dir
         self.dependencies = dependencies or []
         self.git_config = git_config
-        self.enable_cloudwatch_metrics = enable_cloudwatch_metrics
         self.container_log_level = container_log_level
         if code_location:
             self.bucket, self.key_prefix = fw_utils.parse_s3_url(code_location)
@@ -890,7 +884,6 @@ def _framework_env_vars(self):
         return {
             SCRIPT_PARAM_NAME.upper(): script_name,
             DIR_PARAM_NAME.upper(): dir_name,
-            CLOUDWATCH_METRICS_PARAM_NAME.upper(): str(self.enable_cloudwatch_metrics).lower(),
             CONTAINER_LOG_LEVEL_PARAM_NAME.upper(): str(self.container_log_level),
             SAGEMAKER_REGION_PARAM_NAME.upper(): self.sagemaker_session.boto_region_name,
         }

src/sagemaker/mxnet/estimator.py

@@ -225,7 +225,6 @@ def create_model(
             framework_version=self.framework_version,
             py_version=self.py_version,
             source_dir=(source_dir or self._model_source_dir()),
-            enable_cloudwatch_metrics=self.enable_cloudwatch_metrics,
             container_log_level=self.container_log_level,
             code_location=self.code_location,
             model_server_workers=model_server_workers,

src/sagemaker/predictor.py

@@ -13,9 +13,7 @@
 """Placeholder docstring"""
 from __future__ import print_function, absolute_import
 
-from sagemaker.deserializers import BaseDeserializer
 from sagemaker.model_monitor import DataCaptureConfig
-from sagemaker.serializers import BaseSerializer
 from sagemaker.session import production_variant, Session
 from sagemaker.utils import name_from_base
 
@@ -64,11 +62,6 @@ def __init__(
             accept (str): The invocation's "Accept", overriding any accept from
                 the deserializer (default: None).
         """
-        if serializer is not None and not isinstance(serializer, BaseSerializer):
-            serializer = LegacySerializer(serializer)
-        if deserializer is not None and not isinstance(deserializer, BaseDeserializer):
-            deserializer = LegacyDeserializer(deserializer)
-
         self.endpoint_name = endpoint_name
         self.sagemaker_session = sagemaker_session or Session()
         self.serializer = serializer
@@ -115,8 +108,6 @@ def _handle_response(self, response):
         """
         response_body = response["Body"]
         if self.deserializer is not None:
-            if not isinstance(self.deserializer, BaseDeserializer):
-                self.deserializer = LegacyDeserializer(self.deserializer)
             # It's the deserializer's responsibility to close the stream
             return self.deserializer.deserialize(response_body, response["ContentType"])
         data = response_body.read()
@@ -149,8 +140,6 @@ def _create_request_args(self, data, initial_args=None, target_model=None, targe
             args["TargetVariant"] = target_variant
 
         if self.serializer is not None:
-            if not isinstance(self.serializer, BaseSerializer):
-                self.serializer = LegacySerializer(self.serializer)
             data = self.serializer.serialize(data)
 
         args["Body"] = data
@@ -403,85 +392,3 @@ def _get_model_names(self):
         )
         production_variants = endpoint_config["ProductionVariants"]
         return [d["ModelName"] for d in production_variants]
-
-
-class LegacySerializer(BaseSerializer):
-    """Wrapper that makes legacy serializers forward compatibile."""
-
-    def __init__(self, serializer):
-        """Initialize a ``LegacySerializer``.
-
-        Args:
-            serializer (callable): A legacy serializer.
-        """
-        self.serializer = serializer
-        self.content_type = getattr(serializer, "content_type", None)
-
-    def __call__(self, *args, **kwargs):
-        """Wraps the call method of the legacy serializer.
-
-        Args:
-            data (object): Data to be serialized.
-
-        Returns:
-            object: Serialized data used for a request.
-        """
-        return self.serializer(*args, **kwargs)
-
-    def serialize(self, data):
-        """Wraps the call method of the legacy serializer.
-
-        Args:
-            data (object): Data to be serialized.
-
-        Returns:
-            object: Serialized data used for a request.
-        """
-        return self.serializer(data)
-
-    @property
-    def CONTENT_TYPE(self):
-        """The MIME type of the data sent to the inference endpoint."""
-        return self.content_type
-
-
-class LegacyDeserializer(BaseDeserializer):
-    """Wrapper that makes legacy deserializers forward compatibile."""
-
-    def __init__(self, deserializer):
-        """Initialize a ``LegacyDeserializer``.
-
-        Args:
-            deserializer (callable): A legacy deserializer.
-        """
-        self.deserializer = deserializer
-        self.accept = getattr(deserializer, "accept", None)
-
-    def __call__(self, *args, **kwargs):
-        """Wraps the call method of the legacy deserializer.
-
-        Args:
-            data (object): Data to be deserialized.
-            content_type (str): The MIME type of the data.
-
-        Returns:
-            object: The data deserialized into an object.
-        """
-        return self.deserializer(*args, **kwargs)
-
-    def deserialize(self, data, content_type):
-        """Wraps the call method of the legacy deserializer.
-
-        Args:
-            data (object): Data to be deserialized.
-            content_type (str): The MIME type of the data.
-
-        Returns:
-            object: The data deserialized into an object.
-        """
-        return self.deserializer(data, content_type)
-
-    @property
-    def ACCEPT(self):
-        """The content type that is expected from the inference endpoint."""
-        return self.accept

src/sagemaker/pytorch/estimator.py

@@ -179,7 +179,6 @@ def create_model(
             framework_version=self.framework_version,
             py_version=self.py_version,
             source_dir=(source_dir or self._model_source_dir()),
-            enable_cloudwatch_metrics=self.enable_cloudwatch_metrics,
             container_log_level=self.container_log_level,
             code_location=self.code_location,
             model_server_workers=model_server_workers,

src/sagemaker/rl/estimator.py

@@ -244,7 +244,6 @@ def create_model(
             source_dir=source_dir,
             code_location=self.code_location,
             dependencies=dependencies,
-            enable_cloudwatch_metrics=self.enable_cloudwatch_metrics,
         )
         extended_args.update(base_args)
 

src/sagemaker/serializers.py

@@ -20,6 +20,13 @@
 
 import numpy as np
 
+from sagemaker.utils import DeferredError
+
+try:
+    import scipy
+except ImportError as e:
+    scipy = DeferredError(e)
+
 
 class BaseSerializer(abc.ABC):
     """Abstract base class for creation of new serializers.
@@ -183,3 +190,25 @@ def serialize(self, data):
             return json.dumps(data.tolist())
 
         return json.dumps(data)
+
+
+class SparseMatrixSerializer(BaseSerializer):
+    """Serialize a sparse matrix to a buffer using the .npz format."""
+
+    CONTENT_TYPE = "application/x-npz"
+
+    def serialize(self, data):
+        """Serialize a sparse matrix to a buffer using the .npz format.
+
+        Sparse matrices can be in the ``csc``, ``csr``, ``bsr``, ``dia`` or
+        ``coo`` formats.
+
+        Args:
+            data (scipy.sparse.spmatrix): The sparse matrix to serialize.
+
+        Returns:
+            io.BytesIO: A buffer containing the serialized sparse matrix.
+        """
+        buffer = io.BytesIO()
+        scipy.sparse.save_npz(buffer, data)
+        return buffer.getvalue()

src/sagemaker/sklearn/estimator.py

@@ -201,7 +201,6 @@ def create_model(
             role,
             entry_point or self._model_entry_point(),
             source_dir=(source_dir or self._model_source_dir()),
-            enable_cloudwatch_metrics=self.enable_cloudwatch_metrics,
             container_log_level=self.container_log_level,
             code_location=self.code_location,
             py_version=self.py_version,

src/sagemaker/workflow/airflow.py

@@ -59,9 +59,6 @@ def prepare_framework(estimator, s3_operations):
         ]
     estimator._hyperparameters[sagemaker.model.DIR_PARAM_NAME] = code_dir
     estimator._hyperparameters[sagemaker.model.SCRIPT_PARAM_NAME] = script
-    estimator._hyperparameters[
-        sagemaker.model.CLOUDWATCH_METRICS_PARAM_NAME
-    ] = estimator.enable_cloudwatch_metrics
     estimator._hyperparameters[
         sagemaker.model.CONTAINER_LOG_LEVEL_PARAM_NAME
     ] = estimator.container_log_level

src/sagemaker/xgboost/estimator.py

@@ -163,7 +163,6 @@ def create_model(
             entry_point or self._model_entry_point(),
             framework_version=self.framework_version,
             source_dir=(source_dir or self._model_source_dir()),
-            enable_cloudwatch_metrics=self.enable_cloudwatch_metrics,
             container_log_level=self.container_log_level,
             code_location=self.code_location,
             py_version=self.py_version,

tests/component/test_mxnet_estimator.py

@@ -75,7 +75,6 @@ def test_deploy(sagemaker_session, tf_version):
         ROLE,
         {
             "Environment": {
-                "SAGEMAKER_ENABLE_CLOUDWATCH_METRICS": "false",
                 "SAGEMAKER_CONTAINER_LOG_LEVEL": "20",
                 "SAGEMAKER_SUBMIT_DIRECTORY": SOURCE_DIR,
                 "SAGEMAKER_REGION": REGION,

tests/component/test_tf_estimator.py

@@ -74,7 +74,6 @@ def test_deploy(sagemaker_session, tf_version):
         ROLE,
         {
             "Environment": {
-                "SAGEMAKER_ENABLE_CLOUDWATCH_METRICS": "false",
                 "SAGEMAKER_CONTAINER_LOG_LEVEL": "20",
                 "SAGEMAKER_SUBMIT_DIRECTORY": SOURCE_DIR,
                 "SAGEMAKER_REQUIREMENTS": "",