22

How do you call partial_fit() on a scikit-learn classifier wrapped inside a Pipeline()?

I'm trying to build an incrementally trainable text classifier using SGDClassifier like:

from sklearn.linear_model import SGDClassifier
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.feature_extraction.text import TfidfTransformer
from sklearn.multiclass import OneVsRestClassifier

classifier = Pipeline([
    ('vectorizer', HashingVectorizer(ngram_range=(1,4), non_negative=True)),
    ('tfidf', TfidfTransformer()),
    ('clf', OneVsRestClassifier(SGDClassifier())),
])

but I get an AttributeError trying to call classifier.partial_fit(x,y).

It supports fit(), so I don't see why partial_fit() isn't available. Would it be possible to introspect the pipeline, call the data transformers, and then directly call partial_fit() on my classifier?

1
  • 1
    Did you eventually come up with a solution for this?
    – GreenGodot
    Apr 6, 2016 at 13:44

4 Answers 4

17

Here is what I'm doing - where 'mapper' and 'clf' are the 2 steps in my Pipeline obj.

def partial_pipe_fit(pipeline_obj, df):
    X = pipeline_obj.named_steps['mapper'].fit_transform(df)
    Y = df['class']
    pipeline_obj.named_steps['clf'].partial_fit(X,Y)

You probably want to keep track of performance as you keep adjusting/updating your classifier - but that is a secondary point

so more specifically - the original pipeline(s) were constructed as follows

to_vect = Pipeline([('vect', CountVectorizer(min_df=2, max_df=.9, ngram_range=(1, 1), max_features = 100)),
                            ('tfidf', TfidfTransformer())])
full_mapper = DataFrameMapper([
            ('norm_text', to_vect),
            ('norm_fname', to_vect), ])

full_pipe = Pipeline([('mapper', full_mapper), ('clf', SGDClassifier(n_iter=15, warm_start=True,
                                                                n_jobs=-1, random_state=self.random_state))])

google DataFrameMapper to learn more about it - but here it just enables a transformation step that plays nice with pandas

9

Pipeline does not use partial_fit, hence does not expose it. We would probably need a dedicated pipelining scheme for out-of-core computation but that also depends on the capabilities of the previous models.

In particular in this case you would probably want to do several passes over your data, one to fit each stage of the pipeline and then to transform the dataset to fit the next one, except for the first stage which is stateless, hence does not fit parameters from the data.

In the mean time it's probably easier to roll your own wrapper code tailored to your needs.

3
  • 1
    Can you recommend how I might roll my own? I tried using the pipeline's transform() method, and then extracting the classifier and feeding the transformed data to it's partial_fit(), but I get an error about the tdf vector being undefined.
    – Cerin
    Jul 30, 2013 at 14:58
  • 3
    Read the source code of the Pipeline class and this example. Then read the documentation for text feature extraction and the hashing trick to make sure you fully understand the issue with stateful feature extraction. The implementation will depend on what problem you are trying to solve.
    – ogrisel
    Jul 30, 2013 at 15:51
  • 1
    In particular if you use stateful transformers as TfidfTransformer you will need to do several passes on your data.
    – ogrisel
    Jul 30, 2013 at 15:51
2

Even though this question is 8 years old it is still very relevant and has not been updated for quite a time now.

As the matter of fact, there is now a nice package created by Vincent Warmerdam called tokenwiser.

It is used for NLP stuff mostly to fit within the sklearn infrastructure. However, there is the main building block that can be used even for not NLP tasks.

The package has PartialPipeline boiler plate and Documentantation.

Example here:


import numpy as np
from sklearn.linear_model import SGDClassifier
from sklearn.feature_extraction.text import HashingVectorizer

from tokenwiser.textprep import Cleaner, Identity, HyphenTextPrep
from tokenwiser.pipeline import PartialPipeline, PartialFeatureUnion

pipe = PartialPipeline([
    ("clean", Cleaner()),
    ("union", PartialFeatureUnion([
        ("full_text_pipe", PartialPipeline([
            ("identity", Identity()),
            ("hash1", HashingVectorizer()),
        ])),
        ("hyphen_pipe", PartialPipeline([
            ("hyphen", HyphenTextPrep()),
            ("hash2", HashingVectorizer()),
        ]))
    ])),
    ("clf", SGDClassifier())
])

