Control Flow

This doc consists of the following sections:

Condition
Condition Output
Define condition by custom dsl.command_component

By learning and combining them, you would be able to create a runtime If-Else pipeline, which indicate different pipeline inputs will result in different running graphs.

Overview

Below functions in azure.ml.component.dsl package support create runtime condition pipeline.

Name	Decorator	Description
Condition	dsl.condition	If-Else: Execute specific block depends on condition.
Condition Output	dsl.condition_output	Link output to specific input depends on condition.

Condition

Inputs and Parameters

Name	Type	Description
condition	bool, PipelineParameter, PipelineExpression or CustomCondition	The condition to control running flow.
true_block	Instance of Component or Pipeline	The block to be executed if condition runtime evaluation result is True.
false_block	Instance of Component or Pipeline	The block to be executed if condition runtime evaluation result is False.

Sample

from azure.ml.component import dsl

@dsl.pipeline(name="Runtime Condition Pipeline")
def runtime_condition_pipeline(parameter: int):
    # create the true block.
    true_block = component_func_a()
    # create the false block.
    false_block = component_func_b()
    # if parameter > 1, execute the true_block,
    # else, execute the false_block.
    dsl.condition(
        condition=(parameter > 1),
        true_block=true_block,
        false_block=false_block,
    )

Submit the pipeline below with parameter=2, step_a will be executed.

image-dsl-condition-0

Then resubmit the pipeline with parameter=0, step_b will be executed.

image-dsl-condition-1

Condition Output

Condition output allow user select output conditionally.

Inputs and Parameters

Name	Type	Description
condition	bool, PipelineParameter, PipelineExpression or CustomCondition	The condition to control running flow.
input_a	AzureMLDataset or Output	The input to be linked as output if condition runtime evaluation result is True.
input_b	AzureMLDataset or Output	The input to be linked as output if condition runtime evaluation result is False.

Output

Name	Type	Description
output	Output	The output linked to specific input depends on condition.

Sample

from azure.ml.component import dsl

@dsl.pipeline(name="Runtime Condition Pipeline")
def runtime_condition_pipeline(parameter: int):
    # create the step a.
    step_a = component_func_a()
    # create the step b.
    step_b = component_func_b()
    # define the condition as parameter > 1
    condition = (parameter > 1)
    # if condition is true, execute step_a,
    # else, execute step_b.
    dsl.condition(
        condition=condition,
        true_block=step_a,
        false_block=step_b,
    )
    # if condition is true, link output to output of step_a,
    # else, link to output of step_b.
    c = dsl.condition_output(
        condition=condition,
        input_a=step_a.outputs.output_path,
        input_b=step_b.outputs.output_path,
    )
    # pass the condition_output's output to next step.
    post_step = post_processing_func(input=c.outputs.output)
    return post_step.outputs

Submit the pipeline below with parameter=2, step_a will be executed.

image-dsl-condition-output-0

output of condition_output link to input_a(step_a’s output).

image-dsl-condition-output-data-0

Then resubmit the pipeline with parameter=0, step_b will be executed.

image-dsl-condition-output-1

output of condition_output link to input_b(step_b’s output).

image-dsl-condition-output-data-1

Define condition by custom dsl.command_component

Single output condition

Single output condition component can be defined directly with the help of dsl.command_component, and it’s output name will be output.

For example, define the condition component:

from azure.ml.component import dsl
from azure.ml.component.dsl.types import Boolean

@dsl.command_component(name='single_output_condition')
def single_output_condition_func(
        address: str
) -> Boolean(is_control=True):
    # validate the address is https or not
    result = address.startswith('https://')
    return result

Then consume the component in dsl.pipeline:

@dsl.pipeline()
def pipeline_func(address):
    # create the step a and step b.
    step_a = component_func_a()
    step_b = component_func_b()
    # create the condition node
    condition_node = single_output_condition_func(
        address=address
    )
    # pass node's output as condition
    dsl.condition(
        condition=condition_node.outputs.output,
        true_block=step_a,
        false_block=step_b,
    )

Multiple output condition

Multiple output condition component can be defined with the help of dataclass and dsl.command_component.

For example, define the condition component:

from dataclasses import dataclass
from azure.ml.component import dsl, Component
from azure.ml.component.dsl.types import Boolean


@dataclass
class ComponentOutputs(Component):
    output1: Boolean(is_control=True)
    output2: Boolean(is_control=True)


@dsl.command_component(name='multiple_output_condition')
def multiple_output_condition_func(
        address1: str,
        address2: str,
) -> ComponentOutputs:
    # validate the address is https or not
    result1 = address1.startswith('https://')
    result2 = address2.startswith('https://')
    return ComponentOutputs(result1, result2)

Then consume the component in dsl.pipeline:

@dsl.pipeline()
def pipeline_func(address1, address2):
    # create the step a and step b.
    step_a = component_func_a()
    step_b = component_func_b()
    # create the condition node
    condition_node = multiple_output_condition_func(
        address1=address1,
        address2=address2
    )
    # pass node's output1 as condition
    dsl.condition(
        condition=condition_node.outputs.output1,
        true_block=step_a,
        false_block=step_b,
    )

Pass custom condition through pipelines

A pipeline parameter cannot be connected to an output port or pipeline input. Custom condition is usually an output port with primitive type. In order to pass it into a pipeline, a pipeline parameter must have a port annotation with type.

Note that type mismatched error will be raised if wrong annotation is specified on sub pipeline.

For example, the condition parameter on sub_pipeline must have a Input annotation with type specified which indicate it is a primitive type port:

from azure.ml.component import dsl
from azure.ml.component.dsl.types import Boolean, Input

# define the custom condition component
@dsl.command_component(name='single_output_condition')
def single_output_condition_func(
        address: str
) -> Boolean(is_control=True):
    # validate the address is https or not
    result = address.startswith('https://')
    return result

# define the sub pipeline
@dsl.pipeline()
def sub_pipeline_with_condition(
        # annotated condition as a port with type
        condition: Input(type=Boolean.TYPE_NAME)
        # if specified with Boolean and given the condition output, exception will be raised as 
        # "Parameter type mismatched, expected type: 'bool', got 'input'."
        # condition: Boolean
):
    # consume the condition
    dsl.condition(
        condition=condition,
        true_block=...,
        false_block=...,
    )

# define the root pipeline and consume component
@dsl.pipeline()
def root_pipeline(address):
    condition_node = single_output_condition_func(
        address=address
    )
    # pass condition output into sub pipeline
    sub_pipeline_with_condition(condition=condition_node.outputs.output)

Next steps

Learn more about dsl components with our example notebook how-to-create-runtime-condition-pipeline.