We are excited to share the addition of IBM Orchestration Pipelines support to the watsonx.data integration Python SDK. Until now the SDK gave you programmatic control over individual flows. Pipelines give you control over what happens between them: the sequencing, branching, looping, waiting, and error handling that turn a set of jobs into a production workload.

A pipeline is defined as a Kubeflow Pipelines function in your own Python file, and created in your project with a single call. That means the orchestration layer lives in source control alongside everything else, and can be reviewed, diffed, and promoted like any other code.

Creating a pipeline

Decorate a function with @dsl.pipeline, call the tasks you want inside it, and pass the function to Project.create_pipeline().

Code:

>>> from kfp import dsl
>>>
>>> @dsl.component
... def add_two_numbers(a: int, b: int) -> int:
...     print(f"Adding numbers: {a} + {b}")
...     return a + b
>>>
>>> @dsl.pipeline
... def my_pipeline() -> None:
...     add_two_numbers(a=1, b=2)
>>>
>>> pipeline = project.create_pipeline(
...     name='My first pipeline',
...     pipeline_function=my_pipeline,
...     description='A simple addition pipeline'
... )
>>> pipeline
Pipeline(pipeline_id='...', name='My first pipeline')

Retrieving, duplicating, and deleting

Pipelines behave like every other asset collection in the SDK. The Project.pipelines property returns them all, and get() accepts either a name or a pipeline_id.

Code:

>>> project.pipelines
[Pipeline(pipeline_id='...', name='My first pipeline', description='A test pipeline for documentation')]

>>> my_pipeline = project.pipelines.get(name='My first pipeline')
>>> project.pipelines.get(pipeline_id=my_pipeline.pipeline_id)
Pipeline(pipeline_id='...', name='My first pipeline', ...)

Duplication is the fastest way to fork a working pipeline for a new environment or a variant schedule.

Code:

>>> duplicated_pipeline = project.duplicate_pipeline(
...     pipeline,
...     name='My duplicated pipeline',
...     description='A copy of my first pipeline'
... )

Code: project.delete_pipeline(duplicated_pipeline)

Output: <Response [204]>

Running a pipeline

Pipelines run the same way flows do: create a job, then start it. The same Job and JobRun objects you already use for batch and streaming apply here, so scheduling, monitoring, and log retrieval need no new code paths.

Code:

>>> pipeline_job = project.create_job(name='My pipeline job', flow=pipeline)
>>> job_run = pipeline_job.start(name='My pipeline job run', description='First run')
>>> job_run.state
'Running'

Parameter sets can be attached at creation time with the parameter_sets argument, so the same pipeline definition can be pointed at different values per run.

Code:

>>> param_set = project.create_parameter_set(
...     name='pipeline_params',
...     parameters=[
...         {'name': 'input_value', 'type': ParameterType.String, 'value': 'test'},
...         {'name': 'threshold', 'type': ParameterType.Integer, 'value': 100}
...     ]
... )
>>>
>>> pipeline = project.create_pipeline(
...     name='Pipeline with params',
...     pipeline_function=my_pipeline_with_params,
...     parameter_sets=[param_set]
... )

Built-in components

Most orchestration logic does not need to be written from scratch. The platform ships ten built-in components, available through Project.pipeline_components, covering job execution, control flow, and coordination:

The PipelineComponentId enum is the recommended way to reference them, since it removes the risk of a typo in a component identifier.

Code:

>>> from ibm_watsonx_data_integration.services.pipelines.models import PipelineComponentId
>>>
>>> run_datastage_job = project.pipeline_components.get(
...     component_id=PipelineComponentId.RUN_DATASTAGE_JOB
... )
>>> run_datastage_job
PipelineComponent(component_id='run-datastage-job', name='Run DataStage job', description='Run the DataStage job in the project or deployment space', is_built_in=True)

Running a DataStage job from a pipeline, with parameters and environment variables supplied at runtime:

Code:

>>> @dsl.pipeline
... def my_pipeline_with_params() -> None:
...     run_ds_job = project.pipeline_components.get(
...         component_id=PipelineComponentId.RUN_DATASTAGE_JOB
...     )
...     run_ds_job(
...         job=datastage_job,
...         job_parameters={'param1': 'value1', 'param2': 100},
...         env_variables={'ENV_VAR1': 'value1', 'ENV_VAR2': 'value2'}
...     )

Loops take an over collection and, in the parallel case, a parallelism limit, and are used as context managers that yield the current item.

Code:

>>> @dsl.pipeline
... def parallel_loop_pipeline() -> None:
...     loop_component = project.pipeline_components.get(
...         component_id=PipelineComponentId.LOOP_IN_PARALLEL
...     )
...     with loop_component(over=['item1', 'item2', 'item3', 'item4'], parallelism=3) as item:
...         process_item(item=item)

The waiting components handle the parts of a schedule that are hardest to express declaratively. WAIT_FOR_FILE takes a path, a wait_mode of appear or disappear, and a timeout_length, which covers the classic case of a pipeline that cannot start until an upstream system drops a file.

Code:

>>> wait_task = wait_file(
...     file='/path/to/expected/file.txt',
...     wait_mode='appear',
...     timeout_length='00:05:00'
... )
>>> process_task = run_bash(script='cat /path/to/expected/file.txt')
>>> process_task.after(wait_task)

WAIT_FOR_ANY and WAIT_FOR_ALL take no arguments and use .after() to declare which upstream tasks they are gating on, which makes fan-out and fan-in explicit in the pipeline definition.

Beyond the built-ins

Any plain Python function can become a reusable component through Project.create_pipeline_component(), with overwrite=True to publish a new version in place. Pipelines also support user variables, CEL expressions for runtime computation, and dsl.If / dsl.Else branching, and a project-level setting exposes the DataStage optimized pipeline runner through Project.pipeline_settings.runner_type. The components, settings, and examples pages cover each of these in full.

One caveat worth noting: built-in components cannot be deleted, and deleting a custom component that existing pipelines still reference will cause those pipelines to fail. Check usage before removing one.

To get started with our SDK install the ibm-watsonx-data-integration via pip today

pip3 install ibm-watsonx-data-integration

To see more details and view code examples, visit our documentation here.

Try out watsonx.data integration for free today!