X = [
    "i really like this post",
    "thanks for that comment",
    "i enjoy this friendly forum",
    "this is a bad post",
    "i dislike this article",
    "this is not well written"
]

y = np.array([1, 1, 1, 0, 0, 0])

for loop in range(3):
    pipe.partial_fit(X, y, classes=[0, 1])

I can imagine this template working even for non-NLP-related stuff. Hope someone will find this super usefull.

1

I also propose my basic implementation of utilizing partial_fit within a sklearn pipeline.

We just need to use a model that allows for partial fit (e.g. SGDregressor, xgboost, etc) and create own sklearn compatible classes

(Huge KUDOS to VIncent Warmerdam who started this in his TOKENWISER project)


import xgboost as xgb
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklego.preprocessing import PatsyTransformer

class xgboost_partial_trainer(BaseEstimator, TransformerMixin):
    """
    allows for incremental training od xgboost model within a sklean pipeline
    """

    def __init__(self, training_params: dict = None):

        self.training_params = training_params
        self.trained_model = None
        self._first_call = True
        self.evals_result = {}
        self.iter_number = 1
        self._X_train, self._X_test, self._y_train, self._y_test = (
            None,
            None,
            None,
            None,
        )

    def partial_fit(self, X, y=None, classes=None, **fit_params):

        print(f"firts run: {self._first_call}, n_iter = {self.iter_number}")
        self.iter_number += 1

        if self._first_call:

            # Select random subset of data and store within the model (for error loss over time)
            self._X_train, self._X_test, self._y_train, self._y_test = train_test_split(
                X, y, test_size=0.6, random_state=1
            )

            self._xg_train = xgb.DMatrix(self._X_train, label=self._y_train)
            self._xg_test = xgb.DMatrix(self._X_test, label=self._y_test)

            # validations set to watch performance - same testing data, changebla training data
            self.watchlist = [
                (self._xg_train, "train_batch"),
                (self._xg_test, "eval_fixed"),
            ]

            # Trainig Part Itself
            self.trained_model = xgb.train(
                params=self.training_params,
                dtrain=xgb.DMatrix(X, y),
                xgb_model=self.trained_model,
                evals=self.watchlist,
            )

            # Swich outside firts batch
            self._first_call = False

        else:
            self._xg_train = xgb.DMatrix(X, y)
            self.watchlist = [
                (self._xg_train, "train_batch"),
                (self._xg_test, "eval_fixed"),
            ]

            self.trained_model = xgb.train(
                params=self.training_params,
                dtrain=self._xg_train,
                xgb_model=self.trained_model,
                evals=self.watchlist,
            )
        #             self._predicted_y = self.trained_model.predict(xgb.DMatrix(self._X_test))
        #             print(f"mean_squared_error = {mean_squared_error(self._y_test, self._predicted_y, squared = False)}")

        return self

    def predict(self, X, y=None, **fit_params):
        return self.trained_model.predict(xgb.DMatrix(X))

    def transform(self, X, y=None, **fit_params):
        return self.trained_model.predict(xgb.DMatrix(X))

    def fit(self, X, y=None, **fit_params):
        return self


class PartialPipeline(Pipeline):
    """
    Utility function to generate a `PartialPipeline`

    Arguments:
        steps: a collection of text-transformers
    """

    def partial_fit(self, X, y=None, classes=None, **kwargs):
        """
        Fits the components, but allow for batches.
        """

        #         print(f"there are partial steps {self.steps_partial}")

        for _, step in self.steps:
            if hasattr(step, "partial_fit"):
                step.partial_fit(X, y, **kwargs)

            elif hasattr(step, "fit_transform"):
                X = step.fit_transform(X)

            elif hasattr(step, "transform"):
                X = step.transform(X)

            elif hasattr(step, "fit"):
                X = step.fit(X)

        return self

Once we have these sklearn classes we may utilize the Pipeline:

my_pipeline = PartialPipeline([
("patsy", PatsyTransformer(FORMULA2)),
("xgboost_model", xgboost_partial_trainer(training_params=params)),
])

df_chunked = pd.read_csv(your_date, chunksize=5_000)

for df in df_chunked:
    my_pipeline.partial_fit(df, y=df["speed"])
  

Please, provide me with feedback and code cleaning suggestions. I am fully aware that this is not perfect. However, as a nice prototype - not too bad!

Your Answer

Reminder: Answers generated by Artificial Intelligence tools are not allowed on Stack Overflow. Learn more

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.