Pipelines

A pipeline is a set of readers that pump records into the pipeline and a sequence of commands (operations) to apply to the records as they flow through the pipeline.

• Variables can be declared for use throughout the pipeline and are evaluated when the pipeline is constructed. If a variable specifies overridable="true", it may be overridden by a command line variable or a regi.json variable.

• Confirm checks are evaluated after variables and if they fail an error occurs when the pipeline starts.

• Metadata can be pinned to objects to pass state down the pipeline. Variables or metadata starting with # are deemed temporary and variables starting with !are considered sensitive and not available for expression dynamic evaluations.

Pipeline configuration illustration

Example:

<pipeline id="MyFirstPipeline">

<!-- A pipeline may define and evaluate variables for use throughout the pipeline -->

<variable name="input-waypoint" value="{{ $[waypoint-root] }}\captured\tags\*" />

<variable name="output-waypoint" value="{{ $[waypoint-root] }}\consolidated\tags" />

<!-- A pipeline may define confirmation checks as prerequisites to execution -->

<confirm where="{{ requireRegiVersion(2.7).isFalse }}" message="Regi v2.7 is needed" />

<!-- 1-n readers pump records into the pipeline; here we read all excel files from a path -->

<read>

<file folder="input" file="*.xls*" recurse="true">

<excel>

<!-- We can pin metadata, so it is available later -->

<pin>

<metadata name="#sheet" value="{{ $[sheet.index] }}" />

<metadata name="#index" value="{{ $[sheet.name] }}" />

</pin>

</excel>

</file>

<pin>

<metadata name="id" value="{{ @[tag number] }}" />

</pin>

</read>

<!-- Commands manipulate records (filter, modify, sort, etc.) as it flows down the pipeline -->

<where condition="{{ $[this.id].isSet }}" />

<modify>

<update value="{{ $[attr.value].sanitise }}" />

</modify>

<merge key="{{ $[this.id] }}" action="AppendValue" />

<sort key="{{ $[this.id] }}" />

<merge key="{{ $[this.id] }}" action="AppendValue" />

<!-- 0-n writers write records out to a file endpoint which can write out several files -->

<write>

<!-- We can reference variables, attributes and metadata we previously pinned -->

<file

folder="output"

file="{{ $[this.#index].padNumbers(5) }}_{{ $[this.#sheet].toUpper }}.csv"

recurse="true">

<text template="csv" />

</file>

</write>

</pipeline>

A PipelineFactory is used to create a PipelineExecutable, which is then used to perform a set of instructions. The PipelineExecutable comprises several PipelineComponents, that execute commands and manage buffers to control flow through the pipeline.

The following diagram shows an illustration of outputs of pipeline factory:

Each Command managed by a PipelineComponent may specify several settings, which in general accept DataExpressions so that a value may be constant or evaluated to some rule.

The pipeline factory uses a DataExpressionCompiler to configure expressions used in the PipelineComponent settings, as shown in the following diagram:

Pipelines effectively form configurable, testable and reusable bits of logic (simple programs) that may be executed. A useful extension to this would be pipeline sections – for example, using the current capture configurations for the EDW, it may be seen that some logic for reading files is copied and pasted into multiple configurations.

Variables

A pipeline may set up one or more variables to be evaluated when the pipeline begins execution. These may use DataExpressions to evaluate something useful and the resulting values will be available for use by any commands within the pipeline.

The following example illustrates reading some information from ISM GEICL and making it available within processing:

Example:

<pipeline id="Capture {{ $interface }}">

<variable

name="interface-path"

value="{{ '/'.join('InfoInterface:Data Sources', $interface) }}" />

<variable

name="header-row"

value="{{ getClosestExtension($[path], 'HeaderRow').fallback(1) }}" />

<variable

name="first-data-row"

value="{{ getClosestExtension($[path], 'FirstDataRow').fallback(0) }}" />

<variable

name="file-path"

value="{{ getClosestExtension($[path], 'FilePath').fallback('*') }}" />

</pipeline>

Confirmations

A pipeline may specify one or more confirmation checks. If these checks do not pass, this produces error by default and stops the pipeline execution.

The following example specifies some requirements such as minimum exe version:

Example:

<pipeline id="Consolidate $category">

<confirm

where="{{ requireRegiVersion(2.7).isFalse }}"

message="Registers gateway version 2.7 or better is required to run this pipeline" />

<confirm

where="{{ $category.isUnset }}"

message="$category is required" />

</pipeline>

Confirmation checks are used to set pre-requisite expectations and prevent execution, which may result in some difficulty to understand error logs.

Settings and data expressions

The following settings read excel files and produce several outputs,the last write splits data up into folders and files according to the specified expressions, illustrating the declarative, rule-based, and intention of the design:

Example:

<pipeline>

<read>

<file folder="." file="input.xls">

<excel />

</file>

</read>

<write>

<file folder="." file="doc_data.csv">

<text template="csv" />

</file>

</write>

<write>

<file

folder=".\{{@Plant}}\{{COALESCE(@Unit, 'gen')}}"

file="doc_{{@Discipline}}_data.csv">

<text template="csv" />

</file>

</write>

</pipeline>

Waypoints

Waypoints provide a mechanism to allow deltas (changes) to flow from one node (for example, a source system) to another (for example, an operations portal) with the ability to manipulate, filter, and aggregate data along the way. The waypoints provide a mechanism that can handle deltas - both inbound and outbound to enable deltas - to flow in a chained fashion without requiring to work out changes at every step (which becomes expensive and difficult to manage over a larger set of data).

The following diagram illustrates a Waypoint (shown as WP):

This is used for storing TAG or DOCUMENT attribute data.

Waypoint Aggregation

Within the EDW solution, a key usage of Waypoints is to aggregate and track captured data from a source either via full or partial (delta) loads. This is done by identifying a key and capturing all attributes that are found in the source against that key. Subsequent capture updates may replace or update that tracked record.

Waypoint Sharding

Waypoints sharding allows you to partition data into smaller chunks. It prevents an ID to be lost and assigns it an integer index (unless an administrator deletes the waypoint in its entirety).

The following example shows a waypoint sharding where one file can be generated for every 2500 objects. If an object is deleted for some reason (not included by some previous logic), then at least one file will have less than 2500 records:

• Write out portal data

• Shard data in chunks of $[eiwm-size] on waypoint index.

• The tag data is put into an 'operational dataset' because it is previously set.

Example:

<write name="Publish_Staging">

<file

name="Publish_Staging"

template="replace-folder"

folder="{{ $[output-folder] }}"

file="{{ $[this.waypoint.index].divide(2500).padnumbers(9) }}_null.xml">

<eiwm

template="all"

template-id="master://{{ $category }}/{{$[file.Name] }}/full"

context="{{ $[this.context] }}"

include-empty-attributes="false" />

</file>

</write>

This mechanism enables us to manage a staging area by rules rather than manually (importantly, including the removal of information and not moving data between files). General expectation from the implemented projects is that 2500 objects with up to six datasets can be expected to load in 10-30 seconds, meaning that in AIM there might be a delay of up to 30 seconds sometimes when trying to view an object.

Waypoints Nesting

Waypoints support nesting as required, in a folder implementation this means nested folders.

The EDW solution uses waypoint nesting to record all interpreted registers under a root folder. Subsequent processing may then read the root folder waypoint, which simply reads all waypoints found underneath that root.

Since no interpretation or mapping is done at this point, the entry point to the EDW solution tracks the aggregated data in a structured form very close to how it is received. The EDW solution then supports multiple solutions from that one capture place.