Branching

So far, we have only looked at pipelines that operate in a direct sequential manner, one step after the other. However, pyearthtools.pipeline can do more then that, and can create / manage branches of operations, for example, multiple sources, different operations applied to copys of data, mapping tuples over branches and more.

This notebook seeks to explain how to configure this

import pyearthtools.pipeline
import pyearthtools.data

import warnings

with warnings.catch_warnings(action="ignore"):
    import site_archive_nci

To make branches, when setting up the pipeline pass a tuple as a step, with the steps as elements within.

If this tuple is only one layer deep, as in (0,1,2) each step will be treated as a one element branch, i.e. a one step pipeline. However, if it is two deep i.e. ((0,1),(2,3)) each internal tuple is expanded into a pipeline. These rules the repeat to allow branches within branches within branches ….

Multiple Inputs

A common use for branches is multiple sources of data, i.e. from different datasets or the same dataset just prepared in a different way.

Simply pass a tuple as the first element of the pipeline

pyearthtools.pipeline.Pipeline(
    (pyearthtools.data.archive.ERA5.sample(), pyearthtools.data.archive.ERA5.sample()),
)

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             ERA5                           {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             ERA5[1]                        {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • ERA5
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • ERA5[1]
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}

Graph

Pipelines within branches

Using the above example, say we want to apply a step to only one branch.

We do this by nesting another tuple with the operation following the index like we would a normal pipeline

pyearthtools.pipeline.Pipeline(
    (
        (pyearthtools.data.archive.ERA5.sample(), pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'2_metre_temperature'})),
    pyearthtools.data.archive.ERA5.sample()),
)

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             ERA5                           {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             metadata.Rename                {'Rename': {'rename': {'2t': "'2_metre_temperature'"}}}
             ERA5[1]                        {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • ERA5
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • metadata.Rename
    {'Rename': {'rename': {'2t': '2_metre_temperature'}}}
    Rename : {'rename': {'2t': '2_metre_temperature'}}
  • ERA5[1]
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}

Graph

Merge

As part of all data type operations are steps specialised in merging tuples together or splitting an element into a tuple.

These are very useful for working with branches. Simply add them as you would a normal step

merged_pipeline = pyearthtools.pipeline.Pipeline(
    (
        (pyearthtools.data.archive.ERA5.sample(), pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'2_metre_temperature'})),
    pyearthtools.data.archive.ERA5.sample()),
    pyearthtools.pipeline.operations.xarray.Merge(),
)
merged_pipeline

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             ERA5                           {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             metadata.Rename                {'Rename': {'rename': {'2t': "'2_metre_temperature'"}}}
             ERA5[1]                        {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             join.Merge                     {'Merge': {'merge_kwargs': 'None'}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • ERA5
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • metadata.Rename
    {'Rename': {'rename': {'2t': '2_metre_temperature'}}}
    Rename : {'rename': {'2t': '2_metre_temperature'}}
  • ERA5[1]
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • join.Merge
    {'Merge': {'merge_kwargs': None}}
    Merge : {'merge_kwargs': None}

Graph

merged_pipeline['2000-01-01T00']

<xarray.Dataset> Size: 17MB
Dimensions:              (longitude: 1440, latitude: 721, time: 1)
Coordinates:
  * longitude            (longitude) float32 6kB -180.0 -179.8 ... 179.5 179.8
  * latitude             (latitude) float32 3kB 90.0 89.75 89.5 ... -89.75 -90.0
  * time                 (time) datetime64[ns] 8B 2000-01-01
Data variables:
    2_metre_temperature  (time, latitude, longitude) float64 8MB dask.array<chunksize=(1, 182, 360), meta=np.ndarray>
    2t                   (time, latitude, longitude) float64 8MB dask.array<chunksize=(1, 182, 360), meta=np.ndarray>
Attributes:
    Conventions:  CF-1.6
    history:      2020-09-30 21:47:55 UTC+1000 by era5_replication_tools-1.2....
    license:      Licence to use Copernicus Products: https://apps.ecmwf.int/...
    summary:      ERA5 is the fifth generation ECMWF atmospheric reanalysis o...
    title:        ERA5 single-levels reanalysis 2m_temperature 20000101-20000131

xarray.Dataset

• Dimensions:
  • longitude: 1440
  • latitude: 721
  • time: 1
• Coordinates: (3)
  • longitude
    (longitude)
    float32
    -180.0 -179.8 ... 179.5 179.8

    units :

    degrees_east

    long_name :

    longitude

    array([-180.  , -179.75, -179.5 , ...,  179.25,  179.5 ,  179.75],
          dtype=float32)

  • latitude
    (latitude)
    float32
    90.0 89.75 89.5 ... -89.75 -90.0

    units :

    degrees_north

    long_name :

    latitude

    array([ 90.  ,  89.75,  89.5 , ..., -89.5 , -89.75, -90.  ], dtype=float32)

  • time
    (time)
    datetime64[ns]
    2000-01-01

    long_name :

    time

    array(['2000-01-01T00:00:00.000000000'], dtype='datetime64[ns]')

• Data variables: (2)
  • 2_metre_temperature
    (time, latitude, longitude)
    float64
    dask.array<chunksize=(1, 182, 360), meta=np.ndarray>

    units :

    K

    long_name :

    2 metre temperature

    Array Chunk Bytes 7.92 MiB 511.88 kiB Shape (1, 721, 1440) (1, 182, 360) Dask graph 16 chunks in 4 graph layers Data type float64 numpy.ndarray

  • 2t
    (time, latitude, longitude)
    float64
    dask.array<chunksize=(1, 182, 360), meta=np.ndarray>

    units :

    K

    long_name :

    2 metre temperature

    Array Chunk Bytes 7.92 MiB 511.88 kiB Shape (1, 721, 1440) (1, 182, 360) Dask graph 16 chunks in 4 graph layers Data type float64 numpy.ndarray

• Indexes: (3)
  • longitude
    PandasIndex

    PandasIndex(Index([ -180.0, -179.75,  -179.5, -179.25,  -179.0, -178.75,  -178.5, -178.25,
            -178.0, -177.75,
           ...
             177.5,  177.75,   178.0,  178.25,   178.5,  178.75,   179.0,  179.25,
             179.5,  179.75],
          dtype='float32', name='longitude', length=1440))

  • latitude
    PandasIndex

    PandasIndex(Index([  90.0,  89.75,   89.5,  89.25,   89.0,  88.75,   88.5,  88.25,   88.0,
            87.75,
           ...
           -87.75,  -88.0, -88.25,  -88.5, -88.75,  -89.0, -89.25,  -89.5, -89.75,
            -90.0],
          dtype='float32', name='latitude', length=721))

  • time
    PandasIndex

    PandasIndex(DatetimeIndex(['2000-01-01'], dtype='datetime64[ns]', name='time', freq=None))

• Attributes: (5)

  Conventions :

  CF-1.6

  history :

  2020-09-30 21:47:55 UTC+1000 by era5_replication_tools-1.2.1: mv /g/data/rt52/admin/incoming/era5/single-levels/reanalysis/2t/2000/.2t_era5_oper_sfc_20000101-20000131.tmp /g/data/rt52/era5/single-levels/reanalysis/2t/2000/2t_era5_oper_sfc_20000101-20000131.nc 2020-09-30 21:47:01 UTC+1000 by era5_replication_tools-1.2.1: nccopy -k4 -d5 -s /g/data/zt77/admin/incoming/era5/single-levels/reanalysis/2t/2000/2t_era5_oper_sfc_20000101-20000131.nc /g/data/rt52/admin/incoming/era5/single-levels/reanalysis/2t/2000/.2t_era5_oper_sfc_20000101-20000131.tmp 2020-09-30 21:41:02 UTC+1000 by era5_replication_tools-1.2.1: curl --connect-timeout 20 --show-error --silent --max-time 36000 -o /g/data/zt77/admin/incoming/era5/single-levels/reanalysis/2t/2000/2t_era5_oper_sfc_20000101-20000131.nc http://136.156.133.42/cache-compute-0014/cache/data2/adaptor.mars.internal-1601465777.733087-10113-25-3cdf4566-c2b5-4a73-a929-70b2482bd6df.nc 2020-09-30 11:37:52 GMT by grib_to_netcdf-2.16.0: /opt/ecmwf/eccodes/bin/grib_to_netcdf -S param -o /cache/data2/adaptor.mars.internal-1601465777.733087-10113-25-3cdf4566-c2b5-4a73-a929-70b2482bd6df.nc /cache/tmp/3cdf4566-c2b5-4a73-a929-70b2482bd6df-adaptor.mars.internal-1601465777.7336645-10113-10-tmp.grib

  license :

  Licence to use Copernicus Products: https://apps.ecmwf.int/datasets/licences/copernicus/

  summary :

  ERA5 is the fifth generation ECMWF atmospheric reanalysis of the global climate. This file is part of the ERA5 replica hosted at NCI Australia. For more information please see http://dx.doi.org/10.25914/5f48874388857

  title :

  ERA5 single-levels reanalysis 2m_temperature 20000101-20000131

Within Pipeline Branches

Branches at the start are very useful, but they can also be used within a pipeline to take one sample of data and apply more then one sequence of operations to it. For this the data is copied, and then all branches run on a seperate copy.

The sample is then returned as a tuple of the same length as the number of branches, in the same order as the branches are given.

merged_pipeline = pyearthtools.pipeline.Pipeline(
    pyearthtools.data.archive.ERA5.sample(),
    (
        pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'2_metre_temperature'}), pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'t2m'}),
    ),
    pyearthtools.pipeline.operations.xarray.Merge(),
)
merged_pipeline

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             ERA5                           {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             metadata.Rename                {'Rename': {'rename': {'2t': "'2_metre_temperature'"}}}
             metadata.Rename[1]             {'Rename': {'rename': {'2t': "'t2m'"}}}
             join.Merge                     {'Merge': {'merge_kwargs': 'None'}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • ERA5
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • metadata.Rename
    {'Rename': {'rename': {'2t': '2_metre_temperature'}}}
    Rename : {'rename': {'2t': '2_metre_temperature'}}
  • metadata.Rename[1]
    {'Rename': {'rename': {'2t': 't2m'}}}
    Rename : {'rename': {'2t': 't2m'}}
  • join.Merge
    {'Merge': {'merge_kwargs': None}}
    Merge : {'merge_kwargs': None}

Graph

merged_pipeline['2000-01-01T00']

<xarray.Dataset> Size: 17MB
Dimensions:              (longitude: 1440, latitude: 721, time: 1)
Coordinates:
  * longitude            (longitude) float32 6kB -180.0 -179.8 ... 179.5 179.8
  * latitude             (latitude) float32 3kB 90.0 89.75 89.5 ... -89.75 -90.0
  * time                 (time) datetime64[ns] 8B 2000-01-01
Data variables:
    2_metre_temperature  (time, latitude, longitude) float64 8MB dask.array<chunksize=(1, 182, 360), meta=np.ndarray>
    t2m                  (time, latitude, longitude) float64 8MB dask.array<chunksize=(1, 182, 360), meta=np.ndarray>
Attributes:
    Conventions:  CF-1.6
    history:      2020-09-30 21:47:55 UTC+1000 by era5_replication_tools-1.2....
    license:      Licence to use Copernicus Products: https://apps.ecmwf.int/...
    summary:      ERA5 is the fifth generation ECMWF atmospheric reanalysis o...
    title:        ERA5 single-levels reanalysis 2m_temperature 20000101-20000131

xarray.Dataset

• Dimensions:
  • longitude: 1440
  • latitude: 721
  • time: 1
• Coordinates: (3)
  • longitude
    (longitude)
    float32
    -180.0 -179.8 ... 179.5 179.8

    units :

    degrees_east

    long_name :

    longitude

    array([-180.  , -179.75, -179.5 , ...,  179.25,  179.5 ,  179.75],
          dtype=float32)

  • latitude
    (latitude)
    float32
    90.0 89.75 89.5 ... -89.75 -90.0

    units :

    degrees_north

    long_name :

    latitude

    array([ 90.  ,  89.75,  89.5 , ..., -89.5 , -89.75, -90.  ], dtype=float32)

  • time
    (time)
    datetime64[ns]
    2000-01-01

    long_name :

    time

    array(['2000-01-01T00:00:00.000000000'], dtype='datetime64[ns]')

• Data variables: (2)
  • 2_metre_temperature
    (time, latitude, longitude)
    float64
    dask.array<chunksize=(1, 182, 360), meta=np.ndarray>

    units :

    K

    long_name :

    2 metre temperature

    Array Chunk Bytes 7.92 MiB 511.88 kiB Shape (1, 721, 1440) (1, 182, 360) Dask graph 16 chunks in 4 graph layers Data type float64 numpy.ndarray

  • t2m
    (time, latitude, longitude)
    float64
    dask.array<chunksize=(1, 182, 360), meta=np.ndarray>

    units :

    K

    long_name :

    2 metre temperature

    Array Chunk Bytes 7.92 MiB 511.88 kiB Shape (1, 721, 1440) (1, 182, 360) Dask graph 16 chunks in 4 graph layers Data type float64 numpy.ndarray

• Indexes: (3)
  • longitude
    PandasIndex

    PandasIndex(Index([ -180.0, -179.75,  -179.5, -179.25,  -179.0, -178.75,  -178.5, -178.25,
            -178.0, -177.75,
           ...
             177.5,  177.75,   178.0,  178.25,   178.5,  178.75,   179.0,  179.25,
             179.5,  179.75],
          dtype='float32', name='longitude', length=1440))

  • latitude
    PandasIndex

    PandasIndex(Index([  90.0,  89.75,   89.5,  89.25,   89.0,  88.75,   88.5,  88.25,   88.0,
            87.75,
           ...
           -87.75,  -88.0, -88.25,  -88.5, -88.75,  -89.0, -89.25,  -89.5, -89.75,
            -90.0],
          dtype='float32', name='latitude', length=721))

  • time
    PandasIndex

    PandasIndex(DatetimeIndex(['2000-01-01'], dtype='datetime64[ns]', name='time', freq=None))

• Attributes: (5)

  Conventions :

  CF-1.6

  history :

  2020-09-30 21:47:55 UTC+1000 by era5_replication_tools-1.2.1: mv /g/data/rt52/admin/incoming/era5/single-levels/reanalysis/2t/2000/.2t_era5_oper_sfc_20000101-20000131.tmp /g/data/rt52/era5/single-levels/reanalysis/2t/2000/2t_era5_oper_sfc_20000101-20000131.nc 2020-09-30 21:47:01 UTC+1000 by era5_replication_tools-1.2.1: nccopy -k4 -d5 -s /g/data/zt77/admin/incoming/era5/single-levels/reanalysis/2t/2000/2t_era5_oper_sfc_20000101-20000131.nc /g/data/rt52/admin/incoming/era5/single-levels/reanalysis/2t/2000/.2t_era5_oper_sfc_20000101-20000131.tmp 2020-09-30 21:41:02 UTC+1000 by era5_replication_tools-1.2.1: curl --connect-timeout 20 --show-error --silent --max-time 36000 -o /g/data/zt77/admin/incoming/era5/single-levels/reanalysis/2t/2000/2t_era5_oper_sfc_20000101-20000131.nc http://136.156.133.42/cache-compute-0014/cache/data2/adaptor.mars.internal-1601465777.733087-10113-25-3cdf4566-c2b5-4a73-a929-70b2482bd6df.nc 2020-09-30 11:37:52 GMT by grib_to_netcdf-2.16.0: /opt/ecmwf/eccodes/bin/grib_to_netcdf -S param -o /cache/data2/adaptor.mars.internal-1601465777.733087-10113-25-3cdf4566-c2b5-4a73-a929-70b2482bd6df.nc /cache/tmp/3cdf4566-c2b5-4a73-a929-70b2482bd6df-adaptor.mars.internal-1601465777.7336645-10113-10-tmp.grib

  license :

  Licence to use Copernicus Products: https://apps.ecmwf.int/datasets/licences/copernicus/

  summary :

  ERA5 is the fifth generation ECMWF atmospheric reanalysis of the global climate. This file is part of the ERA5 replica hosted at NCI Australia. For more information please see http://dx.doi.org/10.25914/5f48874388857

  title :

  ERA5 single-levels reanalysis 2m_temperature 20000101-20000131

Mapped Branches

In addition to copying the data to be modified multiple different ways, elements of a tuple can be mapped across to the branches instead.

So, given a branch of two, a tuple of size two can be split such that the first element is given to the first branch, and the second to the second and so on.

This uses the exact same syntax as above, but instead include a ‘map’ literal as the last specified branch. It is then removed and the branch configured for mapping.

The size of the tuple and the branches must match exactly, or else an error will be thrown.

pyearthtools.pipeline.Pipeline(
    (pyearthtools.data.archive.ERA5.sample(),pyearthtools.data.archive.ERA5.sample()),
    (
        pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'2_metre_temperature'}), pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'t2m'}), 'map'
    ),
    pyearthtools.pipeline.operations.xarray.Merge(),
)

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             ERA5                           {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             ERA5[1]                        {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             metadata.Rename                {'Rename': {'rename': {'2t': "'2_metre_temperature'"}}}
             metadata.Rename[1]             {'Rename': {'rename': {'2t': "'t2m'"}}}
             join.Merge                     {'Merge': {'merge_kwargs': 'None'}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • ERA5
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • ERA5[1]
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • metadata.Rename
    {'Rename': {'rename': {'2t': '2_metre_temperature'}}}
    Rename : {'rename': {'2t': '2_metre_temperature'}}
  • metadata.Rename[1]
    {'Rename': {'rename': {'2t': 't2m'}}}
    Rename : {'rename': {'2t': 't2m'}}
  • join.Merge
    {'Merge': {'merge_kwargs': None}}
    Merge : {'merge_kwargs': None}

Graph

Empty Branches

If no modifications are to be applied to a certain element of a tuple, a pyearthtools.pipeline.Empty() can be used to not modify the data but fullfill the length requirements.

pyearthtools.pipeline.Pipeline(
    (pyearthtools.data.archive.ERA5.sample(),pyearthtools.data.archive.ERA5.sample()),
    (
        pyearthtools.pipeline.operations.xarray.metadata.Rename({'2t':'2_metre_temperature'}), pyearthtools.pipeline.Empty(), 'map'
    ),
    pyearthtools.pipeline.operations.xarray.Merge(),
)

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             ERA5                           {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             ERA5[1]                        {'ERA5': {'level_value': 'None', 'product': "'reanalysis'", 'variables': "['2t']"}}
             metadata.Rename                {'Rename': {'rename': {'2t': "'2_metre_temperature'"}}}
             marker.Empty                   {'Empty': {}}
             join.Merge                     {'Merge': {'merge_kwargs': 'None'}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • ERA5
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • ERA5[1]
    {'ERA5': {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}}
    ERA5 : {'level_value': None, 'product': 'reanalysis', 'variables': ['2t']}
  • metadata.Rename
    {'Rename': {'rename': {'2t': '2_metre_temperature'}}}
    Rename : {'rename': {'2t': '2_metre_temperature'}}
  • marker.Empty
    {'Empty': {}}
    Empty : {}
  • join.Merge
    {'Merge': {'merge_kwargs': None}}
    Merge : {'merge_kwargs': None}

Graph

Full example

With the above knowledge, you should be equipped to build a branching pipeline, but just to make sure, below is an example using Markers to print out the data as it follows through. Hopefully it is clear how these steps work together

Step = lambda x: pyearthtools.pipeline.Marker(x, print = True) # Fake Step

example_pipe = pyearthtools.pipeline.Pipeline(
    (
        pyearthtools.pipeline.Pipeline(Step('DataSource'), Step('DataSource_specific')), # Data Source with a specific step
        Step('Another_DataSource'), # Lone Data source
    ),
    Step('joint_operation'), # An operation to be applied to the tuple of sources
    (
        Step('Square'), Step('AddByOne'), 'map', # Tuple is then mapped between these two fake operations each recieving a single element
    ),
    Step('Merge'), # A hypothetical merge operation
)
example_pipe

Pipeline
    Description                    `pyearthtools.pipeline` Data Pipeline


    Initialisation
             exceptions_to_ignore           None
             iterator                       None
             sampler                        None
    Steps
             marker.Marker                  {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'DataSource'"}}
             marker.Marker[1]               {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'DataSource_specific'"}}
             marker.Marker[2]               {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'Another_DataSource'"}}
             marker.Marker[3]               {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'joint_operation'"}}
             marker.Marker[4]               {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'Square'"}}
             marker.Marker[5]               {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'AddByOne'"}}
             marker.Marker[6]               {'Marker': {'print': 'True', 'print_shape': 'False', 'shape': "'note'", 'text': "'Merge'"}}

Pipeline

Description:

`pyearthtools.pipeline` Data Pipeline

• Initialisation:
  • exceptions_to_ignore
    None
    None
  • iterator
    None
    None
  • sampler
    None
    None
• Steps:
  • marker.Marker
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'DataSource'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'DataSource'}
  • marker.Marker[1]
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'DataSource_specific'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'DataSource_specific'}
  • marker.Marker[2]
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'Another_DataSource'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'Another_DataSource'}
  • marker.Marker[3]
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'joint_operation'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'joint_operation'}
  • marker.Marker[4]
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'Square'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'Square'}
  • marker.Marker[5]
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'AddByOne'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'AddByOne'}
  • marker.Marker[6]
    {'Marker': {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'Merge'}}
    Marker : {'print': True, 'print_shape': False, 'shape': 'note', 'text': 'Merge'}

Graph

example_pipe.apply(1)

At marker 'DataSource' sample was:
1
At marker 'DataSource_specific' sample was:
1
At marker 'Another_DataSource' sample was:
1
At marker 'joint_operation' sample was:
(1, 1)
At marker 'Square' sample was:
1
At marker 'AddByOne' sample was:
1
At marker 'Merge' sample was:
(1, 1)

(1, 1)