Preprocessing Pipeline

This section explains how to preprocess datasets using the framework's preprocessing pipeline. The preprocessing system provides a modular approach to transform datasets through various operations like subsampling, feedback conversion, core pruning, and data splitting.

All preprocessing operations inherit from the Preprocessor base class, which defines a common interface for processing datasets. Each preprocessor takes a RecSysDataSet as input and returns a transformed dataset, potentially of a different data variant.

Pipe Class

The Pipe class allows you to chain multiple preprocessing steps together into a single preprocessing pipeline:

from omnirec.preprocess.pipe import Pipe
from omnirec.preprocess.subsample import Subsample
from omnirec.preprocess.feedback_conversion import MakeImplicit
from omnirec.preprocess.core_pruning import CorePruning

# Create a preprocessing pipeline
pipeline = Pipe(
    Subsample(0.1),
    MakeImplicit(3),
    CorePruning(5),
)

# Apply all steps sequentially
processed_dataset = pipeline.process(dataset)

The pipeline executes each step in the order they were provided, passing the output of one step as input to the next.

Input Validation

Before transforming a dataset, each preprocessor checks the rules returned by its validation_rules() method. A rule combines a Pandera schema with a failure policy. The default implementation returns no rules; individual preprocessors declare the inputs they require.

The built-in rules are:

Preprocessor	Requirement	On failure
MakeImplicit	A rating column that can be validated as numeric	Raise PreprocessorValidationError
TimeFilter	A timestamp column	Raise PreprocessorValidationError
RatingFilter	A rating column that can be validated as numeric	Log a warning and skip the filter

For example, applying RatingFilter to an implicit dataset without ratings leaves the dataset unchanged. Applying MakeImplicit to that dataset raises an error: it already contains implicit interactions and has no ratings to threshold. Place rating-based steps before MakeImplicit in a pipeline.

Data Processing

Subsample - Reduces the dataset size by sampling a subset of interactions:

from omnirec.preprocess.subsample import Subsample

# Sample 10% of interactions
subsample = Subsample(0.1)
dataset = subsample.process(dataset)

# Sample exactly 1000 interactions
subsample = Subsample(1000)
dataset = subsample.process(dataset)

Parameters:

• sample_size (int | float): Number or fraction of interactions to sample
  • int: Absolute number of interactions
  • float: Fraction of dataset (0.0 to 1.0)

MakeImplicit - Converts explicit feedback to implicit feedback by filtering interactions above a threshold:

from omnirec.preprocess.feedback_conversion import MakeImplicit

# Keep ratings >= 3
make_implicit = MakeImplicit(3)
dataset = make_implicit.process(dataset)

# Keep top 50% of ratings
make_implicit = MakeImplicit(0.5)
dataset = make_implicit.process(dataset)

Parameters:

• threshold (int | float): Threshold for filtering interactions
  • int: Direct rating threshold
  • float: Fraction of maximum rating (0.0 to 1.0)

The result contains user and item, plus timestamp if present in the input. The rating column is removed; each retained row represents a relevant interaction.

CorePruning - Removes users and items with fewer than a specified number of interactions:

from omnirec.preprocess.core_pruning import CorePruning

# Keep only users and items with at least 5 interactions
core_pruning = CorePruning(5)
dataset = core_pruning.process(dataset)

Parameters:

• core (int): Minimum number of interactions required for users and items

Filtering

Apply filtering based on timestamps or ratings to the dataset:

TimeFilter - Filters interactions based on timestamp range:

import pandas as pd
from omnirec.preprocess.filter import TimeFilter

# Keep interactions from January 1998
time_filter = TimeFilter(
    start=pd.Timestamp(year=1998, month=1, day=1),
    end=pd.Timestamp(year=1998, month=1, day=30),
)
dataset = time_filter.process(dataset)

# Keep interactions after a specific date (no end date)
time_filter = TimeFilter(start=pd.Timestamp(year=1998, month=3, day=1))
dataset = time_filter.process(dataset)

Parameters:

• start (Optional[pd.Timestamp]): Start timestamp for the filter (inclusive)
• end (Optional[pd.Timestamp]): End timestamp for the filter (inclusive)

RatingFilter - Filters interactions based on rating values:

from omnirec.preprocess.filter import RatingFilter

# Keep ratings between 1 and 3
rating_filter = RatingFilter(lower=1, upper=3)
dataset = rating_filter.process(dataset)

# Keep ratings >= 4 (no upper bound)
rating_filter = RatingFilter(lower=4)
dataset = rating_filter.process(dataset)

Parameters:

• lower (Optional[int | float]): Lower bound for rating values (inclusive)
• upper (Optional[int | float]): Upper bound for rating values (inclusive)

