# Pipelines¶

Pipelines are an integral part of River. We encourage their usage and apply them in many of their examples.

The `compose.Pipeline`

contains all the logic for building and applying pipelines. A pipeline is essentially a list of estimators that are applied in sequence. The only requirement is that the first `n - 1`

steps be transformers. The last step can be a regressor, a classifier, a clusterer, a transformer, etc.

Here is an example:

```
from river import compose
from river import linear_model
from river import preprocessing
from river import feature_extraction
model = compose.Pipeline(
preprocessing.StandardScaler(),
feature_extraction.PolynomialExtender(),
linear_model.LinearRegression()
)
```

You can also use the `|`

operator, as so:

```
model = (
preprocessing.StandardScaler() |
feature_extraction.PolynomialExtender() |
linear_model.LinearRegression()
)
```

Or, equally:

```
model = preprocessing.StandardScaler()
model |= feature_extraction.PolynomialExtender()
model |= linear_model.LinearRegression()
```

A pipeline, as any River estimator, has a `_repr_html_`

method, which can be used to visualize it in Jupyter-like notebooks:

```
model
```

StandardScaler

```
StandardScaler (
with_std=True
)
```

PolynomialExtender

```
PolynomialExtender (
degree=2
interaction_only=False
include_bias=False
bias_name="bias"
)
```

LinearRegression

```
LinearRegression (
optimizer=SGD (
lr=Constant (
learning_rate=0.01
)
)
loss=Squared ()
l2=0.
l1=0.
intercept_init=0.
intercept_lr=Constant (
learning_rate=0.01
)
clip_gradient=1e+12
initializer=Zeros ()
)
```

`compose.Pipeline`

implements a `learn_one`

method which in sequence calls the `learn_one`

of each component and a `predict_one`

(resp `predict_proba_one`

) method which calls `transform_one`

on the first `n - 1`

steps and `predict_one`

(resp `predict_proba_one`

) on the last step.

Here is a small example to illustrate the previous point:

```
from river import datasets
dataset = datasets.TrumpApproval()
x, y = next(iter(dataset))
x, y
```

```
({'ordinal_date': 736389,
'gallup': 43.843213,
'ipsos': 46.19925042857143,
'morning_consult': 48.318749,
'rasmussen': 44.104692,
'you_gov': 43.636914000000004},
43.75505)
```

We can predict the target value of a new sample by calling the `predict_one`

method, however, by default, `predict_one`

does not update any model parameter, therefore the predictions will be 0 and the model parameters will remain the default values (0 for `StandardScaler`

component):

```
for (x, y) in dataset.take(2):
print(f"{model.predict_one(x)=:.2f}, {y=:.2f}")
print(f"{model['StandardScaler'].means = }")
```

```
model.predict_one(x)=0.00, y=43.76
model['StandardScaler'].means = defaultdict(<class 'float'>, {'ordinal_date': 0.0, 'gallup': 0.0, 'ipsos': 0.0, 'morning_consult': 0.0, 'rasmussen': 0.0, 'you_gov': 0.0})
model.predict_one(x)=0.00, y=43.71
model['StandardScaler'].means = defaultdict(<class 'float'>, {'ordinal_date': 0.0, 'gallup': 0.0, 'ipsos': 0.0, 'morning_consult': 0.0, 'rasmussen': 0.0, 'you_gov': 0.0})
```

`learn_one`

updates pipeline stateful steps, parameters and the prediction change:

```
for (x, y) in dataset.take(2):
model.learn_one(x, y)
print(f"{model.predict_one(x)=:.2f}, {y=:.2f}")
print(f"{model['StandardScaler'].means = }")
```

```
model.predict_one(x)=0.88, y=43.76
model['StandardScaler'].means = defaultdict(<class 'float'>, {'ordinal_date': 736389.0, 'gallup': 43.843213, 'ipsos': 46.19925042857143, 'morning_consult': 48.318749, 'rasmussen': 44.104692, 'you_gov': 43.636914000000004})
model.predict_one(x)=9.44, y=43.71
model['StandardScaler'].means = defaultdict(<class 'float'>, {'ordinal_date': 736389.5, 'gallup': 43.843213, 'ipsos': 46.19925042857143, 'morning_consult': 48.318749, 'rasmussen': 45.104692, 'you_gov': 42.636914000000004})
```

