Modify the training pipeline to load every CSV file in a folder

[svpino]

The current pipeline loads the data from a single file. Modify the code to point the pipeline to a folder instead. The pipeline should load every file from the folder and combine them to create the dataset.

[traycerai]

Plan

Observations

Currently, DatasetMixin uses an IncludeFile to embed one CSV as text and calls pd.read_csv(StringIO(self.dataset)), so only a single file is supported. The training flow (pipelines/training.py) and monitoring flow rely on this mixin. Tests (test_common_flowmixin.py, test_training*.py) expect default behavior but don’t handle folders. We need to switch to reading from a user‑provided filesystem path (file or directory) while preserving the existing sex cleanup and shuffle steps.

Approach

In pipelines/common.py, replace the IncludeFile dataset artifact with a Metaflow Parameter named dataset_path. Update load_dataset to detect whether dataset_path is a file or directory, list all *.csv files in a folder, read each with pandas into DataFrames, then concatenate them into one. Retain the current sex value cleaning and random shuffle. Maintain backward compatibility by defaulting dataset_path to the original CSV file. Update docs (README.md, .guide/training-pipeline/dataset.md) to show how to use --dataset-path. Add new tests to verify multiple‑file loading in both the mixin and the full training pipeline.

Reasoning

I reviewed pipelines/training.py and saw self.load_dataset() come from common.DatasetMixin. In pipelines/common.py, I found IncludeFile and saw its single‑file logic. I inspected existing tests and guides to confirm how the dataset is used and tested. This revealed the need to convert to filesystem‑based ingestion and update downstream code, tests, and documentation accordingly.

Proposed File Changes

📄 pipelines/common.py ^(MODIFY) 🔗

Referred Files

• 📄 pipelines/training.py

In pipelines/common.py, remove the IncludeFile import and its dataset definition. Add a Parameter("dataset-path", default="data/penguins.csv", help="Path to a CSV file or folder of CSV files."). In load_dataset, replace reading from StringIO(self.dataset) with logic to:

1. Create a Path(self.dataset_path).
2. If it’s a directory, list all files matching *.csv. If it’s a file, treat it as a single‑element list.
3. Raise a clear error if no CSV files are found.
4. Loop over each path, read with pd.read_csv, collect DataFrames.
5. Concatenate with pd.concat(dfs, ignore_index=True).
6. Apply existing sex‑column cleaning and shuffle with the same seed logic.

Refer to pipelines/training.py for how load_dataset is consumed.

📄 README.md ^(MODIFY) 🔗

Referred Files

• 📄 README.md ^(MODIFY)

Update the usage section to document the new --dataset-path parameter. Explain that it can point to a single CSV file or a folder of CSVs. Show a full example, e.g.:

python pipelines/training.py \
  --environment conda --production run \
  --dataset-path path/to/csv_folder

📄 .guide/training-pipeline/dataset.md ^(MODIFY) 🔗

Referred Files

• 📄 pipelines/common.py ^(MODIFY)

Revise the documentation to describe --dataset-path instead of embedding a file. Show how to pass a directory of CSVs, note that all *.csv files are read and concatenated, and update the example command accordingly.

📄 tests/test_common_load_multiple_csvs.py ^(NEW) 🔗

Referred Files

• 📄 pipelines/common.py ^(MODIFY)
• 📄 tests/flowmixin_flow.py

Create a new test that:

1. Uses tmp_path or tempfile to build a directory containing two small CSV files with known rows.
2. Defines a minimal Metaflow flow inheriting from DatasetMixin and calling self.data = self.load_dataset().
3. Runs the flow with --dataset-path set to that directory.
4. Asserts the returned DataFrame has a row count equal to the sum of both CSV files’ rows.

📄 tests/test_training_pipeline_folder.py ^(NEW) 🔗

Referred Files

• 📄 pipelines/training.py
• 📄 tests/conftest.py

Add an integration test for pipelines/training.py that:

1. Prepares a temp directory with multiple CSVs matching the penguins schema.
2. Invokes the pipeline via Runner("pipelines/training.py").run(...), passing mlflow_tracking_uri, training_epochs, accuracy_threshold, and --dataset-path.
3. Verifies in the start step that task.data.data exists, has the species column, and rows equal the combined total.

Diagram

sequenceDiagram
    participant Flow as Training Flow
    participant Mixin as DatasetMixin
    participant FS as FileSystem
    participant Pandas as pandas

    Flow->>Mixin: load_dataset()
    Mixin->>FS: check dataset_path
    alt directory
        Mixin->>FS: list *.csv files
    else single file
        Mixin->>FS: single CSV
    end
    loop each file
        Mixin->>Pandas: read_csv(file)
        Pandas-->>Mixin: DataFrame
    end
    Mixin->>Pandas: concat(DataFrames)
    Pandas-->>Mixin: combined DataFrame
    Mixin-->>Flow: return data

Loading

Import In IDE

Developer Humor

"We ventured into the CSV jungle 🏞️"
"Gathered every file with a roar 🦁"
"Concatenated them all in one great scroll 📜"
"Our pipeline now stands tall and grand 💪"
"Model training, here we come! 🚀"

---

Execution Information

Branch: ai
Commit: c1c3cdb

---

💡 Tips

Supported Commands (Inside Comments)

• Use @traycerai generate to iterate on the previous version of the implementation plan.

Supported Commands (Inside Description)

• Add @traycerai ignore anywhere in the ticket description to prevent this ticket from being processed.
• Add @traycerai branch:<branch-name> anywhere in the ticket description to specify the target branch for the implementation plan.

Community

• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.