Data Splitting

Holdout Splits

Create train/validation/test splits:

from omnirec.preprocess.split import UserHoldout, RandomHoldout

# User-aware split (each user appears in all sets)
user_split = UserHoldout(validation_size=0.15, test_size=0.15)
dataset = user_split.process(dataset)

# Random split (no user constraints)
random_split = RandomHoldout(validation_size=0.15, test_size=0.15)
dataset = random_split.process(dataset)

UserHoldout ensures that each user has interactions in all splits, while RandomHoldout randomly assigns interactions without user constraints.

Parameters:

• validation_size (float | int): Size of validation set
  • float: Proportion of dataset (0.0 to 1.0)
  • int: Absolute number of interactions
• test_size (float | int): Size of test set
  • float: Proportion of dataset (0.0 to 1.0)
  • int: Absolute number of interactions

Cross-Validation

Create multiple folds for cross-validation:

from omnirec.preprocess.split import UserCrossValidation, RandomCrossValidation

# User-aware cross-validation (each user appears in all splits)
user_cv = UserCrossValidation(num_folds=5, validation_size=0.2)
dataset = user_cv.process(dataset)

# Random cross-validation (no user constraints)
random_cv = RandomCrossValidation(num_folds=5, validation_size=0.2)
dataset = random_cv.process(dataset)

Cross-Validation creates multiple folds for evaluation, so different interactions are used for validation in each fold. UserCrossValidation ensures that each user has interactions in all splits, while RandomCrossValidation randomly assigns interactions without user constraints.

Parameters:

• num_folds (int): Number of cross-validation folds
• validation_size (float | int): Size of validation set in each fold
  • float: Proportion of training data (0.0 to 1.0)
  • int: Absolute number of interactions

Time-based Splits

Create splits based on timestamps.

TimeBasedHoldout - Splits the dataset chronologically into train, validation, and test sets:

import pandas as pd
from omnirec.preprocess.split import TimeBasedHoldout

# Split using timestamp cutoffs
time_split = TimeBasedHoldout(
    validation=pd.Timestamp(year=1998, month=1, day=3),
    test=pd.Timestamp(year=1998, month=3, day=12),
)
dataset = time_split.process(dataset)

# Split using proportions (newest 15% for test, next 15% for validation)
time_split = TimeBasedHoldout(validation=0.15, test=0.15)
dataset = time_split.process(dataset)

# Split using absolute counts (newest 1000 for test, next 500 for validation)
time_split = TimeBasedHoldout(validation=500, test=1000)
dataset = time_split.process(dataset)

The dataset is sorted by timestamp and split, with older interactions going to training and newer ones to validation/test.

Parameters:

• validation (float | int | pd.Timestamp): Validation set specification
  • float: Proportion of newest interactions (0.0 to 1.0)
  • int: Absolute number of newest interactions
  • pd.Timestamp: Timestamp cutoff (interactions after this go to validation)
• test (float | int | pd.Timestamp): Test set specification (same type as validation)
  • float: Proportion of newest interactions (0.0 to 1.0)
  • int: Absolute number of newest interactions
  • pd.Timestamp: Timestamp cutoff (interactions after this go to test)

Note: Both parameters must be of the same type. The dataset is sorted by timestamp, and interactions are split based on the specified criteria, with older interactions in training and newer ones in validation/test.

Random State

All operations that involve randomness (sampling, splitting) use a consistent random state for reproducibility. See Getting Started > Reproducibility for details how to handle random states.

Data Variant Transformation

The preprocessing operations transform datasets between different data variants:

• Subsample, MakeImplicit, CorePruning, TimeFilter, RatingFilter: RawData → RawData
• UserHoldout, RandomHoldout, TimeBasedHoldout: RawData → SplitData
• UserCrossValidation, RandomCrossValidation: RawData → FoldedData

Custom Preprocessing Steps

You can create custom preprocessing steps by inheriting from the Preprocessor base class and implementing the _process() method (which is called internally by process()).

Implementation

Custom preprocessors must:

1. Inherit from Preprocessor[T, U] where T is the input data variant and U is the output data variant
2. Call super().__init__() in the constructor
3. Implement the _process() method that transforms a RecSysDataSet[T] to RecSysDataSet[U]

Available Data Variants:

• RawData: Single DataFrame with all interactions
• SplitData: Separate train, validation, and test DataFrames
• FoldedData: Multiple folds for cross-validation

Example:

from omnirec.preprocess.base import Preprocessor
from omnirec.recsys_data_set import RawData, RecSysDataSet