Each component of the pipeline has been updated with the new data point.

A pipeline is a very powerful tool that can be used to chain together multiple steps in a machine learning workflow.

Notice that it is also possible to call `transform_one`

with a pipeline, this method will run `transform_one`

of each transformer in it, and return the result of the last transformer (which is thus the penultimate step if the last step is a *predictor* or *clusterer*, while it is the last step if the last step is a *transformer*):

```
model.transform_one(x)
```

```
{'ordinal_date': 1.0,
'gallup': 0.0,
'ipsos': 0.0,
'morning_consult': 0.0,
'rasmussen': 1.0,
'you_gov': -1.0,
'ordinal_date*ordinal_date': 1.0,
'gallup*ordinal_date': 0.0,
'ipsos*ordinal_date': 0.0,
'morning_consult*ordinal_date': 0.0,
'ordinal_date*rasmussen': 1.0,
'ordinal_date*you_gov': -1.0,
'gallup*gallup': 0.0,
'gallup*ipsos': 0.0,
'gallup*morning_consult': 0.0,
'gallup*rasmussen': 0.0,
'gallup*you_gov': -0.0,
'ipsos*ipsos': 0.0,
'ipsos*morning_consult': 0.0,
'ipsos*rasmussen': 0.0,
'ipsos*you_gov': -0.0,
'morning_consult*morning_consult': 0.0,
'morning_consult*rasmussen': 0.0,
'morning_consult*you_gov': -0.0,
'rasmussen*rasmussen': 1.0,
'rasmussen*you_gov': -1.0,
'you_gov*you_gov': 1.0}
```

In many cases, you might want to connect a step to multiple steps. For instance, you might to extract different kinds of features from a single input. An elegant way to do this is to use a `compose.TransformerUnion`

. Essentially, the latter is a list of transformers who's results will be merged into a single `dict`

when `transform_one`

is called.

As an example let's say that we want to apply a `feature_extraction.RBFSampler`

as well as the `feature_extraction.PolynomialExtender`

. This may be done as so:

```
model = (
preprocessing.StandardScaler() |
(feature_extraction.PolynomialExtender() + feature_extraction.RBFSampler()) |
linear_model.LinearRegression()
)
model
```

StandardScaler

```
StandardScaler (
with_std=True
)
```

PolynomialExtender

```
PolynomialExtender (
degree=2
interaction_only=False
include_bias=False
bias_name="bias"
)
```

RBFSampler

```
RBFSampler (
gamma=1.
n_components=100
seed=None
)
```

LinearRegression

```
LinearRegression (
optimizer=SGD (
lr=Constant (
learning_rate=0.01
)
)
loss=Squared ()
l2=0.
l1=0.
intercept_init=0.
intercept_lr=Constant (
learning_rate=0.01
)
clip_gradient=1e+12
initializer=Zeros ()
)
```

Note that the `+`

symbol acts as a shorthand notation for creating a `compose.TransformerUnion`

, which means that we could have declared the above pipeline as so:

```
model = (
preprocessing.StandardScaler() |
compose.TransformerUnion(
feature_extraction.PolynomialExtender(),
feature_extraction.RBFSampler()
) |
linear_model.LinearRegression()
)
```

Pipelines provide the benefit of removing a lot of cruft by taking care of tedious details for you. They also enable to clearly define what steps your model is made of.

Finally, having your model in a single object means that you can move it around more easily.

Note that you can include user-defined functions in a pipeline by using a `compose.FuncTransformer`

.

## Learning during predict¶

In online machine learning, we can update the unsupervised parts of our model when a sample arrives. We don't *really* have to wait for the ground truth to arrive in order to update unsupervised estimators that don't depend on it.

In other words, in a pipeline, `learn_one`

updates the supervised parts, whilst `predict_one`

(or `predict_proba_one`

for that matter) **can** update the unsupervised parts, which often yields better results.

In river, we can achieve this behavior using a dedicated context manager: `compose.learn_during_predict`

.

Here is the same example as before, with the only difference of activating the such learning during predict behavior:

