Flows

Flows#

A flow is an object used to store the execution flow of a data pipeline. It is composed of multiple stages, with each stage defining how data is handled at that part of the execution flow.

The SDK provides the following functionality to interact with streaming flows:

  • Creating a flow

  • Retrieving flows

  • Editing a flow

  • Updating a flow

  • Duplicating a flow

  • Deleting a flow

  • Exporting flows

  • Importing flows

  • Validating a flow

  • Previewing a flow

  • Handling error records

Prerequisites#

To create a flow using the SDK, make sure you have completed the following steps:

Note

Currently, the SDK only supports creating a flow with an engine installed.

Creating a Flow#

In the UI, you can create a flow by navigating to Assets -> New asset -> Create a flow.

Warning

Creating a streaming flow in the UI requires setting an environment.

Screenshot of the flow creation page.

In the SDK, you can create a flow from a Project object using the Project.create_flow() method. You are required to supply the name and environment parameters. To learn how to create an environment, see creating an environment. You can also provide an optional description. You do not need to specify flow_type because its default value is streaming. This method returns a StreamingFlow instance.

>>> new_flow = project.create_flow(name='My streaming flow', description='optional description', environment=environment)
>>> new_flow
StreamingFlow(name='My streaming flow', description='optional description', flow_id=..., engine_version=...)

Note

Stage names and availability may differ depending on the engine type (DataCollector or Jetstream) used by your environment. Ensure you reference the correct stage names for your specific engine type when building flows.

Retrieving Flows#

Flows can be retrieved through a Project object using the Project.flows property. If you want to retrieve only streaming flows, use the Project.flows.get_all() method to filter by flow_type. You can also retrieve a single flow using the Project.flows.get() method, which takes unique identifiers such as flow_id or name.

>>> project.flows  # a list of all flows
[StreamingFlow(...), StreamingFlow(...)...]
>>> project.flows.get_all(flow_type='streaming')
[...StreamingFlow(name='My streaming flow', description='optional description', ...)...]
>>> project.flows.get(name='My streaming flow')
StreamingFlow(name='My streaming flow', description='optional description', ...)

Editing a Flow#

You can edit a flow in multiple ways.

For starters, you can edit a flow’s attributes like name or description.

>>> new_flow.description = 'new description for the flow'
>>> new_flow
StreamingFlow(name='My first flow', description='new description for the flow', ...)

You can also edit any flow by editing its stages. This can include adding a stage, removing a stage, updating a stage’s configuration, or connecting a stage in a different way than before. All of these operations are covered in the Stages documentation (see Batch Stages or Streaming Stages).

In addition, streaming flows have many properties related to pipelines, engines, and more. You can edit a streaming flow’s configuration through the Flow.configuration property. This property returns a Configuration object that encapsulates a flow’s configuration. You can print the configuration and edit it similarly to a dict.

>>> new_flow.configuration['retry_pipeline_on_error']
True
>>> new_flow.configuration['retry_pipeline_on_error'] = False

Arranging Stages#

In the UI, stages in a flow can be manually positioned by dragging them around the canvas. However, when creating flows programmatically or when a flow becomes cluttered, you may want to automatically arrange the stages in a clean, organized layout.

By default, flows are automatically arranged when updated. This behavior can be controlled using the StreamingFlow.use_auto_arrange() method.

Automatic Arrangement (Default Behavior)#

When you create a new flow, automatic arrangement is enabled by default. This means stages will be automatically positioned based on their stages and links whenever the flow is updated:

>>> # Create a flow - auto-arrange is enabled by default
>>> streaming_flow_auto = project.create_flow(name='My auto arranged flow', environment=environment)
>>> # ... add stages and connections ...
>>>
>>> # When you update, stages are automatically arranged
>>> project.update_flow(streaming_flow_auto)
<Response [200]>

Disabling Automatic Arrangement#

If you want to preserve the exact positioning of stages in an existing flow, you can disable automatic arrangement. This is useful when you’ve manually positioned stages in the UI and want to maintain that layout when making programmatic changes:

>>> # Get an existing flow and disable auto-arrange
>>> streaming_flow_auto = project.flows.get(name='My auto arranged flow')
>>> streaming_flow_auto.use_auto_arrange(False)
StreamingFlow(...)
>>> # ... add stages and connections ...
>>>
>>> # When you update, the original stage positions are preserved
>>> project.update_flow(streaming_flow_auto)
<Response [200]>

Note

When auto-arrange is disabled, newly added stages may not be positioned optimally relative to existing stages. You may need to manually position them in the UI or call StreamingFlow.auto_arrange() once to establish a clean layout.

