fix: use unique names for test training jobs (#738)

Author: jesterhazy

tests/integ/test_byo_estimator.py

@@ -54,6 +54,7 @@ def test_byo_estimator(sagemaker_session, region):
     """
     image_name = registry(region) + "/factorization-machines:1"
     training_data_path = os.path.join(DATA_DIR, 'dummy_tensor')
+    job_name = unique_name_from_base('byo')
 
     with timeout(minutes=TRAINING_DEFAULT_TIMEOUT_MINUTES):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
@@ -71,21 +72,19 @@ def test_byo_estimator(sagemaker_session, region):
         estimator = Estimator(image_name=image_name,
                               role='SageMakerRole', train_instance_count=1,
                               train_instance_type='ml.c4.xlarge',
-                              sagemaker_session=sagemaker_session, base_job_name='test-byo')
+                              sagemaker_session=sagemaker_session)
 
         estimator.set_hyperparameters(num_factors=10,
                                       feature_dim=784,
                                       mini_batch_size=100,
                                       predictor_type='binary_classifier')
 
         # training labels must be 'float32'
-        estimator.fit({'train': s3_train_data})
+        estimator.fit({'train': s3_train_data}, job_name=job_name)
 
-    endpoint_name = unique_name_from_base('byo')
-
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
         model = estimator.create_model()
-        predictor = model.deploy(1, 'ml.m4.xlarge', endpoint_name=endpoint_name)
+        predictor = model.deploy(1, 'ml.m4.xlarge', endpoint_name=job_name)
         predictor.serializer = fm_serializer
         predictor.content_type = 'application/json'
         predictor.deserializer = sagemaker.predictor.json_deserializer
@@ -101,7 +100,7 @@ def test_async_byo_estimator(sagemaker_session, region):
     image_name = registry(region) + "/factorization-machines:1"
     endpoint_name = unique_name_from_base('byo')
     training_data_path = os.path.join(DATA_DIR, 'dummy_tensor')
-    training_job_name = ""
+    job_name = unique_name_from_base('byo')
 
     with timeout(minutes=5):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
@@ -119,19 +118,19 @@ def test_async_byo_estimator(sagemaker_session, region):
         estimator = Estimator(image_name=image_name,
                               role='SageMakerRole', train_instance_count=1,
                               train_instance_type='ml.c4.xlarge',
-                              sagemaker_session=sagemaker_session, base_job_name='test-byo')
+                              sagemaker_session=sagemaker_session)
 
         estimator.set_hyperparameters(num_factors=10,
                                       feature_dim=784,
                                       mini_batch_size=100,
                                       predictor_type='binary_classifier')
 
         # training labels must be 'float32'
-        estimator.fit({'train': s3_train_data}, wait=False)
-        training_job_name = estimator.latest_training_job.name
+        estimator.fit({'train': s3_train_data}, wait=False, job_name=job_name)
 
     with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        estimator = Estimator.attach(training_job_name=training_job_name, sagemaker_session=sagemaker_session)
+        estimator = Estimator.attach(training_job_name=job_name,
+                                     sagemaker_session=sagemaker_session)
         model = estimator.create_model()
         predictor = model.deploy(1, 'ml.m4.xlarge', endpoint_name=endpoint_name)
         predictor.serializer = fm_serializer

tests/integ/test_factorization_machines.py

@@ -25,6 +25,8 @@
 
 
 def test_factorization_machines(sagemaker_session):
+    job_name = unique_name_from_base('fm')
+
     with timeout(minutes=TRAINING_DEFAULT_TIMEOUT_MINUTES):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
         pickle_args = {} if sys.version_info.major == 2 else {'encoding': 'latin1'}
@@ -37,15 +39,16 @@ def test_factorization_machines(sagemaker_session):
                                    train_instance_type='ml.c4.xlarge',
                                    num_factors=10, predictor_type='regressor',
                                    epochs=2, clip_gradient=1e2, eps=0.001, rescale_grad=1.0 / 100,
-                                   sagemaker_session=sagemaker_session, base_job_name='test-fm')
+                                   sagemaker_session=sagemaker_session)
 
         # training labels must be 'float32'
-        fm.fit(fm.record_set(train_set[0][:200], train_set[1][:200].astype('float32')))
+        fm.fit(fm.record_set(train_set[0][:200], train_set[1][:200].astype('float32')),
+               job_name=job_name)
 
-    endpoint_name = unique_name_from_base('fm')
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        model = FactorizationMachinesModel(fm.model_data, role='SageMakerRole', sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
+        model = FactorizationMachinesModel(fm.model_data, role='SageMakerRole',
+                                           sagemaker_session=sagemaker_session)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         result = predictor.predict(train_set[0][:10])
 
         assert len(result) == 10
@@ -54,8 +57,7 @@ def test_factorization_machines(sagemaker_session):
 
 
 def test_async_factorization_machines(sagemaker_session):
-    training_job_name = ""
-    endpoint_name = unique_name_from_base('factorizationMachines')
+    job_name = unique_name_from_base('fm')
 
     with timeout(minutes=5):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
@@ -69,22 +71,23 @@ def test_async_factorization_machines(sagemaker_session):
                                    train_instance_type='ml.c4.xlarge',
                                    num_factors=10, predictor_type='regressor',
                                    epochs=2, clip_gradient=1e2, eps=0.001, rescale_grad=1.0 / 100,
-                                   sagemaker_session=sagemaker_session, base_job_name='test-fm')
+                                   sagemaker_session=sagemaker_session)
 
         # training labels must be 'float32'
-        fm.fit(fm.record_set(train_set[0][:200], train_set[1][:200].astype('float32')), wait=False)
-        training_job_name = fm.latest_training_job.name
+        fm.fit(fm.record_set(train_set[0][:200], train_set[1][:200].astype('float32')),
+               job_name=job_name,
+               wait=False)
 
         print("Detached from training job. Will re-attach in 20 seconds")
         time.sleep(20)
         print("attaching now...")
 
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        estimator = FactorizationMachines.attach(training_job_name=training_job_name,
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
+        estimator = FactorizationMachines.attach(training_job_name=job_name,
                                                  sagemaker_session=sagemaker_session)
         model = FactorizationMachinesModel(estimator.model_data, role='SageMakerRole',
                                            sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         result = predictor.predict(train_set[0][:10])
 
         assert len(result) == 10

tests/integ/test_horovod.py

@@ -20,6 +20,7 @@
 import boto3
 import pytest
 
+import sagemaker.utils
 import tests.integ as integ
 from sagemaker.tensorflow import TensorFlow
 from tests.integ import timeout
@@ -30,7 +31,7 @@
 @pytest.mark.canary_quick
 @pytest.mark.parametrize('instance_type', ['ml.c5.xlarge', 'ml.p3.2xlarge'])
 def test_horovod(sagemaker_session, instance_type, tmpdir):
-
+    job_name = sagemaker.utils.unique_name_from_base('tf-horovod')
     estimator = TensorFlow(entry_point=os.path.join(horovod_dir, 'test_hvd_basic.py'),
                            role='SageMakerRole',
                            train_instance_count=2,
@@ -39,11 +40,10 @@ def test_horovod(sagemaker_session, instance_type, tmpdir):
                            py_version=integ.PYTHON_VERSION,
                            script_mode=True,
                            framework_version='1.12',
-                           distributions={'mpi': {'enabled': True}},
-                           base_job_name='test-tf-horovod')
+                           distributions={'mpi': {'enabled': True}})
 
     with timeout.timeout(minutes=integ.TRAINING_DEFAULT_TIMEOUT_MINUTES):
-        estimator.fit()
+        estimator.fit(job_name=job_name)
 
         tmp = str(tmpdir)
         extract_files_from_s3(estimator.model_data, tmp)
@@ -59,7 +59,7 @@ def test_horovod(sagemaker_session, instance_type, tmpdir):
     (2, 2)])
 def test_horovod_local_mode(sagemaker_local_session, instances, processes, tmpdir):
     output_path = 'file://%s' % tmpdir
-
+    job_name = sagemaker.utils.unique_name_from_base('tf-horovod')
     estimator = TensorFlow(entry_point=os.path.join(horovod_dir, 'test_hvd_basic.py'),
                            role='SageMakerRole',
                            train_instance_count=2,
@@ -70,11 +70,10 @@ def test_horovod_local_mode(sagemaker_local_session, instances, processes, tmpdi
                            output_path=output_path,
                            framework_version='1.12',
                            distributions={'mpi': {'enabled': True,
-                                                  'processes_per_host': processes}},
-                           base_job_name='test-tf-horovod')
+                                                  'processes_per_host': processes}})
 
     with timeout.timeout(minutes=integ.TRAINING_DEFAULT_TIMEOUT_MINUTES):
-        estimator.fit()
+        estimator.fit(job_name=job_name)
 
         tmp = str(tmpdir)
         extract_files(output_path.replace('file://', ''), tmp)

tests/integ/test_ipinsights.py

@@ -25,30 +25,32 @@
 
 
 def test_ipinsights(sagemaker_session):
+    job_name = unique_name_from_base('ipinsights')
+
     with timeout(minutes=TRAINING_DEFAULT_TIMEOUT_MINUTES):
         data_path = os.path.join(DATA_DIR, 'ipinsights')
         data_filename = 'train.csv'
 
         with open(os.path.join(data_path, data_filename), 'rb') as f:
             num_records = len(f.readlines())
 
-        ipinsights = IPInsights(
-            role='SageMakerRole',
-            train_instance_count=1,
-            train_instance_type='ml.c4.xlarge',
-            num_entity_vectors=10,
-            vector_dim=100,
-            sagemaker_session=sagemaker_session,
-            base_job_name='test-ipinsights')
+            ipinsights = IPInsights(
+                role='SageMakerRole',
+                train_instance_count=1,
+                train_instance_type='ml.c4.xlarge',
+                num_entity_vectors=10,
+                vector_dim=100,
+                sagemaker_session=sagemaker_session)
 
         record_set = prepare_record_set_from_local_files(data_path, ipinsights.data_location,
-                                                         num_records, FEATURE_DIM, sagemaker_session)
-        ipinsights.fit(record_set, None)
-
-    endpoint_name = unique_name_from_base('ipinsights')
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        model = IPInsightsModel(ipinsights.model_data, role='SageMakerRole', sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+                                                         num_records, FEATURE_DIM,
+                                                         sagemaker_session)
+        ipinsights.fit(records=record_set, job_name=job_name)
+
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
+        model = IPInsightsModel(ipinsights.model_data, role='SageMakerRole',
+                                sagemaker_session=sagemaker_session)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         assert isinstance(predictor, RealTimePredictor)
 
         predict_input = [['user_1', '1.1.1.1']]

tests/integ/test_kmeans.py

@@ -27,6 +27,7 @@
 
 
 def test_kmeans(sagemaker_session):
+    job_name = unique_name_from_base('kmeans')
     with timeout(minutes=TRAINING_DEFAULT_TIMEOUT_MINUTES):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
         pickle_args = {} if sys.version_info.major == 2 else {'encoding': 'latin1'}
@@ -37,7 +38,7 @@ def test_kmeans(sagemaker_session):
 
         kmeans = KMeans(role='SageMakerRole', train_instance_count=1,
                         train_instance_type='ml.c4.xlarge',
-                        k=10, sagemaker_session=sagemaker_session, base_job_name='test-kmeans')
+                        k=10, sagemaker_session=sagemaker_session)
 
         kmeans.init_method = 'random'
         kmeans.max_iterations = 1
@@ -61,12 +62,12 @@ def test_kmeans(sagemaker_session):
             force_dense='True',
         )
 
-        kmeans.fit(kmeans.record_set(train_set[0][:100]))
+        kmeans.fit(kmeans.record_set(train_set[0][:100]), job_name=job_name)
 
-    endpoint_name = unique_name_from_base('kmeans')
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        model = KMeansModel(kmeans.model_data, role='SageMakerRole', sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
+        model = KMeansModel(kmeans.model_data, role='SageMakerRole',
+                            sagemaker_session=sagemaker_session)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         result = predictor.predict(train_set[0][:10])
 
         assert len(result) == 10
@@ -81,8 +82,7 @@ def test_kmeans(sagemaker_session):
 
 
 def test_async_kmeans(sagemaker_session):
-    training_job_name = ""
-    endpoint_name = unique_name_from_base('kmeans')
+    job_name = unique_name_from_base('kmeans')
 
     with timeout(minutes=5):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
@@ -94,7 +94,7 @@ def test_async_kmeans(sagemaker_session):
 
         kmeans = KMeans(role='SageMakerRole', train_instance_count=1,
                         train_instance_type='ml.c4.xlarge',
-                        k=10, sagemaker_session=sagemaker_session, base_job_name='test-kmeans')
+                        k=10, sagemaker_session=sagemaker_session)
 
         kmeans.init_method = 'random'
         kmeans.max_iterations = 1
@@ -118,17 +118,17 @@ def test_async_kmeans(sagemaker_session):
             force_dense='True',
         )
 
-        kmeans.fit(kmeans.record_set(train_set[0][:100]), wait=False)
-        training_job_name = kmeans.latest_training_job.name
+        kmeans.fit(kmeans.record_set(train_set[0][:100]), wait=False, job_name=job_name)
 
         print("Detached from training job. Will re-attach in 20 seconds")
         time.sleep(20)
         print("attaching now...")
 
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        estimator = KMeans.attach(training_job_name=training_job_name, sagemaker_session=sagemaker_session)
-        model = KMeansModel(estimator.model_data, role='SageMakerRole', sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
+        estimator = KMeans.attach(training_job_name=job_name, sagemaker_session=sagemaker_session)
+        model = KMeansModel(estimator.model_data, role='SageMakerRole',
+                            sagemaker_session=sagemaker_session)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         result = predictor.predict(train_set[0][:10])
 
         assert len(result) == 10

tests/integ/test_knn.py

@@ -25,6 +25,8 @@
 
 
 def test_knn_regressor(sagemaker_session):
+    job_name = unique_name_from_base('knn')
+
     with timeout(minutes=TRAINING_DEFAULT_TIMEOUT_MINUTES):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
         pickle_args = {} if sys.version_info.major == 2 else {'encoding': 'latin1'}
@@ -36,15 +38,15 @@ def test_knn_regressor(sagemaker_session):
         knn = KNN(role='SageMakerRole', train_instance_count=1,
                   train_instance_type='ml.c4.xlarge',
                   k=10, predictor_type='regressor', sample_size=500,
-                  sagemaker_session=sagemaker_session, base_job_name='test-knn-rr')
+                  sagemaker_session=sagemaker_session)
 
         # training labels must be 'float32'
-        knn.fit(knn.record_set(train_set[0][:200], train_set[1][:200].astype('float32')))
+        knn.fit(knn.record_set(train_set[0][:200], train_set[1][:200].astype('float32')),
+                job_name=job_name)
 
-    endpoint_name = unique_name_from_base('knn')
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
         model = KNNModel(knn.model_data, role='SageMakerRole', sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         result = predictor.predict(train_set[0][:10])
 
         assert len(result) == 10
@@ -53,8 +55,7 @@ def test_knn_regressor(sagemaker_session):
 
 
 def test_async_knn_classifier(sagemaker_session):
-    training_job_name = ""
-    endpoint_name = unique_name_from_base('knn')
+    job_name = unique_name_from_base('knn')
 
     with timeout(minutes=5):
         data_path = os.path.join(DATA_DIR, 'one_p_mnist', 'mnist.pkl.gz')
@@ -68,22 +69,22 @@ def test_async_knn_classifier(sagemaker_session):
                   train_instance_count=1, train_instance_type='ml.c4.xlarge',
                   k=10, predictor_type='classifier', sample_size=500,
                   index_type='faiss.IVFFlat', index_metric='L2',
-                  sagemaker_session=sagemaker_session, base_job_name='test-knn-cl')
+                  sagemaker_session=sagemaker_session)
 
         # training labels must be 'float32'
-        knn.fit(knn.record_set(train_set[0][:200], train_set[1][:200].astype('float32')), wait=False)
-        training_job_name = knn.latest_training_job.name
+        knn.fit(knn.record_set(train_set[0][:200], train_set[1][:200].astype('float32')),
+                wait=False, job_name=job_name)
 
         print("Detached from training job. Will re-attach in 20 seconds")
         time.sleep(20)
         print("attaching now...")
 
-    with timeout_and_delete_endpoint_by_name(endpoint_name, sagemaker_session):
-        estimator = KNN.attach(training_job_name=training_job_name,
+    with timeout_and_delete_endpoint_by_name(job_name, sagemaker_session):
+        estimator = KNN.attach(training_job_name=job_name,
                                sagemaker_session=sagemaker_session)
         model = KNNModel(estimator.model_data, role='SageMakerRole',
                          sagemaker_session=sagemaker_session)
-        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=endpoint_name)
+        predictor = model.deploy(1, 'ml.c4.xlarge', endpoint_name=job_name)
         result = predictor.predict(train_set[0][:10])
 
         assert len(result) == 10