Merge branch 'master' into tf-script-mode

Author: icywang86rui

.pylintrc

@@ -84,3 +84,7 @@ dummy-variables-rgx=_|unused_
 # Apply logging string format checks to calls on these modules.
 logging-modules=
     logging
+
+[TYPECHECK]
+ignored-modules=
+    distutils

CHANGELOG.rst

@@ -2,8 +2,13 @@
 CHANGELOG
 =========
 
-1.14.2-dev
-==========
+1.14.3-dev
+=====
+
+* bug-fix: Local Mode: correctly handle the case where the model output folder doesn't exist yet
+
+1.14.2
+=====
 
 * bug-fix: support ``CustomAttributes`` argument in local mode ``invoke_endpoint`` requests
 * enhancement: add ``content_type`` parameter to ``sagemaker.tensorflow.serving.Predictor``
@@ -14,6 +19,10 @@ CHANGELOG
 * build: upgrade docker-compose to 1.23
 * enhancement: Frameworks: update warning for not setting framework_version as we aren't planning a breaking change anymore
 * feature: Estimator: add script mode and Python 3 support for TensorFlow
+* enhancement: Session: remove hardcoded 'training' from job status error message
+* bug-fix: Updated Cloudwatch namespace for metrics in TrainingJobsAnalytics
+* bug-fix: Changes to use correct s3 bucket and time range for dataframes in TrainingJobAnalytics.
+
 
 1.14.1
 ======

src/sagemaker/__init__.py

@@ -36,4 +36,4 @@
 from sagemaker.session import s3_input  # noqa: F401
 from sagemaker.session import get_execution_role  # noqa: F401
 
-__version__ = '1.14.1'
+__version__ = '1.14.2'

src/sagemaker/analytics.py

