Skip to content

Commit d8b28aa

Browse files
committed
fix(mlops): support StepCollection in the pipeline steps compiler
The v3 StepsCompiler dropped the StepCollection handling the v2 compiler had: _simplify_step_list raised ValueError('Invalid input step type') for a StepCollection, and _generate_step_map did not register its sub-steps. As a result Pipeline(steps=[MonitorBatchTransformStep(...)]) (still a StepCollection), depends_on=[collection] and depends_on=['collection name'] all failed to build, although Step.depends_on still documents StepCollection. Restore the v2 behaviour: flatten a StepCollection into its sub-steps and register them in the step map.
1 parent 0495265 commit d8b28aa

2 files changed

Lines changed: 46 additions & 1 deletion

File tree

‎sagemaker-mlops/src/sagemaker/mlops/workflow/_steps_compiler.py‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
from sagemaker.mlops.workflow.function_step import _FunctionStep
2424
from sagemaker.mlops.workflow.steps import Step, StepTypeEnum, PropertyFile
2525
from sagemaker.mlops.workflow.condition_step import ConditionStep
26+
from sagemaker.mlops.workflow.step_collections import StepCollection
2627

2728
from sagemaker.core.workflow.step_outputs import get_step
2829
from sagemaker.core.workflow.pipeline_definition_config import PipelineDefinitionConfig
@@ -211,6 +212,8 @@ def _generate_step_map(steps: Sequence[Union[Step, StepOutput]], step_map: dict)
211212
step_map[step.name] = step
212213
if isinstance(step, ConditionStep):
213214
StepsCompiler._generate_step_map(step.if_steps + step.else_steps, step_map)
215+
if isinstance(step, StepCollection):
216+
StepsCompiler._generate_step_map(step.steps, step_map)
214217

215218
def _simplify_step_list(
216219
self,
@@ -238,7 +241,10 @@ def _simplify_step_list(
238241
if isinstance(step, Step):
239242
if step not in resolved_step_set:
240243
resolved_step_set.add(step)
241-
244+
elif isinstance(step, StepCollection):
245+
for sub_step in step.steps:
246+
if sub_step not in resolved_step_set:
247+
resolved_step_set.add(sub_step)
242248
else:
243249
raise ValueError(f"Invalid input step type: {type(step)}")
244250

‎sagemaker-mlops/tests/unit/workflow/test_steps_compiler.py‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -459,3 +459,42 @@ def test_push_to_build_queue_duplicate(self, mock_session, mock_step):
459459

460460
# Should only be added once to queue
461461
assert len(compiler._build_queue) == 1
462+
463+
464+
class TestStepCollectionSupport:
465+
"""StepCollections (e.g. MonitorBatchTransformStep) are flattened into their sub-steps."""
466+
467+
@staticmethod
468+
def _definition_step_names(steps):
469+
import json
470+
from sagemaker.mlops.workflow.pipeline import Pipeline
471+
472+
session = Mock(boto_region_name="us-west-2", local_mode=False)
473+
definition = json.loads(
474+
Pipeline(name="p", steps=steps, sagemaker_session=session).definition()
475+
)
476+
return {step["Name"]: step for step in definition["Steps"]}
477+
478+
def test_pipeline_with_step_collection(self):
479+
from sagemaker.mlops.workflow.fail_step import FailStep
480+
from sagemaker.mlops.workflow.step_collections import StepCollection
481+
482+
first = FailStep("first")
483+
second = FailStep("second", depends_on=[first])
484+
485+
steps = self._definition_step_names([StepCollection(name="coll", steps=[first, second])])
486+
487+
assert set(steps) == {"first", "second"}
488+
assert steps["second"]["DependsOn"] == ["first"]
489+
490+
@pytest.mark.parametrize("by_name", [False, True])
491+
def test_depends_on_step_collection(self, by_name):
492+
from sagemaker.mlops.workflow.fail_step import FailStep
493+
from sagemaker.mlops.workflow.step_collections import StepCollection
494+
495+
collection = StepCollection(name="coll", steps=[FailStep("first"), FailStep("second")])
496+
after = FailStep("after", depends_on=["coll" if by_name else collection])
497+
498+
steps = self._definition_step_names([collection, after])
499+
500+
assert sorted(steps["after"]["DependsOn"]) == ["first", "second"]

0 commit comments

Comments
 (0)