ProcessingStep in SageMaker Pipelines - allow customisation of the S3 destination path

[tomaszdudek7]

Hi,

I want to fully customize the S3 destination path of my ProcessingStep. For example, I'd like to save its results to s3://bucket/<pipeline_execution_id>/preprocessing/data.csv.

However, it seems like you can not customize it that way. The ProcessingOutput and its destination parameter seems to be explicitly IGNORED, as can be seen in the source code.

Question: Why is that the case? Are there any workarounds?

---

APPENDIX 1:

Such behavior is fully possible when using Transformer and TransformerStep.

If you use sagemaker.workflow.functions.Join and sagemaker.workflow.execution_variables.ExecutionVariables, you can create desired path as follows:

path_that_i_want = Join(on='/', values=["s3://bucket", ExecutionVariables. PIPELINE_EXECUTION_ID, "transformer"])

and then set it in output_path of the Transformer:

from sagemaker.transformer import Transformer
from sagemaker.workflow.steps import TransformStep

trf = Transformer(
    #other params omitted
    output_path=path_that_i_want
)

trf_step = TransformStep(
    #other params omitted
    transformer=trf
)

If I then add trf_step to my Pipeline, its result will be saved to s3://bucket/<execution_id>/transformer.

---

APPENDIX 2:

Such behavior is partially possible with the Estimator.

Similar to appendix 1, if you specify the output_path inside the Estimator as follows:

path_that_i_want = Join(on='/', values=["s3://bucket", ExecutionVariables. PIPELINE_EXECUTION_ID, "estimator"])

and then add it to the Estimator, your models will be saved to s3://bucket/<execution_id>/estimator/pipelines-<execution_id>-<step_name>-<some_weird_id>/.

This partially does the trick, but still adds an unnecessary prefix in S3 in a form of pipelines-<execution_id>-<step_name>-<some_weird_id>/.

---

APPENDIX 3:

(not to make X-Y question).

The reason I ask is because I don't like the current pathing of SM Pipelines. The S3 artifacts are grouped by steps and then by executions. I'd like to group it by execution and then by steps.

Having this:

• s3://bucket/<execution_id>/models
• s3://bucket/<execution_id>/preprocessing
• s3://bucket/<execution_id>/postprocessing
  would be way better than having this:
• s3://bucket/models/<execution_id>
• s3://bucket/preprocessing/<execution_id>
• s3://bucket/postprocessing/<execution_id>

If something goes wrong with a given execution, you'd be able to quickly browse all artifacts of that execution just by opening one subdirectory, namely <execution_id>. The current state requires you to juggle between many subdirectories.

[andriusdc]

Same issue here !

[cbsouza]

I've been dealing with the same problem.

[shashankprasanna]

Hi @tomaszdudek7
I'm not able to reproduce your issue, and I'm able to provide custom paths in the format you describe. Here's the snippet of code showing ProcessingOutput for your reference

from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.execution_variables import ExecutionVariables
from sagemaker.workflow.functions import Join

sklearn_framework_version = "0.23-1"

sklearn_processor = SKLearnProcessor(
    framework_version=sklearn_framework_version,
    instance_type="ml.m5.large",
    instance_count=processing_instance_count,
    base_job_name="sklearn-housing-data-process",
    role=role,
    sagemaker_session=pipeline_session,
)

s3_output_prefix = "custom-destination"
s3_output_path = f"s3://{bucket}/{s3_output_prefix}"
destination_train=Join(on="/",
                 values=[f"s3://{bucket}","dataset",
                         ExecutionVariables.PIPELINE_EXECUTION_ID,
                         "train"])
destination_test=Join(on="/",
                 values=[f"s3://{bucket}","dataset",
                         ExecutionVariables.PIPELINE_EXECUTION_ID,
                         "test"])
destination_scalar_model=Join(on="/",
                 values=[f"s3://{bucket}","models",
                         ExecutionVariables.PIPELINE_EXECUTION_ID,
                         "scaler_model"])

processor_args = sklearn_processor.run(
    inputs=[
        ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="scaler_model", source="/opt/ml/processing/scaler_model",destination=destination_scalar_model,s3_upload_mode="EndOfJob"),
        ProcessingOutput(output_name="train", source="/opt/ml/processing/train",destination=destination_train,s3_upload_mode="EndOfJob"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test",destination=destination_test,s3_upload_mode="EndOfJob"),
    ],
    code="code/preprocess.py",
)

[tomaszdudek7]

I've just checked that and I can confirm that this indeed is the case. You can customize output path from a ProcessingJob ran by SageMaker Pipelines.

Sorry for a misleading ticket, must have been a bug in my code somewhere. Thanks @shashankprasanna !

[andriusdc]

@shashankprasanna , if you use the processor_args to initiate a ProcessingStep, does it still works ? I am asking this because from my experience with it, it seems that the interation of ProcessingStep with ProcessingOutputs is the problem, as the documentation from the latter explicit says that the destination arg is ignored when using with ProcessingStep

[butakun]

@andriusdc

this is just to confirm, as of version 2.184.0, it does work even when invoked from a ProcessingStep, contrary to the documentation for ProcessingOutput.

output_path = Join(
    values=[
        f"s3://{sm_bucket}/pipelines/",
        ExecutionVariables.PIPELINE_NAME,
        "/processing-awsgt-manifest-",
        ExecutionVariables.PIPELINE_EXECUTION_ID
    ]
)

output_awsgt_manifest = ProcessingOutput(
    output_name="groundtruth_manifest",
    source="/opt/ml/processing/output",
    destination=output_path,
    s3_upload_mode="Continuous"
)

processor_groundtruth_manifest = ScriptProcessor(
    command=["python3"],
    image_uri="foobar/latest",
    role=sm_role,
    instance_count=1,
    instance_type="ml.t3.medium",
)

step_awsgt_manifest = ProcessingStep(
    name="GroundTruthManifest",
    processor = processor_groundtruth_manifest,
    inputs=[
        input_detected,
    ],
    outputs=[
        output_awsgt_manifest,
    ],
    code="pipeline_scripts/processing_awsgt_manifest.py",
)