Hierarchical Data Stage#

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#

Every Hierarchical Data stage has exactly one Assembly. Accessing the HierarchicalDataStage.assembly property for the first time creates a fresh assembly with an InputStep and an OutputStep already connected to each other.

The simplest possible assembly passes data straight through without any processing step in between. In that case you only need to call OutputStep.propagate() pointing at the Input step, which copies all columns from the incoming link schema:

>>> from ibm_watsonx_data_integration.services.datastage.models.hierarchical.steps import (
...     InputStep, OutputStep
... )
>>>
>>> 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'
>>>
>>> # Access the assembly — Input and Output steps are created automatically
>>> assembly = hd_stage.assembly
>>> isinstance(assembly.input_step, InputStep)
True
>>> isinstance(assembly.output_step, OutputStep)
True
>>>
>>> # Propagate all columns from the incoming link straight to the output
>>> assembly.output_step.propagate(assembly.input_step)
>>>
>>> project.update_flow(flow)
<Response [201]>

Adding Steps to the Assembly#

Use Assembly.add_step() to insert a processing step into the assembly. The method takes the step type as a string and an optional label, and returns the typed step instance.

Because the Input and Output steps are connected by default, you must:

Disconnect the default Input → Output link.
Connect the new step between them using connect_output_to().
Call OutputStep.propagate() passing the new step so the output columns are derived from it.

The following example adds a Sort step:

>>> 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 → Output connection
>>> assembly.input_step.disconnect_output_from(assembly.output_step)
>>>
>>> # 2. Wire: Input → Sort → 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)
<Response [201]>

Propagating Columns to the Output Step#

OutputStep.propagate() copies columns from the given source to the Output step. The source argument can be:

A step object — the output path is derived automatically from the step’s configuration.
A raw path string such as 'top/InputLinks/Link_1' — used when pointing directly at a list path.

The chunk_path keyword argument is used by XML and JSON Parser steps to identify the repeating element in the schema. When omitted the deepest repeating element is auto-detected.

Finalizing the Assembly#

Calling HierarchicalDataStage.finalize_assembly() serializes the in-memory assembly into the compressed stage parameters the platform stores. This is called automatically by Project.update_flow(), so you do not need to call it explicitly. The only reason to call it manually is to inspect the serialized parameters before saving:

>>> hd_stage.finalize_assembly()
>>> hd_stage.configuration.e2_assembly is not None
True

Available Step Types#

The table below lists every step type string accepted by assembly.add_step() and the class it returns.

Type string	Returned class	Purpose
`'Sort'`	`SortStep`	Sort a list of records by one or more fields
`'Aggregate'`	`AggregateStep`	Apply aggregate functions (Count, Sum, Min, Max, …) to a list
`'HJoin'`	`HJoinStep`	Hierarchical join — embed child records inside matching parent records
`'HPivot'`	`HPivotStep`	Pivot column values into child rows (horizontal → vertical)
`'VPivot'`	`VPivotStep`	Pivot categorised rows into columns (vertical → horizontal)
`'Regroup'`	`RegroupStep`	Re-structure flat records into nested parent/child groups
`'Switch'`	`SwitchStep`	Route records into named branches based on conditions
`'OrderJoin'`	`OrderJoinStep`	Merge two ordered lists by interleaving their records
`'JSONParser'`	`JSONParserStep`	Parse JSON input from a file, string set, or LOB column
`'JSONComposer'`	`JSONComposerStep`	Compose JSON output to a file, string set, or LOB column
`'XMLParser'`	`XMLParserStep`	Parse XML input using an XSD schema from a schema library
`'XMLComposer'`	`XMLComposerStep`	Compose XML output using an XSD schema from a schema library

Attaching a Schema Library#

XML and JSON Parser/Composer steps require a SchemaLibrary to be attached to the stage. Use HierarchicalDataStage.add_schema_library():

>>> hd_stage.add_schema_library(schema_lib)
>>> project.update_flow(flow)
<Response [200]>

Duplicate entries are silently ignored. To detach a library use HierarchicalDataStage.remove_schema_library(). For full details on creating and managing schema libraries, see XML Schema Libraries.