We are excited to share the release of watsonx.data integration SDK version 2.0.0. This release introduces Unstructured Data Integration flows, a new flow paradigm that brings documents, chunking, and embeddings into the SDK alongside batch and streaming. It also delivers XML and hierarchical data processing, custom stage extensibility, parallel execution control, and a new cloud native JetStream (Flink based) engine.

Unstructured Data Integration Flows

Until now, the SDK modeled two kinds of work: batch flows for scheduled, partitioned movement of structured records, and streaming flows for continuous pipelines. Both assume rows and columns.

UDI is a third paradigm, and it assumes documents. A UDI flow ingests PDFs, contracts, invoices, and reports, extracts their text, scores their quality, splits them into chunks, redacts sensitive content, generates embeddings, and writes vectors to a database. It is the path from a folder of unstructured files to a working retrieval index, expressed in Python.

The structural difference matters as much as the use case. Batch and streaming flows are built from stages connected by links, where each link carries a schema. UDI flows use a declarative pipeline of operators executed by a Python orchestrator. You do not wire links between operators or define column schemas. You append operators in sequence, and data flows through them.

Creating a flow and adding operators

Create a UDI flow by passing flow_type='udi' to Project.create_flow(), then append operators with add_operator().

Code:

udi_flow = project.create_flow(name='My UDI flow', flow_type='udi')

ingest_op = udi_flow.add_operator('ingest_cpd_assets', data_assets=asset_ids)
extract_op = udi_flow.add_operator('extract_cpd', ocr_mode='disabled')
quality_op = udi_flow.add_operator('doc_quality')
chunk_op = udi_flow.add_operator('chunker', chunk_size=1000, chunk_overlap=200)
embed_op = udi_flow.add_operator('embeddings')
store_op = udi_flow.add_operator('milvusdb_cp4d', connection=milvus_conn, collection_name='documents')

project.update_flow(udi_flow)

Output: <Response [200]>

The built-in operators cover the full document pipeline:

• ingest_cpd_assets — ingest assets from Cloud Pak for Data
• extract_cpd — extract text and structure from those assets
• doc_quality — score document quality and emit features such as docq_total_words
• chunker — split documents into chunks for embedding
• pii_and_hap_extract_redact — detect and redact PII and hateful, abusive, or profane content
• embeddings — generate embeddings for text
• milvusdb_cp4d — write vectors to a Milvus vector database
• branching and merge — conditional routing and recombination

Note: To add operators that require data assets, such as ingest_cpd_assets, you first need to upload assets to your project. Adding operators changes the flow in memory only — you must call Project.update_flow() to save.

Discovering operators instead of memorizing them

Rather than hardcoding operator names and parameters, the operators_metadata property returns each operator type along with its required and optional parameters, types, descriptions, and defaults.

Code:

all_metadata = udi_flow.operators_metadata.get_all()
ingest_metadata = udi_flow.operators_metadata.get(operator_type='ingest_cpd_assets')

Output:

OperatorMetadata(operator_type='ingest_cpd_assets', category=..., attributes=..., features=...)

Conditional branching and merging

Document pipelines rarely treat every input the same way. A scanned two-page form and a hundred-page contract need different chunking. UDI flows support conditional branching on feature values emitted by upstream operators, then merging the branches back into a single stream.

Code:

quality_branch = udi_flow.add_operator('branching', name='Quality-Based Processing')

high_quality = quality_branch.add_branch(
    condition='docq_total_words >= 100',
    label='High Quality',
    merge_link_name='high_quality_link'
)
high_quality.add_operator('chunker', chunk_size=1000, chunk_overlap=200, chunk_type='watsonx')
high_quality.add_operator('embeddings', name='High Quality Embeddings')

low_quality = quality_branch.add_branch(
    condition='docq_total_words < 100',
    label='Low Quality',
    merge_link_name='low_quality_link'
)
low_quality.add_operator('doc_quality', name='Re-check Quality')
low_quality.add_operator('chunker', chunk_size=500, chunk_overlap=100)
low_quality.add_operator('embeddings', name='Low Quality Embeddings')

merge_op = udi_flow.add_operator('merge', merge_type='rows')
project.update_flow(udi_flow)

Output: <Response [200]>