```
model = (
preprocessing.StandardScaler() |
feature_extraction.PolynomialExtender() |
linear_model.LinearRegression()
)
```

```
with compose.learn_during_predict():
for (x, y) in dataset.take(2):
print(f"{model.predict_one(x)=:.2f}, {y=:.2f}")
print(f"{model['StandardScaler'].means = }")
```

```
model.predict_one(x)=0.00, y=43.76
model['StandardScaler'].means = defaultdict(<class 'float'>, {'ordinal_date': 736389.0, 'gallup': 43.843213, 'ipsos': 46.19925042857143, 'morning_consult': 48.318749, 'rasmussen': 44.104692, 'you_gov': 43.636914000000004})
model.predict_one(x)=0.00, y=43.71
model['StandardScaler'].means = defaultdict(<class 'float'>, {'ordinal_date': 736389.5, 'gallup': 43.843213, 'ipsos': 46.19925042857143, 'morning_consult': 48.318749, 'rasmussen': 45.104692, 'you_gov': 42.636914000000004})
```

Calling `predict_one`

within this context will update each transformer of the pipeline. For instance here we can see that the mean of each feature of the standard scaler step have been updated.

On the other hand, the supervised part of our pipeline, the linear regression, has not been updated or learned anything yet. Hence the prediction on any sample will be nil because each weight is still equal to 0.

```
model.predict_one(x), model["LinearRegression"].weights
```

```
(0.0, {})
```

## Performance Comparison¶

One may wonder what is the advantage of learning during predict. Let's compare the performance of a pipeline with and without learning during predict, in two scenarios: one in which the flow of data stays the same, we just update

```
from contextlib import nullcontext
from river import metrics
import pandas as pd
```

```
def score_pipeline(learn_during_predict: bool, n_learning_samples: int | None = None) -> float:
"""Scores a pipeline on the TrumpApproval dataset.
Parameters
----------
learn_during_predict : bool
Whether or not to learn the unsupervided components during the prediction step.
If False it will only learn when `learn_one` is explicitly called.
n_learning_samples : int | None
Number of samples used to `learn_one`.
Return
------
MAE : float
Mean absolute error of the pipeline on the dataset
"""
dataset = datasets.TrumpApproval()
model = (
preprocessing.StandardScaler() |
linear_model.LinearRegression()
)
metric = metrics.MAE()
ctx = compose.learn_during_predict if learn_during_predict else nullcontext
n_learning_samples = n_learning_samples or dataset.n_samples
with ctx():
for _idx, (x, y) in enumerate(dataset):
y_pred = model.predict_one(x)
metric.update(y, y_pred)
if _idx < n_learning_samples:
model.learn_one(x, y)
return metric.get()
```

```
max_samples = datasets.TrumpApproval().n_samples
results = [
{
"learn_during_predict": learn_during_predict,
"pct_learning_samples": round(100*n_learning_samples/max_samples, 0),
"mae": score_pipeline(learn_during_predict=learn_during_predict, n_learning_samples=n_learning_samples)
}
for learn_during_predict in (True, False)
for n_learning_samples in range(max_samples, max_samples//10, -(max_samples//10))
]
```

```
(pd.DataFrame(results)
.pivot(columns="learn_during_predict", index="pct_learning_samples", values="mae")
.sort_index(ascending=False)
.style.format_index('{0}%')
)
```

learn_during_predict | False | True |
---|---|---|

pct_learning_samples | ||

100.0% | 1.314548 | 1.347434 |

90.0% | 1.629333 | 1.355274 |

80.0% | 2.712125 | 1.371599 |

70.0% | 4.840620 | 1.440773 |

60.0% | 8.918634 | 1.498240 |

50.0% | 15.112753 | 1.878434 |

40.0% | 26.387331 | 2.105553 |

30.0% | 42.997083 | 3.654709 |

20.0% | 90.703102 | 3.504950 |

10.0% | 226.836953 | 4.803600 |

As we can see from the resulting table above, the scores are comparable only in the case in which the percentage of learning samples above 90%. After that the score starts to degrade quite fast as the percentage of learning samples decreases, and it is very remarkable (one order of magnitude or more) when less than 50% of the samples are used for learning.

Although a simple case, this examplify how powerful it can be to learn unsupervised components during predict.