@@ -246,7 +246,12 @@ def _determine_timeinterval(self):
         """
         description = self._sage_client.describe_training_job(TrainingJobName=self.name)
         start_time = description[u'TrainingStartTime']  # datetime object
-        end_time = description.get(u'TrainingEndTime', datetime.datetime.utcnow())
+        # Incrementing end time by 1 min since CloudWatch drops seconds before finding the logs.
+        # This results in logs being searched in the time range in which the correct log line was not present.
+        # Example - Log time - 2018-10-22 08:25:55
+        #           Here calculated end time would also be 2018-10-22 08:25:55 (without 1 min addition)
+        #           CW will consider end time as 2018-10-22 08:25 and will not be able to search the correct log.
+        end_time = description.get(u'TrainingEndTime', datetime.datetime.utcnow()) + datetime.timedelta(minutes=1)
         return {
             'start_time': start_time,
             'end_time': end_time,

src/sagemaker/job.py

@@ -51,9 +51,9 @@ def wait(self):
         pass
 
     @staticmethod
-    def _load_config(inputs, estimator):
-        input_config = _Job._format_inputs_to_input_config(inputs)
-        role = estimator.sagemaker_session.expand_role(estimator.role)
+    def _load_config(inputs, estimator, expand_role=True, validate_uri=True):
+        input_config = _Job._format_inputs_to_input_config(inputs, validate_uri)
+        role = estimator.sagemaker_session.expand_role(estimator.role) if expand_role else estimator.role
         output_config = _Job._prepare_output_config(estimator.output_path, estimator.output_kms_key)
         resource_config = _Job._prepare_resource_config(estimator.train_instance_count,
                                                         estimator.train_instance_type,
@@ -62,7 +62,8 @@ def _load_config(inputs, estimator):
         stop_condition = _Job._prepare_stop_condition(estimator.train_max_run)
         vpc_config = estimator.get_vpc_config()
 
-        model_channel = _Job._prepare_model_channel(input_config, estimator.model_uri, estimator.model_channel_name)
+        model_channel = _Job._prepare_model_channel(input_config, estimator.model_uri, estimator.model_channel_name,
+                                                    validate_uri)
         if model_channel:
             input_config = [] if input_config is None else input_config
             input_config.append(model_channel)
@@ -75,7 +76,7 @@ def _load_config(inputs, estimator):
                 'vpc_config': vpc_config}
 
     @staticmethod
-    def _format_inputs_to_input_config(inputs):
+    def _format_inputs_to_input_config(inputs, validate_uri=True):
         if inputs is None:
             return None
 
@@ -86,14 +87,14 @@ def _format_inputs_to_input_config(inputs):
 
         input_dict = {}
         if isinstance(inputs, string_types):
-            input_dict['training'] = _Job._format_string_uri_input(inputs)
+            input_dict['training'] = _Job._format_string_uri_input(inputs, validate_uri)
         elif isinstance(inputs, s3_input):
             input_dict['training'] = inputs
         elif isinstance(inputs, file_input):
             input_dict['training'] = inputs
         elif isinstance(inputs, dict):
             for k, v in inputs.items():
-                input_dict[k] = _Job._format_string_uri_input(v)
+                input_dict[k] = _Job._format_string_uri_input(v, validate_uri)
         elif isinstance(inputs, list):
             input_dict = _Job._format_record_set_list_input(inputs)
         else:
@@ -111,15 +112,16 @@ def _convert_input_to_channel(channel_name, channel_s3_input):
         return channel_config
 
     @staticmethod
-    def _format_string_uri_input(uri_input):
-        if isinstance(uri_input, str):
-            if uri_input.startswith('s3://'):
-                return s3_input(uri_input)
-            elif uri_input.startswith('file://'):
-                return file_input(uri_input)
-            else:
-                raise ValueError('Training input data must be a valid S3 or FILE URI: must start with "s3://" or '
-                                 '"file://"')
+    def _format_string_uri_input(uri_input, validate_uri=True):
+        if isinstance(uri_input, str) and validate_uri and uri_input.startswith('s3://'):
+            return s3_input(uri_input)
+        elif isinstance(uri_input, str) and validate_uri and uri_input.startswith('file://'):
+            return file_input(uri_input)
+        elif isinstance(uri_input, str) and validate_uri:
+            raise ValueError('Training input data must be a valid S3 or FILE URI: must start with "s3://" or '
+                             '"file://"')
+        elif isinstance(uri_input, str):
+            return s3_input(uri_input)
         elif isinstance(uri_input, s3_input):
             return uri_input
         elif isinstance(uri_input, file_input):
@@ -128,7 +130,7 @@ def _format_string_uri_input(uri_input):
             raise ValueError('Cannot format input {}. Expecting one of str, s3_input, or file_input'.format(uri_input))
 
     @staticmethod
-    def _prepare_model_channel(input_config, model_uri=None, model_channel_name=None):
+    def _prepare_model_channel(input_config, model_uri=None, model_channel_name=None, validate_uri=True):
         if not model_uri:
             return
         elif not model_channel_name:
@@ -139,22 +141,24 @@ def _prepare_model_channel(input_config, model_uri=None, model_channel_name=None
                 if channel['ChannelName'] == model_channel_name:
                     raise ValueError('Duplicate channels not allowed.')
 
-        model_input = _Job._format_model_uri_input(model_uri)
+        model_input = _Job._format_model_uri_input(model_uri, validate_uri)
         model_channel = _Job._convert_input_to_channel(model_channel_name, model_input)
 
         return model_channel
 
     @staticmethod
-    def _format_model_uri_input(model_uri):
-        if isinstance(model_uri, string_types):
-            if model_uri.startswith('s3://'):
-                return s3_input(model_uri, input_mode='File', distribution='FullyReplicated',
-                                content_type='application/x-sagemaker-model')
-            elif model_uri.startswith('file://'):
-                return file_input(model_uri)
-            else:
-                raise ValueError('Model URI must be a valid S3 or FILE URI: must start with "s3://" or '
-                                 '"file://')
+    def _format_model_uri_input(model_uri, validate_uri=True):
+        if isinstance(model_uri, string_types)and validate_uri and model_uri.startswith('s3://'):
+            return s3_input(model_uri, input_mode='File', distribution='FullyReplicated',
+                            content_type='application/x-sagemaker-model')
+        elif isinstance(model_uri, string_types) and validate_uri and model_uri.startswith('file://'):
+            return file_input(model_uri)
+        elif isinstance(model_uri, string_types) and validate_uri:
+            raise ValueError('Model URI must be a valid S3 or FILE URI: must start with "s3://" or '
+                             '"file://')
+        elif isinstance(model_uri, string_types):
+            return s3_input(model_uri, input_mode='File', distribution='FullyReplicated',
+                            content_type='application/x-sagemaker-model')
         else:
             raise ValueError('Cannot format model URI {}. Expecting str'.format(model_uri))
 

src/sagemaker/local/utils.py

@@ -15,6 +15,7 @@
 import os
 import shutil
 
+from distutils.dir_util import copy_tree
 from six.moves.urllib.parse import urlparse
 
 
@@ -45,6 +46,7 @@ def move_to_destination(source, destination, job_name, sagemaker_session):
     Args:
         source (str): root directory to move
         destination (str): file:// or s3:// URI that source will be moved to.
+        job_name (str): SageMaker job name.
         sagemaker_session (sagemaker.Session): a sagemaker_session to interact with S3 if needed
 
     Returns:
@@ -67,19 +69,12 @@ def move_to_destination(source, destination, job_name, sagemaker_session):
 
 
 def recursive_copy(source, destination):
-    """Similar to shutil.copy but the destination directory can exist. Existing files will be overriden.
+    """A wrapper around distutils.dir_util.copy_tree but won't throw any exception when the source
+    directory does not exist.
+
     Args:
         source (str): source path
         destination (str): destination path
     """
