fix(spark): allow directory paths in submit_deps (#6368)

PySparkProcessor/SparkJarProcessor rejected a local directory passed via
submit_py_files / submit_jars / submit_files: _stage_submit_deps required
os.path.isfile(dep_path) and raised "is not a valid local file" for a
directory. Users could only list individual files.

Accept a directory as well: copy its tree into the staging tempdir (via
shutil.copytree) so its contents are uploaded to the dependency input
channel and mounted into the Spark container, alongside the existing
single-file handling. Update the error message to mention directories.

Add a unit test that a directory is staged (not rejected).

Relates-to #2447

Co-authored-by: Roja Reddy Sareddy <rsareddy@amazon.com>

Author: rsareddy0329

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 "

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",
     [