Manual Arrangement#

You can also manually trigger arrangement at any time using the StreamingFlow.auto_arrange() method:

>>> # Manually arrange stages at any time
>>> streaming_flow_auto.auto_arrange()
StreamingFlow(name='My auto arranged flow', ...)
>>>
>>> # Update the flow with the new layout
>>> project.update_flow(streaming_flow_auto)
<Response [200]>

This is particularly useful when:

  • Creating flows programmatically where manual positioning is not practical

  • Cleaning up flows that have become cluttered over time

  • Standardizing the layout of multiple similar flows

  • Importing flows that may have inconsistent positioning

Note

The actual layout calculation happens when the flow is updated or visualized. The auto_arrange() method removes existing position data to trigger the automatic arrangement, while use_auto_arrange() controls whether this happens automatically on every update.

Supported Flow Architectures#

Streaming flows support several common connection patterns. You can combine these patterns within the same flow depending on your processing requirements:

Linear Flows#

One stage connects to one downstream stage in sequence:

>>> linear_flow = project.create_flow(name='Linear flow', description='optional description', environment=environment)
>>> source = linear_flow.add_stage('Dev Raw Data Source')
>>> processor = linear_flow.add_stage('Field Renamer')
>>> destination = linear_flow.add_stage('Trash')
>>>
>>> # Linear connection: source → processor → destination
>>> source.connect_output_to(processor)
FieldRenamer_01(name='Field Renamer 1')
>>> processor.connect_output_to(destination)
Trash_01(name='Trash 1')

This creates a topology like:

Source ─→ Processor ─→ Destination

Fan-out Flows#

One stage connects to multiple downstream stages:

>>> fan_out_flow = project.create_flow(name='Fan-out flow', description='optional description', environment=environment)
>>> source = fan_out_flow.add_stage('Dev Raw Data Source')
>>> dest1 = fan_out_flow.add_stage('Trash')
>>> dest2 = fan_out_flow.add_stage('Trash')
>>> dest3 = fan_out_flow.add_stage('Trash')
>>>
>>> # Fan-out: source connects to multiple destinations
>>> source.connect_output_to(dest1, dest2, dest3)
[Trash_01(name='Trash 1'), Trash_02(name='Trash 2'), Trash_03(name='Trash 3')]

This creates a topology like:

Source
  ├─→ Destination 1
  ├─→ Destination 2
  └─→ Destination 3

Conditional Routing Flows#

Stages with multiple outputs can route records to different downstream stages based on predicates:

>>> conditional_flow = project.create_flow(name='Conditional Routing Flow', description='optional description', environment=environment)
>>> source = conditional_flow.add_stage('Dev Raw Data Source')
>>> stream_selector = conditional_flow.add_stage('Stream Selector')
>>> high_value = conditional_flow.add_stage('Trash')
>>> low_value = conditional_flow.add_stage('Trash')
>>> default_dest = conditional_flow.add_stage('Trash')
>>>
>>> # Add predicates for conditional routing
>>> stream_selector.add_predicates([
...     '${record:value(\'/amount\') > 1000}',
...     '${record:value(\'/amount\') < 100}'
... ])
>>>
>>> # Connect source to stream selector
>>> source.connect_output_to(stream_selector)
StreamSelector_01(name='Stream Selector 1')
>>>
>>> # Route to different destinations based on predicates
>>> stream_selector.connect_output_to(high_value, predicate=stream_selector.predicates[0])
Trash_01(name='Trash 1')
>>> stream_selector.connect_output_to(low_value, predicate=stream_selector.predicates[1])
Trash_02(name='Trash 2')
>>> stream_selector.connect_output_to(default_dest, predicate=stream_selector.predicates[2])
Trash_03(name='Trash 3')

This creates a topology like:

Source
  │
  ▼
Stream Selector
  ├─→ High Value Destination (amount > 1000)
  ├─→ Low Value Destination (amount < 100)
  └─→ Default Destination (all others)

Converging Flows (Fan-In)#

Multiple upstream stages connect into a downstream stage:

>>> converging_flow = project.create_flow(name='Converging Flow', description='optional description', environment=environment)
>>> source1 = converging_flow.add_stage('Dev Raw Data Source')
>>> source2 = converging_flow.add_stage('Dev Raw Data Source')
>>> source3 = converging_flow.add_stage('Dev Raw Data Source')
>>> destination = converging_flow.add_stage('Trash')
>>>
>>> # Converging: multiple sources connect to one destination
>>> source1.connect_output_to(destination)
Trash_01(name='Trash 1')
>>> source2.connect_output_to(destination)
Trash_01(name='Trash 1')
>>> source3.connect_output_to(destination)
Trash_01(name='Trash 1')

