diff --git a/src/sagemaker/spark/processing.py b/src/sagemaker/spark/processing.py index 5c6c64852d..2eadef6165 100644 --- a/src/sagemaker/spark/processing.py +++ b/src/sagemaker/spark/processing.py @@ -448,17 +448,29 @@ def _stage_submit_deps(self, submit_deps, input_channel_name): # S3 URIs are included as-is in the spark-submit argument if dep_url.scheme in ["s3", "s3a"]: spark_opt_s3_uris.append(dep_path) - # Local files are copied to temp directory to be uploaded to S3 + # Local files and directories are copied to a temp directory to be + # uploaded to S3 elif not dep_url.scheme or dep_url.scheme == "file": - if not os.path.isfile(dep_path): + if os.path.isdir(dep_path): + logger.info( + "Copying dependency directory from local path %s to tmpdir %s", + dep_path, + tmpdir, + ) + shutil.copytree( + dep_path, + os.path.join(tmpdir, os.path.basename(os.path.normpath(dep_path))), + ) + elif os.path.isfile(dep_path): + logger.info( + "Copying dependency from local path %s to tmpdir %s", dep_path, tmpdir + ) + shutil.copy(dep_path, tmpdir) + else: raise ValueError( - f"submit_deps path {dep_path} is not a valid local file. " - f"{self._submit_deps_error_message}" + f"submit_deps path {dep_path} is not a valid local file " + f"or directory. {self._submit_deps_error_message}" ) - logger.info( - "Copying dependency from local path %s to tmpdir %s", dep_path, tmpdir - ) - shutil.copy(dep_path, tmpdir) else: raise ValueError( f"submit_deps path {dep_path} references unsupported filesystem " diff --git a/tests/unit/sagemaker/spark/test_processing.py b/tests/unit/sagemaker/spark/test_processing.py index c20f16c35c..6c46e875cc 100644 --- a/tests/unit/sagemaker/spark/test_processing.py +++ b/tests/unit/sagemaker/spark/test_processing.py @@ -476,6 +476,33 @@ def test_stage_submit_deps(mock_s3_uploader, jar_file, config, expected, sagemak assert spark_opt == expected[1] +@patch("sagemaker.spark.processing.S3Uploader") +def test_stage_submit_deps_accepts_directory(mock_s3_uploader, sagemaker_session): + """A local directory in submit_deps is staged rather than rejected (issue #2447).""" + spark_processor = PySparkProcessor( + base_job_name="sm-spark", + role="AmazonSageMaker-ExecutionRole", + framework_version="2.4", + instance_count=1, + instance_type="ml.c5.xlarge", + image_uri="790336243319.dkr.ecr.us-west-2.amazonaws.com/sagemaker-spark:0.1", + sagemaker_session=sagemaker_session, + ) + + with tempfile.TemporaryDirectory() as dep_dir: + (pathlib.Path(dep_dir) / "module.py").write_text("x = 1\n") + (pathlib.Path(dep_dir) / "helper.py").write_text("y = 2\n") + + input_channel, spark_opt = spark_processor._stage_submit_deps([dep_dir], "pySparkFiles") + + # The directory was accepted (no ValueError) and staged as an input channel, + # with its contents uploaded to the dependency location. + assert input_channel is not None + assert input_channel.destination == "/opt/ml/processing/input/pySparkFiles" + assert spark_opt == "/opt/ml/processing/input/pySparkFiles" + mock_s3_uploader.upload.assert_called_once() + + @pytest.mark.parametrize( "config, expected", [