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 has a draw method that can be used to visualize it:

model

StandardScaler


{'counts': Counter(),
 'means': defaultdict(<class 'float'>, {}),
 'vars': defaultdict(<class 'float'>, {}),
 'with_std': True}

PolynomialExtender


{'bias_name': 'bias',
 'degree': 2,
 'include_bias': False,
 'interaction_only': False}

LinearRegression


{'_weights': {},
 '_y_name': None,
 'clip_gradient': 1000000000000.0,
 'initializer': Zeros (),
 'intercept': 0.0,
 'intercept_init': 0.0,
 'intercept_lr': Constant({'learning_rate': 0.01}),
 'l1': 0.0,
 'l2': 0.0,
 'loss': Squared({}),
 'optimizer': SGD({'lr': Constant({'learning_rate': 0.01}), 'n_iterations': 0})}

compose.Pipeline inherits from base.Estimator, which means that it has a learn_one method. You would expect learn_one to update each estimator, but that's not actually what happens. Instead, the transformers are updated when predict_one (or predict_proba_one for that matter) is called. Indeed, in online machine learning, we can update the unsupervised parts of our model when a sample arrives. We don't 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 updates the unsupervised parts. It's important to be aware of this behavior, as it is quite different to what is done in other libraries that rely on batch machine learning.

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)

Let us call predict_one, which will update each transformer, but won't update the linear regression.

model.predict_one(x)

0.0

The prediction is nil because each weight of the linear regression is equal to 0.

model['StandardScaler'].means

defaultdict(float,
            {'ordinal_date': 736389.0,
             'gallup': 43.843213,
             'ipsos': 46.19925042857143,
             'morning_consult': 48.318749,
             'rasmussen': 44.104692,
             'you_gov': 43.636914000000004})

As we can see, the means of each feature have been updated, even though we called predict_one and not learn_one.

Note that if you call transform_one with a pipeline who's last step is not a transformer, then the output from the last transformer (which is thus the penultimate step) will be returned:

model.transform_one(x)

{'ordinal_date': 0.0,
 'gallup': 0.0,
 'ipsos': 0.0,
 'morning_consult': 0.0,
 'rasmussen': 0.0,
 'you_gov': 0.0,
 'ordinal_date*ordinal_date': 0.0,
 'gallup*ordinal_date': 0.0,
 'ipsos*ordinal_date': 0.0,
 'morning_consult*ordinal_date': 0.0,
 'ordinal_date*rasmussen': 0.0,
 'ordinal_date*you_gov': 0.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': 0.0,
 'rasmussen*you_gov': 0.0,
 'you_gov*you_gov': 0.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


{'counts': Counter(),
 'means': defaultdict(<class 'float'>, {}),
 'vars': defaultdict(<class 'float'>, {}),
 'with_std': True}

PolynomialExtender


{'bias_name': 'bias',
 'degree': 2,
 'include_bias': False,
 'interaction_only': False}

RBFSampler


{'gamma': 1.0,
 'n_components': 100,
 'offsets': [5.640140149193941,
             3.115387494021426,
             3.397356353128928,
             5.848583958173694,
             1.2645399074557724,
             3.8992639236407554,
             5.660479707859923,
             4.857060184675885,
             3.8058174892255097,
             4.9980094920420415,
             4.395895881824347,
             2.630883068069719,
             2.60892891862631,
             3.9815351134078383,
             4.912106597254161,
             0.056999659962539556,
             3.3159402900424784,
             2.919859817936021,
             0.6939187579830447,
             2.969260238791456,
             3.573050668467207,
             2.4357103584571425,
             1.9007971908114685,
             1.0125744492278121,
             6.076361757229541,
             5.606962982664035,
             1.6474853309894082,
             3.413360746258298,
             0.3511746525525173,
             5.045878864708639,
             5.831306938626703,
             2.3211008099586183,
             4.02990228918545,
             3.3885653471345165,
             4.659529406277638,
             1.5826092479293281,
             2.1706573941810037,
             4.8866784551804985,
             6.093562588561336,
             4.372996749669299,
             0.42074932878785853,
             5.86892269101574,
             5.617767622148946,
             5.575640678749888,
             1.325444409887302,
             0.03995315915961741,
             4.278165953273511,
             6.204091417395023,
             1.90242754342364,
             2.586518320856655,
             1.0386949218969874,
             1.9735631236540185,
             6.208843737483572,
             5.577674007653164,
             4.142265683571589,
             3.114125801351687,
             2.6352785013457627,
             2.91061674723862,
             2.828235591203184,
             2.8566089734510034,
             5.345065770111299,
             3.2348748557419293,
             0.9004606532198621,
             0.5272855215510872,
             5.828011073962635,
             3.9427714880286273,
             4.929197678090623,
             3.5239232648204117,
             0.9608452025842658,
             4.897773738106849,
             3.513569194700428,
             5.639424120811468,
             2.764443530840953,
             5.789216806014152,
             2.9921840158699182,
             0.4162599546183657,
             4.304222113391852,
             3.7291833568481123,
             3.914348831541803,
             5.517117362621679,
             4.755396341835751,
             4.802191613714404,
             5.9527600663717015,
             5.608689288051799,
             2.2203134407649543,
             6.209379060884266,
             4.608492263178772,
             1.4403323701840198,
             1.0219160962364475,
             1.3989206727140113,
             4.8664577357143255,
             5.490467603242901,
             1.0440021523834675,
             2.2650815610162542,
             6.128404722258214,
             4.908230428554553,
             0.07602375603630986,
             5.053087512732401,
             6.164486507326813,
             5.065640328672257],
 'rng': <random.Random object at 0x5585b88635e0>,
 'seed': None,
 'weights': defaultdict(<bound method RBFSampler._random_weights of RBFSampler (
  gamma=1.
  n_components=100
  seed=None
)>, {})}

LinearRegression


{'_weights': {},
 '_y_name': None,
 'clip_gradient': 1000000000000.0,
 'initializer': Zeros (),
 'intercept': 0.0,
 'intercept_init': 0.0,
 'intercept_lr': Constant({'learning_rate': 0.01}),
 'l1': 0.0,
 'l2': 0.0,
 'loss': Squared({}),
 'optimizer': SGD({'lr': Constant({'learning_rate': 0.01}), 'n_iterations': 0})}

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.