class CustomPreprocessor(Preprocessor[RawData, RawData]):
    def __init__(self, param1: int, param2: float) -> None:
        """Your custom preprocessing step.

        Args:
            param1: Description of parameter 1
            param2: Description of parameter 2
        """
        super().__init__()
        self.param1 = param1
        self.param2 = param2

    def _process(self, dataset: RecSysDataSet[RawData]) -> RecSysDataSet[RawData]:
        # Log what you're doing
        self.logger.info(f"Applying custom preprocessing with params: {self.param1}, {self.param2}")

        # Transform the data
        # For RawData: modify dataset._data.df directly
        # For SplitData: access dataset._data.train, dataset._data.val, dataset._data.test
        dataset._data.df = dataset._data.df[...]  # Your transformation logic

        # Return the dataset
        return dataset

Custom preprocessors can be used directly (via process()) or within a Pipe:

from omnirec.preprocess.pipe import Pipe
from omnirec.preprocess.core_pruning import CorePruning

my_preprocessor = CustomPreprocessor(10, 5)
dataset = my_preprocessor.process(dataset)

# Alternative:

pipeline = Pipe(
    CustomPreprocessor(10, 0.5),
    CorePruning(5)
)
dataset = pipeline.process(dataset)

Declaring Validation Rules

Override validation_rules() when a custom step requires particular columns, types, or values. Return a list of ValidationRule objects using pandera.pandas.DataFrameSchema instances or DataFrameModel classes. For example, this step requires non-null, non-negative numeric ratings before applying a square-root transform:

import numpy as np
import pandera.pandas as pa

from omnirec.preprocess.base import Preprocessor
from omnirec.preprocess.validation import ValidationRule
from omnirec.recsys_data_set import RawData, RecSysDataSet


class SqrtRatings(Preprocessor[RawData, RawData]):
    def validation_rules(self) -> list[ValidationRule]:
        return [
            ValidationRule(
                schema=pa.DataFrameSchema(
                    {"rating": pa.Column(float, pa.Check.ge(0), coerce=True)}
                ),
                message="SqrtRatings requires non-negative numeric ratings.",
            )
        ]

    def _process(self, dataset: RecSysDataSet[RawData]) -> RecSysDataSet[RawData]:
        # Convert explicitly: validation does not replace the input DataFrame.
        df = dataset._data.df.copy()
        df["rating"] = np.sqrt(df["rating"].astype(float))
        return dataset.replace_data(RawData(df))


# Use with a raw dataset that contains ratings.
dataset = SqrtRatings().process(dataset)

process() applies each rule in list order to every input DataFrame: raw, all train/validation/test splits, or those splits in every fold. Validation happens before _process(). It checks requirements within the input data variant; it does not convert a raw-data preprocessor into one that supports split or folded data.

Set a rule's on_error argument using ValidationFailureMode:

Mode	Behavior
RAISE (default)	Stop with PreprocessorValidationError before transforming the dataset.
WARN	Log a warning, continue checking rules, and run _process() unless a later check raises or skips. The transformation must handle the invalid input.
SKIP	Log a warning and skip the entire step, even if only one split or fold failed. Return the original dataset without adding a lineage entry. A surrounding Pipe continues with its next step.

For example, use on_error=ValidationFailureMode.SKIP for an optional step, importing the enum from omnirec.preprocess.validation. On failure, messages identify the dataset and input frame; message replaces the generated summary. Pandera's validated or coerced output is not assigned back to the dataset, so perform any required conversions inside _process().

Complete Example

from omnirec import RecSysDataSet
from omnirec.data_loaders.datasets import DataSet
from omnirec.preprocess.pipe import Pipe
from omnirec.preprocess.subsample import Subsample
from omnirec.preprocess.feedback_conversion import MakeImplicit
from omnirec.preprocess.core_pruning import CorePruning
from omnirec.preprocess.split import UserCrossValidation

# Load dataset
dataset = RecSysDataSet.use_dataloader(DataSet.MovieLens100K)

# Create and apply comprehensive preprocessing pipeline
pipeline = Pipe(
    Subsample(0.1),                    # Sample 10% of interactions
    MakeImplicit(3),                   # Convert to implicit (ratings >= 3)
    CorePruning(5),                    # Keep 5-core users and items
    UserCrossValidation(5, 0.1)        # 5-fold CV with 10% validation
)

processed_dataset = pipeline.process(dataset)

# Access the cross-validation folds
for fold_idx, split_data in processed_dataset._data.folds.items():
    print(f"Fold {fold_idx}:")
    print(f"  Train: {len(split_data.train)} interactions")
    print(f"  Validation: {len(split_data.val)} interactions")
    print(f"  Test: {len(split_data.test)} interactions")