Branches are evaluated in the order they are added, and each branch can hold its own sequence of operators. The merge_link_name parameter identifies which branch data came from once the streams recombine. Branches can also nest via add_branching(), so you can route on document quality first and content type second.

Code:

content_type_branch = high_quality.add_branching(name='Content Type Branch')
expense_branch = content_type_branch.add_branch(condition="type == 'expense'")
expense_branch.add_operator('embeddings')

Custom operators

When the built-in operators do not cover a transformation, upload your own Python operator as a project-level resource and use it in any flow like a built-in.

Code:

custom_op = project.upload_custom_operator(
    name='my_custom_operator',
    file_path='/path/to/operator.py',
    dependency='/path/to/dependencies.tar.gz'   # optional
)

udi_flow.add_operator('my_custom_operator', param1='value1')

Output:

<BaseOperator(type='my_custom_operator', ...)>

Custom operators are listed through Project.custom_operators and removed with Project.delete_custom_operator().

Running flows and promoting to production

UDI flows execute as jobs using the same pattern as batch and streaming, and they support parameter sets, local parameters, and per-run overrides identically.

Code:

job = project.create_job(flow=udi_flow, name='My UDI Job')
job_run = job.start()

logs = udi_flow.get_execution_logs(run_id=job_run.job_run_id)

get_execution_logs() returns a dictionary of execution detail for the run, covering entries such as:

[INFO] Starting flow execution...
[INFO] Processing operator: ingest_cpd_assets
[INFO] Flow execution completed successfully

For production use, promote a flow from a project to a deployment space.

Code:

promotion_response = udi_flow.promote_to_space(
    target_space_id='12345-abcde-67890',
    flow_name='Production Flow',
    description='Flow promoted to production',
    duplicate_action='REPLACE'    # or 'IGNORE' to skip if exists
)

Output: 'promoted-flow-id-123'

Note: Space-based jobs are managed separately from project-based jobs. Use the space's job management APIs directly rather than mixing project and space contexts.

XML Schema Libraries (Batch)

A SchemaLibrary is a project level asset that holds one or more XSD or JSON schema files defining XML document structures. A library is created once and can be referenced by multiple Hierarchical Data stages across many flows within the same project.

Creating a schema library and uploading a schema

Code:

from pathlib import Path

schema_lib = project.create_schema_library(
    name='plant_collection',
    description='XSD for plant collection XML documents'
)

response = schema_lib.upload_schema_file(file=Path('plant_collection.xsd'))

Supported extensions are .xsd, .json, .jsn, and .jsd. The file is automatically wrapped in a zip archive before being sent to the platform, and after upload the library's parse_status field reflects whether the schema was parsed successfully.

Attaching a library to a stage

Code:

flow = project.create_flow(name='xml_processing_flow', flow_type='batch')
stage = flow.add_stage('Hierarchical Data', 'hierarchical_data')
stage.add_schema_library(schema_lib)

project.update_flow(flow)

Output: <Response [200]>

Duplicate entries are silently ignored, and remove_schema_library() detaches a library. You can list every library in a project through the Project.schema_libraries property.

Code: project.schema_libraries

Output:

[SchemaLibrary(library_id='...', name='plant_collection')]

Note: Schema libraries are project level assets, independent of any flow. Deleting a library does not automatically remove references to it from existing Hierarchical Data stages.

Hierarchical Data Stages (Batch)

The HierarchicalDataStage processes XML and JSON documents by running an internal pipeline of processing steps called an assembly. Think of the assembly as a mini flow that lives inside the stage: it has its own Input and Output steps, and you connect processing steps between them just as you connect stages inside a batch flow.

The assembly

Accessing the assembly property for the first time creates a fresh assembly with an InputStep and an OutputStep already connected to each other. The simplest assembly passes data straight through, so you only need to call propagate() pointing at the Input step.

Code:

flow = project.create_flow(name='xml_flow', flow_type='batch')
row_gen = flow.add_stage('Row Generator', 'row_gen')
hd_stage = flow.add_stage('Hierarchical Data', 'hd')
peek = flow.add_stage('Peek', 'peek')

link_1 = row_gen.connect_output_to(hd_stage)
link_1.name = 'Link_1'
link_1_schema = link_1.create_schema()
link_1_schema.add_field('VARCHAR', 'name', length=100)
link_1_schema.add_field('INTEGER', 'age')

