Fan-out, convergence, and terminal fan-out

This example has two stages of parallel execution:

• Start fans out to Load Orders and Load Customers.
• Converge is reached from both load steps.
• Converge fans out to Publish API and Publish File.
• The two publish steps have no outgoing transitions, so they end the workflow.

The complete runnable example is available here:

"""Create a two-stage parallel workflow using only Sleep steps.

The first fan-out runs two independent loads.  Converge waits for both loads,
then fans out to two terminal publishing steps.
"""

from getpass import getpass

from datatransforms.workflow import Sleep, Workflow
from datatransforms.workbench import DataTransformsWorkbench, WorkbenchConfig


def build_two_stage_parallel_workflow(
    name="Parallel Sleep Two Stage",
    project="Your Project",
):
    """Build the workflow without making a REST call."""
    start = Sleep("Start", delay=1000, firstStep=True)
    load_orders = Sleep("Load Orders", delay=8000)
    load_customers = Sleep("Load Customers", delay=6000)
    converge = Sleep("Converge", delay=1000)
    publish_api = Sleep("Publish API", delay=3000)
    publish_file = Sleep("Publish File", delay=3000)

    # Repeated OK transitions create the first fan-out.
    start.ok(load_orders)
    start.ok(load_customers)

    # Both branches point to Converge, creating the convergence.
    load_orders.ok(converge)
    load_customers.ok(converge)

    # Repeated OK transitions create the second fan-out. Both branches end.
    converge.ok(publish_api)
    converge.ok(publish_file)

    return Workflow.parallel(name, project=project).add_execution_steps(
        [
            start,
            load_orders,
            load_customers,
            converge,
            publish_api,
            publish_file,
        ]
    )


def create_two_stage_parallel_workflow(
    password,
    name="Parallel Sleep Two Stage",
    project="Your Project",
):
    """Connect to Data Transforms and create or update the workflow."""
    connect_params = WorkbenchConfig.get_workbench_config(password)
    workbench = DataTransformsWorkbench()
    workbench.connect_workbench(connect_params)

    workflow = build_two_stage_parallel_workflow(name=name, project=project)
    workflow.create()
    return workflow


if __name__ == "__main__":
    create_two_stage_parallel_workflow("<your-workbench-password>")

Using helper methods

The first version expresses fan-out and convergence with the convenience methods fan_out and depends_on:

fan_out(*next_steps) declares multiple success transitions from the current step in one call. depends_on(*predecessor_steps) declares that the current step waits for all of the listed predecessor steps to succeed. These helpers simplify the equivalent repeated ok calls, so developers do not need to write every transition explicitly:

start.fan_out(load_orders, load_customers)
publish.depends_on(load_orders, load_customers)

The fan-out above is equivalent to start.ok(load_orders) followed by start.ok(load_customers). The dependency above is equivalent to adding load_orders.ok(publish) and load_customers.ok(publish).

"""Create a parallel workflow using only Sleep steps.

The workflow starts one step, runs two independent branches concurrently, and
starts the final step only after both branches have completed successfully.
"""

from getpass import getpass

from datatransforms.workflow import Sleep, Workflow
from datatransforms.workbench import DataTransformsWorkbench, WorkbenchConfig


def build_parallel_sleep_workflow(
    name="Parallel Sleep Fanout Join",
    project="Your Project",
):
    """Build the workflow without making a REST call."""
    start = Sleep("Start", delay=1000, firstStep=True)
    load_orders = Sleep("Load Orders", delay=10000)
    load_customers = Sleep("Load Customers", delay=7000)
    publish = Sleep("Publish", delay=1000)

    # Fan-out: both branches become eligible after Start succeeds.
    start.fan_out(load_orders, load_customers)

    # Join: Publish waits for both branches to succeed.
    publish.depends_on(load_orders, load_customers)

    return Workflow.parallel(name, project=project).add_execution_steps(
        [start, load_orders, load_customers, publish]
    )


def create_parallel_sleep_workflow(
    password,
    name="Parallel Sleep Fanout Join",
    project="Your Project",
):
    """Connect to Data Transforms and create or update the workflow."""
    connect_params = WorkbenchConfig.get_workbench_config(password)
    workbench = DataTransformsWorkbench()
    workbench.connect_workbench(connect_params)

    workflow = build_parallel_sleep_workflow(name=name, project=project)
    workflow.create()
    return workflow


if __name__ == "__main__":
    create_parallel_sleep_workflow("<your-workbench-password>")

Using only ok transitions

The same workflow can be authored with repeated ok calls. Repeated calls append successor names, so they create a fan-out. Multiple predecessors pointing to the same successor create the convergence in a Parallel workflow.

start.ok(load_orders)
start.ok(load_customers)

load_orders.ok(converge)
load_customers.ok(converge)

converge.ok(publish_api)
converge.ok(publish_file)

The complete ok-only example is available here:

"""Create a two-stage parallel workflow using only Sleep steps.

The first fan-out runs two independent loads.  Converge waits for both loads,
then fans out to two terminal publishing steps.
"""

from getpass import getpass

from datatransforms.workflow import Sleep, Workflow
from datatransforms.workbench import DataTransformsWorkbench, WorkbenchConfig


def build_two_stage_parallel_workflow(
    name="Parallel Sleep Two Stage",
    project="Your Project",
):
    """Build the workflow without making a REST call."""
    start = Sleep("Start", delay=1000, firstStep=True)
    load_orders = Sleep("Load Orders", delay=8000)
    load_customers = Sleep("Load Customers", delay=6000)
    converge = Sleep("Converge", delay=1000)
    publish_api = Sleep("Publish API", delay=3000)
    publish_file = Sleep("Publish File", delay=3000)

    # Repeated OK transitions create the first fan-out.
    start.ok(load_orders)
    start.ok(load_customers)

    # Both branches point to Converge, creating the convergence.
    load_orders.ok(converge)
    load_customers.ok(converge)

    # Repeated OK transitions create the second fan-out. Both branches end.
    converge.ok(publish_api)
    converge.ok(publish_file)

    return Workflow.parallel(name, project=project).add_execution_steps(
        [
            start,
            load_orders,
            load_customers,
            converge,
            publish_api,
            publish_file,
        ]
    )


def create_two_stage_parallel_workflow(
    password,
    name="Parallel Sleep Two Stage",
    project="Your Project",
):
    """Connect to Data Transforms and create or update the workflow."""
    connect_params = WorkbenchConfig.get_workbench_config(password)
    workbench = DataTransformsWorkbench()
    workbench.connect_workbench(connect_params)

    workflow = build_two_stage_parallel_workflow(name=name, project=project)
    workflow.create()
    return workflow


if __name__ == "__main__":
    create_two_stage_parallel_workflow("<your-workbench-password>")