This creates a topology like:

Source 1 ─┐
          │
Source 2 ─┼─→ Destination
          │
Source 3 ─┘

Diamond Flows#

Data splits into multiple branches that later converge back into a single stage, allowing parallel processing paths.

>>> diamond_flow = project.create_flow(name='Diamond Flow', description='optional description', environment=environment)
>>> source = diamond_flow.add_stage('Dev Raw Data Source')
>>>
>>> # Split into two branches
>>> branch1_processor = diamond_flow.add_stage('Field Renamer')
>>> branch2_processor = diamond_flow.add_stage('Field Masker')
>>>
>>> # Converge back to single destination
>>> destination = diamond_flow.add_stage('Trash')
>>>
>>> # Create diamond topology
>>> source.connect_output_to(branch1_processor, branch2_processor)
[FieldRenamer_01(name='Field Renamer 1'), FieldMasker_01(name='Field Masker 1')]
>>> branch1_processor.connect_output_to(destination)
Trash_01(name='Trash 1')
>>> branch2_processor.connect_output_to(destination)
Trash_01(name='Trash 1')

This creates a topology like:

Source
  ├─→ Branch 1 Processor ─┐
  │                       ├─→ Destination
  └─→ Branch 2 Processor ─┘

Multistage Branching Flows#

Complex flows with multiple levels of branching and processing, where different branches can have their own processing pipelines.

>>> multistage_flow = project.create_flow(name='Multistage Flow', description='optional description', environment=environment)
>>> source = multistage_flow.add_stage('Dev Raw Data Source')
>>>
>>> # First level branching
>>> stream_selector = multistage_flow.add_stage('Stream Selector')
>>> stream_selector.add_predicates(['${record:value(\'/type\') == \'A\'}', '${record:value(\'/type\') == \'B\'}'])
>>>
>>> # Branch A: Multiple processing stages
>>> branch_a_processor1 = multistage_flow.add_stage('Field Renamer')
>>> branch_a_processor2 = multistage_flow.add_stage('Field Masker')
>>> branch_a_dest = multistage_flow.add_stage('Trash')
>>>
>>> # Branch B: Different processing pipeline
>>> branch_b_processor1 = multistage_flow.add_stage('Expression Evaluator')
>>> branch_b_processor2 = multistage_flow.add_stage('Field Remover')
>>> branch_b_dest = multistage_flow.add_stage('Trash')
>>>
>>> # Connect source to stream selector
>>> source.connect_output_to(stream_selector)
StreamSelector_01(name='Stream Selector 1')
>>>
>>> # Connect Branch A (type == 'A')
>>> stream_selector.connect_output_to(branch_a_processor1, predicate=stream_selector.predicates[0])
FieldRenamer_01(name='Field Renamer 1')
>>> branch_a_processor1.connect_output_to(branch_a_processor2)
FieldMasker_01(name='Field Masker 1')
>>> branch_a_processor2.connect_output_to(branch_a_dest)
Trash_01(name='Trash 1')
>>>
>>> # Connect Branch B (default)
>>> stream_selector.connect_output_to(branch_b_processor1, predicate=stream_selector.predicates[1])
ExpressionEvaluator_01(name='Expression Evaluator 1')
>>> branch_b_processor1.connect_output_to(branch_b_processor2)
FieldRemover_01(name='Field Remover 1')
>>> branch_b_processor2.connect_output_to(branch_b_dest)
Trash_02(name='Trash 2')

This creates a topology like:

Source
  │
  ▼
Stream Selector
  ├─→ Branch A Processor 1 ─→ Branch A Processor 2 ─→ Branch A Destination
  │
  └─→ Branch B Processor 1 ─→ Branch B Processor 2 ─→ Branch B Destination

Event-Driven Side Flows#

A stage’s event output can be connected separately from its data output:

>>> event_flow = project.create_flow(name='Event Flow', description='optional description', environment=environment)
>>> source = event_flow.add_stage('Dev Raw Data Source')
>>> destination = event_flow.add_stage('Trash')
>>> executor = event_flow.add_stage('Pipeline Finisher Executor')
>>>
>>> # Connect data output
>>> source.connect_output_to(destination)
Trash_01(name='Trash 1')
>>>
>>> # Connect event output separately
>>> source.connect_event_to(executor)
PipelineFinisherExecutor_01(name='Pipeline Finisher Executor 1')