link_2 = hd_stage.connect_output_to(peek)
link_2.name = 'Link_2'

# Input and Output steps are created automatically
assembly = hd_stage.assembly
assembly.output_step.propagate(assembly.input_step)

project.update_flow(flow)

Output: <Response [201]>

Adding steps to the assembly

Use Assembly.add_step() to insert a processing step. Because the Input and Output steps are connected by default, you disconnect the default link, wire the new step between them, then propagate columns from the new step.

Code:

sort_step = assembly.add_step('Sort', 'Sort_1')
sort_step.configuration.list_to_sort = 'top/InputLinks/Link_1'
sort_step.configuration.add_key('age', 'DESC')

# 1. Remove the default Input to Output connection
assembly.input_step.disconnect_output_from(assembly.output_step)

# 2. Wire: Input to Sort to Output
link_3 = assembly.input_step.connect_output_to(sort_step)
link_3.name = 'Link_3'
link_4 = sort_step.connect_output_to(assembly.output_step)
link_4.name = 'Link_4'

# 3. Propagate columns from the sort step to the output
assembly.output_step.propagate(source=sort_step)

project.update_flow(flow)

Output: <Response [201]>

Twelve step types are available through add_step(): Sort, Aggregate, HJoin, HPivot, VPivot, Regroup, Switch, OrderJoin, JSONParser, JSONComposer, XMLParser, and XMLComposer.

Note: finalize_assembly() serializes the in-memory assembly into the compressed stage parameters the platform stores. It is called automatically by Project.update_flow(), so the only reason to call it manually is to inspect the serialized parameters before saving. XML and JSON Parser and Composer steps require a schema library to be attached to the stage.

Wrapped Stages (Batch)

A wrapped stage is a reusable project asset that encapsulates an external executable command into a DataStage stage. Once created and configured, it appears in the batch flow editor just like any built-in stage, letting you invoke custom scripts or binaries as part of a flow.

Wrapped stage management follows a two step process: define the asset in the project, then use it in a flow.

Defining the asset

Code:

from ibm_watsonx_data_integration.services.datastage.models.components.wrapped_stage import (
    DataType, FileDescriptor
)

wrapped_stage = project.create_wrapped_stage(
    name='My Custom Stage',
    description='Executes a custom post-processing script',
)
wrapped_stage.command = '/opt/scripts/my_script.sh'
wrapped_stage.execution_mode = 'seq'          # 'par' (default) or 'seq'

# Stream-based links pipe data through standard file descriptors
wrapped_stage.add_input('input1', FileDescriptor.STDIN)
wrapped_stage.add_output('output1', FileDescriptor.STDOUT)

# Non-stream links pass a file path as a command-line argument or env var
wrapped_stage.add_output('output2', 'OUTPUT_FILE', is_command_line=True)

wrapped_stage.add_property(
    name='mode',
    data_type=DataType.LIST,
    prompt='Processing mode',
    list_values=['fast', 'normal', 'safe'],
)
wrapped_stage.add_environment_variable('LOG_LEVEL', 'INFO')
wrapped_stage.add_failure_code('255')

project.update_wrapped_stage(wrapped_stage)

Output: <Response [200]>

Properties accept the DataType values STRING, INTEGER, FLOAT, PATHNAME, LIST, INPUTCOLUMN, and BOOLEAN.

Generating and using the stage

Before a wrapped stage can be used in a batch flow it must be generated. Generation compiles the saved configuration into a usable DataStage stage type that the flow editor recognises. This mirrors the separate Save and Generate buttons in the UI.

Code:

project.generate_wrapped_stage(wrapped_stage)

stage_node = batch_flow.add_stage(
    type='Wrapped Stage',
    label='Pre-processing Step',
    wrapped_stage_name='My Custom Stage',
)
project.update_flow(batch_flow)

Output: <Response [201]>

Note: Wrapped stages are an on-premises feature and require the platform to be configured for an on-premises deployment. Calling any wrapped stage method from a SaaS or AWS deployment raises a RuntimeError. You must also generate a wrapped stage at least once before adding it to a flow, and re-generate after any configuration change so the flow editor picks up the latest definition.

