final full model + streaming data from disk #33

2025-12-04T00:04:22Z

AndreFerreira5 commented

2025-12-04 00:04:22 +00:00

(Migrated from github.com)

No description provided.

copilot-pull-request-reviewer[bot] (Migrated from github.com) reviewed

2025-12-04 00:10:38 +00:00

copilot-pull-request-reviewer[bot] (Migrated from github.com) left a comment

Pull request overview

This PR introduces a unified MLflow model wrapper for the two-stage prediction system (trend model + residual booster) and refactors the data loading mechanism to stream parquet files directly from disk using PyArrow instead of loading entire files into memory.

Introduces TrendResidualModel class wrapping both Ridge and LightGBM models as a single MLflow pyfunc model
Refactors DataLoader to use PyArrow's iter_batches for memory-efficient streaming from parquet files
Adds predicting flag to preprocessor to handle differences between training and inference data availability

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 12 comments.

File	Description
src/training_api/train.py	Adds TrendResidualModel wrapper class and updates model registration to use mlflow.pyfunc; refactors parameter logging order and adds artifact cleanup
src/training_api/main.py	Adds pandas import and removes unused test_predictor import; removes trailing newline
src/training_api/data/processer.py	Adds 'predicting' parameter to handle inference-time data differences (excludes dropoff_datetime when predicting)
src/training_api/data/loader.py	Replaces in-memory file loading with PyArrow streaming using iter_batches; simplifies state tracking and removes del method

Comments suppressed due to low confidence (2)

src/training_api/main.py:19

Call to function run_hyperparameter_tuning with too few arguments; should be no fewer than 2.

    model_params = run_hyperparameter_tuning()

src/training_api/main.py:3

Import of 'pd' is not used.

import pandas as pd

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

## Pull request overview This PR introduces a unified MLflow model wrapper for the two-stage prediction system (trend model + residual booster) and refactors the data loading mechanism to stream parquet files directly from disk using PyArrow instead of loading entire files into memory. - Introduces `TrendResidualModel` class wrapping both Ridge and LightGBM models as a single MLflow pyfunc model - Refactors `DataLoader` to use PyArrow's `iter_batches` for memory-efficient streaming from parquet files - Adds `predicting` flag to preprocessor to handle differences between training and inference data availability ### Reviewed changes Copilot reviewed 4 out of 4 changed files in this pull request and generated 12 comments. | File | Description | | ---- | ----------- | | src/training_api/train.py | Adds TrendResidualModel wrapper class and updates model registration to use mlflow.pyfunc; refactors parameter logging order and adds artifact cleanup | | src/training_api/main.py | Adds pandas import and removes unused test_predictor import; removes trailing newline | | src/training_api/data/processer.py | Adds 'predicting' parameter to handle inference-time data differences (excludes dropoff_datetime when predicting) | | src/training_api/data/loader.py | Replaces in-memory file loading with PyArrow streaming using iter_batches; simplifies state tracking and removes __del__ method | <details> <summary>Comments suppressed due to low confidence (2)</summary> **src/training_api/main.py:19** * Call to [function run_hyperparameter_tuning](1) with too few arguments; should be no fewer than 2. ``` model_params = run_hyperparameter_tuning() ``` **src/training_api/main.py:3** * Import of 'pd' is not used. ``` import pandas as pd ``` </details> --- 💡 <a href="/AndreFerreira5/downtown-cab-co/new/main/.github/instructions?filename=*.instructions.md" class="Link--inTextBlock" target="_blank" rel="noopener noreferrer">Add Copilot custom instructions</a> for smarter, more guided reviews. <a href="https://docs.github.com/en/copilot/customizing-copilot/adding-repository-custom-instructions-for-github-copilot" class="Link--inTextBlock" target="_blank" rel="noopener noreferrer">Learn how to get started</a>.

src/training_api/data/loader.py

Missing import: pq (pyarrow.parquet) is used here but not imported. Add import pyarrow.parquet as pq at the top of the file to fix the NameError.

Missing import: `pq` (pyarrow.parquet) is used here but not imported. Add `import pyarrow.parquet as pq` at the top of the file to fix the `NameError`.

src/training_api/data/loader.py

Missing docstring: The reset method no longer has a docstring after the refactoring. Consider adding one to explain its purpose, e.g., "Reset the loader to start from the beginning and clear memory."

    def reset(self) -> None:
        """
        Reset the loader to start from the beginning and clear memory.
        """