-    for root, dirs, files in os.walk(source):
-        root = os.path.relpath(root, source)
-        current_path = os.path.join(source, root)
-        target_path = os.path.join(destination, root)
-
-        for file in files:
-            shutil.copy(os.path.join(current_path, file), os.path.join(target_path, file))
-        for d in dirs:
-            new_dir = os.path.join(target_path, d)
-            if not os.path.exists(new_dir):
-                os.mkdir(os.path.join(target_path, d))
+    if os.path.isdir(source):
+        copy_tree(source, destination)

src/sagemaker/session.py

@@ -169,11 +169,11 @@ def default_bucket(self):
         if self._default_bucket:
             return self._default_bucket
 
-        s3 = self.boto_session.resource('s3')
         account = self.boto_session.client('sts').get_caller_identity()['Account']
         region = self.boto_session.region_name
         default_bucket = 'sagemaker-{}-{}'.format(region, account)
 
+        s3 = self.boto_session.resource('s3')
         try:
             # 'us-east-1' cannot be specified because it is the default region:
             # https://github.com/boto/boto3/issues/125
@@ -636,7 +636,8 @@ def _check_job_status(self, job, desc, status_key_name):
 
         if status != 'Completed' and status != 'Stopped':
             reason = desc.get('FailureReason', '(No reason provided)')
-            raise ValueError('Error training {}: {} Reason: {}'.format(job, status, reason))
+            job_type = status_key_name.replace('JobStatus', ' job')
+            raise ValueError('Error for {} {}: {} Reason: {}'.format(job_type, job, status, reason))
 
     def wait_for_endpoint(self, endpoint, poll=5):
         """Wait for an Amazon SageMaker endpoint deployment to complete.

src/sagemaker/utils.py

@@ -26,6 +26,9 @@
 import six
 
 
+AIRFLOW_TIME_MACRO = "{{ execution_date.strftime('%Y-%m-%d-%H-%M-%S') }}"
+
+
 # Use the base name of the image as the job name if the user doesn't give us one
 def name_from_image(image):
     """Create a training job name based on the image name and a timestamp.
@@ -58,6 +61,21 @@ def name_from_base(base, max_length=63, short=False):
     return '{}-{}'.format(trimmed_base, timestamp)
 
 
+def airflow_name_from_base(base):
+    """Append airflow execution_date macro (https://airflow.apache.org/code.html?#macros)
+    to the provided string. The macro will beevaluated in Airflow operator runtime.
+    This guarantees that different operators will have same name returned by this function.
+
+    Args:
+        base (str): String used as prefix to generate the unique name.
+
+    Returns:
+        str: Input parameter with appended macro.
+    """
+
+    return "{}-{}".format(base, AIRFLOW_TIME_MACRO)
+
+
 def base_name_from_image(image):
     """Extract the base name of the image to use as the 'algorithm name' for the job.
 

src/sagemaker/workflow/__init__.py

@@ -0,0 +1,12 @@
+# Copyright 2017-2018 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.