Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 20 additions & 8 deletions src/sagemaker/spark/processing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 "
Expand Down
27 changes: 27 additions & 0 deletions tests/unit/sagemaker/spark/test_processing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
[
Expand Down
Loading