Missing docstring: The `reset` method no longer has a docstring after the refactoring. Consider adding one to explain its purpose, e.g., "Reset the loader to start from the beginning and clear memory." ```suggestion def reset(self) -> None: """ Reset the loader to start from the beginning and clear memory. """ ```

src/training_api/data/loader.py

Missing docstring: The get_file_count method no longer has a docstring after the refactoring. Consider adding one to explain its purpose, e.g., "Return the number of parquet files."

    def get_file_count(self) -> int:
        """
        Return the number of parquet files.
        """

Missing docstring: The `get_file_count` method no longer has a docstring after the refactoring. Consider adding one to explain its purpose, e.g., "Return the number of parquet files." ```suggestion def get_file_count(self) -> int: """ Return the number of parquet files. """ ```

src/training_api/main.py

					
				@ -1,6 +1,8 @@

				import logging

				import pandas as pd

Unused import: pandas as pd is imported but not used in this file. Consider removing it to keep imports clean.

Unused import: `pandas as pd` is imported but not used in this file. Consider removing it to keep imports clean. ```suggestion ```

src/training_api/train.py

					
				@ -20,0 +21,4 @@

				        with open(context.artifacts["regressor"], "rb") as f:

				            self.trend_model = pickle.load(f)

				        with open(context.artifacts["booster"], "rb") as f:

				            self.booster_model = pickle.load(f)

Inconsistent attribute naming: The loaded model is stored as self.trend_model but later referenced as self.model_a in line 30. This will cause an AttributeError at runtime. Change this to self.trend_model to match the usage in the predict method.

Inconsistent attribute naming: The loaded model is stored as `self.trend_model` but later referenced as `self.model_a` in line 30. This will cause an `AttributeError` at runtime. Change this to `self.trend_model` to match the usage in the predict method.

src/training_api/train.py

					
				@ -20,0 +23,4 @@

				        with open(context.artifacts["booster"], "rb") as f:

				            self.booster_model = pickle.load(f)

				    def predict(self, context, model_input):

Inconsistent attribute naming: The loaded model is stored as self.booster_model but later referenced as self.model_b in line 34. This will cause an AttributeError at runtime. Change this to self.booster_model to match the usage in the predict method.

Inconsistent attribute naming: The loaded model is stored as `self.booster_model` but later referenced as `self.model_b` in line 34. This will cause an `AttributeError` at runtime. Change this to `self.booster_model` to match the usage in the predict method.

src/training_api/train.py

					
				@ -20,0 +27,4 @@

				        model_input["tpep_pickup_datetime"] = pd.to_datetime(model_input["tpep_pickup_datetime"])

				        (X, t) = preprocess_taxi_data(pd.DataFrame([model_input]), remove_outliers=False, create_features=True, predicting=True)

				        trend_pred = self.model_a.predict(X[['date_int', 'sin_time', 'cos_time']])

Incorrect unpacking: preprocess_taxi_data returns a tuple of (pd.DataFrame, TaxiDataPreprocessor), not (X, t). The second element is the preprocessor object, not a target variable. This line should be: X, _ = preprocess_taxi_data(...) or X = preprocess_taxi_data(...)[0].

        X, _ = preprocess_taxi_data(pd.DataFrame([model_input]), remove_outliers=False, create_features=True, predicting=True)

Incorrect unpacking: `preprocess_taxi_data` returns a tuple of `(pd.DataFrame, TaxiDataPreprocessor)`, not `(X, t)`. The second element is the preprocessor object, not a target variable. This line should be: `X, _ = preprocess_taxi_data(...)` or `X = preprocess_taxi_data(...)[0]`. ```suggestion X, _ = preprocess_taxi_data(pd.DataFrame([model_input]), remove_outliers=False, create_features=True, predicting=True) ```

src/training_api/train.py

					
				@ -20,0 +29,4 @@

				        trend_pred = self.model_a.predict(X[['date_int', 'sin_time', 'cos_time']])

				        trend_pred = np.maximum(trend_pred, 1.0)

Incorrect attribute reference: Using self.model_a but the attribute is defined as self.trend_model in line 22. Change to self.trend_model to match the load_context method.

        trend_pred = self.trend_model.predict(X[['date_int', 'sin_time', 'cos_time']])

Incorrect attribute reference: Using `self.model_a` but the attribute is defined as `self.trend_model` in line 22. Change to `self.trend_model` to match the load_context method. ```suggestion trend_pred = self.trend_model.predict(X[['date_int', 'sin_time', 'cos_time']]) ```