This creates a topology like:

Source
  ├─→ Destination (data output)
  └─→ Executor (event output)

These architectures are built using the same stage connection APIs described in Stages. For complex topologies with multiple branches, you can also use StreamingFlow.auto_arrange() to improve the visual layout after making changes.

Updating a Flow#

In the UI, you can update a flow by making changes to it and clicking the Save icon.

Screenshot of updating the flow.

In the SDK, you can make changes to a Flow instance in memory and update it by passing that object to the Project.update_flow() method.

This method returns an HTTP response indicating the status of the update operation.

>>> new_flow.name = 'new flow name'  # you can also update the stages, configuration, etc.
>>> project.update_flow(new_flow)
<Response [200]>
>>> new_flow
StreamingFlow(name='new flow name', description='new description for the flow', ...)

Duplicating a Flow#

To duplicate a flow using the SDK, pass a Flow instance to the Project.duplicate_flow() method, along with the name parameter for the new flow and an optional description parameter.

This duplicates the flow and returns a new instance of Flow.

>>> duplicated_flow = project.duplicate_flow(new_flow, name='duplicated flow', description='duplicated flow description')
>>> duplicated_flow
StreamsetsFlow(name='duplicated flow', description='duplicated flow description', ...)

Deleting a Flow#

To delete a flow in the UI, go to Assets, choose a flow, click the three dots next to it, and select Delete.

Screenshot of deleting a flow.

To delete a flow using the SDK, pass a Flow instance to the Project.delete_flow() method.

This method returns an HTTP response indicating the status of the delete operation.

>>> project.delete_flow(duplicated_flow)
<Response [204]>

Validating a Flow#

In the UI, you can validate a streaming flow by making changes to the flow and clicking the Validate icon.

Screenshot of the validate button.

To validate a streaming flow via the SDK, first update the flow and then call the StreamingFlow.validate() method. This returns a ValidationResult object. The issues attribute contains a list of validation errors. There are two types:

  • FlowValidationError — stage and pipeline issues reported by the SDC engine (e.g. missing input stream, disconnected stage).

  • FlowstoreIssue — connection-validation issues reported by the Flowstore API (e.g. a connection ID referenced in the flow is not found in CAMS or is not defined for the project).

>>> new_flow.add_stage('Trash')
Trash_01(name='Trash 1')
>>> project.update_flow(new_flow)
<Response [200]>
>>> new_flow.validate()
ValidationResult(success=False, issues=[FlowValidationError(type='stageIssues', instance_name='Trash_01', human_readable_message='The first stage must be an origin'), FlowValidationError(type='stageIssues', instance_name='Trash_01', human_readable_message='Target must have input streams')], message='Validation Failed')

Previewing a flow#

In the UI, you can preview a flow by clicking the Preview icon.

Screenshot of the preview button.

To preview a flow via the SDK, call the StreamingFlow.preview() method. This will return a list of PreviewStage instances. Each PreviewStage provides access to its input and output properties, which contain the input and output data for that stage.

>>> preview = flow.preview()
>>> preview
[PreviewStage(instance_name='DevRawDataSource_01'), PreviewStage(instance_name='Trash_01')]
>>> dev_raw_data_preview, trash_preview = preview
>>> dev_raw_data_preview.input
>>> dev_raw_data_preview.output
[('abc', 'xyz', 'lmn')]

Exporting Flows#

To export a flow using the SDK, call the Project.export_flow() method and pass an individual Flow object. If you want to export multiple flows at the same time, call the Project.export_flows() method and pass a list of Flow objects.

Note

When using Project.export_flows() to export multiple flows at once, all objects in the list must be of the same type. You cannot pass a list containing both StreamingFlow and BatchFlow objects. It must contain one type or the other.

You can set the following additional parameters:

  • with_plain_text_credentials – export credentials in plain text. (only relevant to streaming flows)

  • destination – specify the export location.

  • stream – stream the ZIP file data.

The function returns the location where the exported ZIP file was written.

>>> flow
StreamingFlow(name='My streaming flow', description='optional description', ...)

>>> project.export_flow(flow=flow)
PosixPath('flows.zip')

Importing Flows#

To import a flow via the SDK, call the Project.import_flows() method and specify the source parameter, which is the path to the ZIP file containing the JSON file or files for the streaming flow or flows to be imported, along with the flow_type parameter set to 'streaming'.