Link Partitioning (Batch)

Link partitioning controls how data is distributed across parallel processing nodes in batch flows. The SDK now exposes partitioning configuration directly on Link objects, with every method returning the link so calls can be chained.

Valid part_type values are 'auto', 'hash', 'modulus', 'range', 'roundrobin', 'entire', 'same', and 'random'.

Code:

batch_flow = project.create_flow(name='Partitioning Example', flow_type='batch')
source = batch_flow.add_stage('Row Generator', 'Source')
target = batch_flow.add_stage('Peek', 'Target')
link = source.connect_output_to(target)
link.name = 'Link_1'

link.set_partitioning(part_type='hash', perform_sort=True)
link.add_partition_key('CUSTOMER_ID', sorting=True, sort_order='asc')
link.add_partition_key('ORDER_DATE', sorting=True, sort_order='desc')

Output: Link_1 (src='Source', dest='Target')

Because each method returns the link, the same configuration can be written as a single chained call.

Code: link.set_partitioning(part_type='hash', perform_sort=True).add_partition_key('CUSTOMER_ID', sorting=True).add_partition_key('ORDER_DATE', sorting=True)

You can inspect the resulting configuration through the link's part_type and key_cols_part attributes.

Code: link.key_cols_part[0]

Output:

{'keyCol': 'CUSTOMER_ID', 'partitioning': True, 'sorting': True, 'ci-cs': 'cs', 'asc-desc': 'asc'}

Note: perform_sort is only valid for 'hash', 'range', and 'modulus' partitioning. When using modulus partitioning with perform_sort=False, only one partition key is allowed.

Jetstream Engines (Streaming)

Streaming environments now support two engine types. DataCollector remains the default and the traditional choice, with support for Docker, Podman, and Kubernetes container providers and only basic configuration required. Jetstream is a cloud-native streaming engine optimized for scalability.

A Jetstream environment requires two additional connections, both mandatory for this engine type:

• cluster_connection — a Confluent-Flink connection
• artifact_location_connection — an Amazon S3 connection, used to store engine artifacts

Code:

# Get the connections required by Jetstream
cluster_conn = project.connections.get(name='my-flink-cluster')
storage_conn = project.connections.get(name='my-object-storage')

jetstream_env = project.create_environment(
    name='Jetstream Production',
    description='Production Jetstream environment',
    engine_type='jetstream',
    cluster_connection=cluster_conn,
    artifact_location_connection=storage_conn,
    container_provider='docker'   # or 'podman'
)

Output:

Environment(name='Jetstream Production', environment_id='...', engine_type='jetstream')

Existing environments can be filtered by engine type, which makes it straightforward to inventory a project mid-migration.

Code:

datacollector_envs = project.environments.get_all(engine_type='data_collector')
jetstream_envs = project.environments.get_all(engine_type='jetstream')

Note: container_provider accepts 'docker', 'podman', or 'kubernetes' for DataCollector, but only 'docker' or 'podman' for Jetstream — Jetstream reaches the cluster through cluster_connection instead. Environments default to engine_type='data_collector' and container_provider='docker', so existing code is unaffected.

Engine Manager Commands (Streaming)

Restart and shutdown commands for streaming engines are now available at three levels, so you can target one engine, an entire environment, or a specific set of engines in a project.

A single engine

Code:

engine = environment.engines[0]
engine.restart()
engine.shutdown()

Output: <Response [200]>

All engines in an environment

Code:

environment.restart_all_engines()
environment.shutdown_all_engines()

Output: <Response [200]>

Specific engines from a project

Code:

project.restart_streaming_engines(engine)
project.shutdown_streaming_engines(engine)

Output: <Response [200]>

Note: The project level command methods require at least one engine argument. Passing no engines raises ValueError.

SaaS Role Lookups Now Use display_name

One behavioral change to note: when retrieving a single SaaS role, use display_name rather than role_id.

Code: platform.roles.get(display_name='New Role')

Output:

Role(display_name='New Role', role_type='custom_role', actions=['iam-groups.groups.update'])

Quality of Life Enhancements and Bug Fixes

In addition to these features, 2.0.0 includes numerous quality of life enhancements and bug fixes that improve the overall stability and reliability of the SDK.

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!