src/training_api/train.py

					
				@ -20,0 +32,4 @@

				        X.drop(columns=['trip_duration', 'tpep_pickup_datetime', 'date_int', 'sin_time', 'cos_time'], errors='ignore')

				        ratio_pred = self.model_b.predict(X)

				        final_pred = trend_pred * ratio_pred

Missing assignment: The drop method returns a new DataFrame and doesn't modify in place. The result is not being assigned to X, so the columns are not actually removed. Change to: X = X.drop(columns=[...], errors='ignore').

        X = X.drop(columns=['trip_duration', 'tpep_pickup_datetime', 'date_int', 'sin_time', 'cos_time'], errors='ignore')

Missing assignment: The `drop` method returns a new DataFrame and doesn't modify in place. The result is not being assigned to `X`, so the columns are not actually removed. Change to: `X = X.drop(columns=[...], errors='ignore')`. ```suggestion X = X.drop(columns=['trip_duration', 'tpep_pickup_datetime', 'date_int', 'sin_time', 'cos_time'], errors='ignore') ```

src/training_api/train.py

					
				@ -20,0 +33,4 @@

				        X.drop(columns=['trip_duration', 'tpep_pickup_datetime', 'date_int', 'sin_time', 'cos_time'], errors='ignore')

				        ratio_pred = self.model_b.predict(X)

				        final_pred = trend_pred * ratio_pred

Incorrect attribute reference: Using self.model_b but the attribute is defined as self.booster_model in line 24. Change to self.booster_model to match the load_context method.

        trend_pred = self.trend_model.predict(X[['date_int', 'sin_time', 'cos_time']])
        trend_pred = np.maximum(trend_pred, 1.0)

        X.drop(columns=['trip_duration', 'tpep_pickup_datetime', 'date_int', 'sin_time', 'cos_time'], errors='ignore')
        ratio_pred = self.booster_model.predict(X)

Incorrect attribute reference: Using `self.model_b` but the attribute is defined as `self.booster_model` in line 24. Change to `self.booster_model` to match the load_context method. ```suggestion trend_pred = self.trend_model.predict(X[['date_int', 'sin_time', 'cos_time']]) trend_pred = np.maximum(trend_pred, 1.0) X.drop(columns=['trip_duration', 'tpep_pickup_datetime', 'date_int', 'sin_time', 'cos_time'], errors='ignore') ratio_pred = self.booster_model.predict(X) ```

src/training_api/train.py

Missing cleanup: The pickle files created at regressor_path and booster_path are not cleaned up after logging the model to MLflow (unlike in run_training where they are removed on lines 438-439). This creates temporary files that persist in the filesystem during hyperparameter tuning. Consider adding cleanup after line 378:

if os.path.exists(regressor_path): os.remove(regressor_path)
if os.path.exists(booster_path): os.remove(booster_path)

        )
        if os.path.exists(regressor_path): os.remove(regressor_path)
        if os.path.exists(booster_path): os.remove(booster_path)

Missing cleanup: The pickle files created at `regressor_path` and `booster_path` are not cleaned up after logging the model to MLflow (unlike in `run_training` where they are removed on lines 438-439). This creates temporary files that persist in the filesystem during hyperparameter tuning. Consider adding cleanup after line 378: ```python if os.path.exists(regressor_path): os.remove(regressor_path) if os.path.exists(booster_path): os.remove(booster_path) ``` ```suggestion ) if os.path.exists(regressor_path): os.remove(regressor_path) if os.path.exists(booster_path): os.remove(booster_path) ```

src/training_api/train.py

Multiple statements on one line: Consider splitting this into separate lines for better readability:

if os.path.exists(regressor_path): 
    os.remove(regressor_path)
if os.path.exists(booster_path): 
    os.remove(booster_path)

    if os.path.exists(regressor_path):
        os.remove(regressor_path)
    if os.path.exists(booster_path):
        os.remove(booster_path)

Multiple statements on one line: Consider splitting this into separate lines for better readability: ```python if os.path.exists(regressor_path): os.remove(regressor_path) if os.path.exists(booster_path): os.remove(booster_path) ``` ```suggestion if os.path.exists(regressor_path): os.remove(regressor_path) if os.path.exists(booster_path): os.remove(booster_path) ```

Sign in to join this conversation.

No reviewers