You must also set the conflict_resolution parameter, which determines how to handle an attempted import of a duplicate flow that already exists in the project. The options for conflict_resolution are listed below:

  • 'skip' – skip this particular flow and move to the next flow to be imported.

  • 'replace' - replace the existing flow with the flow to be imported.

  • 'rename' - keep the existing flow and rename the flow that is being imported.

You can optionally import a flow’s associated jobs and connections alongside the flow itself using the import_dependencies parameter. Pass a list of StreamingImportDependency values to control which dependency types are included:

  • StreamingImportDependency.JOB — re-creates jobs that were associated with the exported flow.

  • StreamingImportDependency.CONNECTION — re-creates connections that were referenced by the exported flow.

The function returns either an imported StreamingFlow or a list of imported StreamingFlow objects.

>>> from ibm_watsonx_data_integration.services.streamsets.models.flow_model import StreamingImportDependency
>>>
>>> project.import_flows(
...     flow_type='streaming',
...     source='flows_to_import.zip',
...     conflict_resolution='replace',
...     import_dependencies=[StreamingImportDependency.JOB, StreamingImportDependency.CONNECTION],
... )
StreamingFlow(name='dummy_flow', description='dummy_description', flow_id='...', engine_version='...')

Handling Error Records#

To edit error record handling in the UI, click the gear icon in the top-right corner of the screen on a flow’s edit page.

Screenshot of opening a flow settings.

This opens a new pop-up window with a tab for Error records on the left. This will let you adjust the error record handling for the flow.

Screenshot of flow settings.

This page lets you change how error records are handled by policy and which stage should handle them.

The possible options for error record policy are Original record as it was generated by the origin and Record as it was seen by the stage that sent it to error stream. In the SDK, these equate to ORIGINAL_RECORD and STAGE_RECORD.

This can be updated in a flow’s configuration.

>>> new_flow.configuration['error_record_policy']
'ORIGINAL_RECORD'
>>> new_flow.configuration['error_record_policy'] = 'STAGE_RECORD'

To change the error record stage, you can call StreamingFlow.set_error_stage() method. You need to pass either the label or the name of the new error stage. You can also optionally pass the new stage’s library.

Note

All error stages other than Discard will have configuration options for you to customize your experience.

>>> write_to_file = new_flow.set_error_stage('Write to File')
>>> write_to_file.configuration['directory'] = '/path/to/some/directory'

You can view the current error stage for a flow at any point using the StreamingFlow.error_stage property.

>>> new_flow.error_stage
WritetoFile_ErrorStage(name='Error Records - Write to File')

Using Parameter Sets#

Parameter sets allow you to make your streaming flows more flexible by defining reusable parameters that can be referenced throughout your flow.

In the UI, you can use a Parameter Set in a streaming flow by navigating to Flow parameters -> Parameter sets -> Add parameter sets and choosing the desired parameter set from the list.

In the SDK, to use a Parameter Set instance in a streaming flow, you can pass it to the StreamingFlow.use_parameter_set() method. This method establishes a relationship between the flow and the parameter set, and automatically adds the parameter set’s parameters as constants in the flow’s configuration.

>>> # Retrieve the parameter set
>>> paramset = project.parameter_sets.get(name='my_streaming_params')
>>>
>>> # Use the parameter set in the streaming flow
>>> streaming_flow_with_parameters.use_parameter_set(paramset)
StreamingFlow(name='flow with parameters', ...)
>>>
>>> # Update the flow to save the changes
>>> project.update_flow(streaming_flow_with_parameters)
<Response [200]>

To use a parameter from a parameter set in your streaming flow stages, reference it using the notation: ${parameter_set_name__parameter_name} (note the double underscore __ separator).

>>> # Example: Using parameters in stage configurations
>>> dev_raw_data = streaming_flow_with_parameters.add_stage('Dev Raw Data Source')
>>> dev_raw_data.data_format = '${my_streaming_params__data_format}'
>>> dev_raw_data.number_of_threads = '${my_streaming_params__number_of_threads}'
>>> dev_raw_data.stop_after_first_batch = '${my_streaming_params__stop_after_first_batch}'

You can retrieve all parameter sets associated with a streaming flow using the StreamingFlow.parameter_sets property.

>>> streaming_flow_with_parameters.parameter_sets
[ParameterSet(name='my_streaming_params', parameters=[...], description='', value_sets=[])]

Note

Streaming flows currently support only ParameterType.String type parameters.

Streaming flows do not support:

  • Local parameters

  • PROJDEF parameter sets

  • Value sets for parameter sets

For more information about parameter sets, see Parameter Sets.

Section Contents