Skip to content

dqm_ml_job.dataloaders.pandas

Pandas data loader for reading CSV files.

This module contains the PandasDataLoader and PandasDataSelection classes for loading and iterating over CSV file data using Pandas.

logger = logging.getLogger(__name__) module-attribute

PandasDataLoader

Data loader for CSV files using Pandas.

This loader reads CSV files and provides DataSelections for processing by the DQM pipeline.

Attributes:

Name Type Description
type str

The loader type identifier ("csv").
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
class PandasDataLoader:
    """Data loader for CSV files using Pandas.

    This loader reads CSV files and provides DataSelections for
    processing by the DQM pipeline.

    Attributes:
        type: The loader type identifier ("csv").
    """

    type: str = "csv"

    def __init__(self, name: str, config: dict[str, Any] | None = None):
        """Initialize the Pandas data loader.

        Args:
            name: Unique name for this loader instance.
            config: Configuration dictionary containing:
                - path: Path to CSV file (required)

        Raises:
            ValueError: If required config keys are missing.
        """
        if config is None:
            config = {}
        self.name = name
        self.path = config["path"]
        # Use SplitConfig model fields instead of hardcoded keys
        from dqm_ml_core.models.dataloaders import SplitConfig

        split = config.get("split")
        self.split = SplitConfig.model_validate(split) if split else None
        self.split_by = self.split.by if self.split else None
        self.split_values = self.split.values if self.split else None
        filters = config.get("filters")
        # transform the list of dict into a dict
        self.filters_dict = {}
        if filters is not None:
            for item in filters:
                column = item["column"]
                self.filters_dict[column] = item["values"]
        self.id_column = config.get("id_column")
        self.sample_path = config.get("sample_path", [])
        self.transforms = config.get("transform", [])

        # Storage filesystem configuration - only for S3 paths, not local paths
        self.filesystem = None
        storage_cfg = config.get("storage")
        if storage_cfg:
            # Use StorageConfig model to validate and access fields
            from dqm_ml_core.models.global_ import StorageConfig

            storage_config = StorageConfig.model_validate(storage_cfg)

            if storage_config.type == "s3":
                from dqm_ml_job.utils.s3 import get_s3_filesystem

                self.filesystem = get_s3_filesystem(storage_config)

    def get_selections(self) -> list[DataSelection]:
        """Create one or more PandasDataSelection instances based on split config.

        If split is configured, returns one selection per split value.
        Otherwise returns a single selection for the entire CSV file.

        Returns:
            A list of DataSelection instances.
        """
        if not self.split_by:
            return [
                PandasDataSelection(
                    name=self.name,
                    path=self.path,
                    sample_path=self.sample_path,
                    transforms=self.transforms,
                    filters_dict=self.filters_dict,
                )
            ]

        # Determine split values
        values = self.split_values
        if values is None:
            # Auto-discover unique values from the CSV
            df = pd.read_csv(self.path, sep=",", usecols=[self.split_by])
            values = [str(v) for v in df[self.split_by].unique() if v is not None]
        else:
            # Expand wildcard patterns in values against available data
            if any(has_pattern(v) for v in values):
                df = pd.read_csv(self.path, sep=",", usecols=[self.split_by])
                available = [str(v) for v in df[self.split_by].unique() if v is not None]
                values = resolve_patterns(values, available)

        # Apply split.exclude (including wildcard patterns)
        if self.split and self.split.exclude:
            values = resolve_include_exclude(None, self.split.exclude, values)

        # Create one selection per value
        selections: list[DataSelection] = []
        for val in values:
            selection_name = f"{self.name}_{val}"
            merged_filters = (self.filters_dict or {}).copy()
            merged_filters[self.split_by] = val
            selections.append(
                PandasDataSelection(
                    name=selection_name,
                    path=self.path,
                    sample_path=self.sample_path,
                    transforms=self.transforms,
                    filters_dict=merged_filters,
                )
            )
        return selections

filesystem = None instance-attribute

filters_dict = {} instance-attribute

id_column = config.get('id_column') instance-attribute

name = name instance-attribute

path = config['path'] instance-attribute

sample_path = config.get('sample_path', []) instance-attribute

split = SplitConfig.model_validate(split) if split else None instance-attribute

split_by = self.split.by if self.split else None instance-attribute

split_values = self.split.values if self.split else None instance-attribute

transforms = config.get('transform', []) instance-attribute

type: str = 'csv' class-attribute instance-attribute

__init__(name: str, config: dict[str, Any] | None = None)

Initialize the Pandas data loader.

Parameters:

Name Type Description Default
name str

Unique name for this loader instance.

 required
config dict[str, Any] | None

Configuration dictionary containing: - path: Path to CSV file (required)

 None

Raises:

Type Description
ValueError

If required config keys are missing.
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def __init__(self, name: str, config: dict[str, Any] | None = None):
    """Initialize the Pandas data loader.

    Args:
        name: Unique name for this loader instance.
        config: Configuration dictionary containing:
            - path: Path to CSV file (required)

    Raises:
        ValueError: If required config keys are missing.
    """
    if config is None:
        config = {}
    self.name = name
    self.path = config["path"]
    # Use SplitConfig model fields instead of hardcoded keys
    from dqm_ml_core.models.dataloaders import SplitConfig

    split = config.get("split")
    self.split = SplitConfig.model_validate(split) if split else None
    self.split_by = self.split.by if self.split else None
    self.split_values = self.split.values if self.split else None
    filters = config.get("filters")
    # transform the list of dict into a dict
    self.filters_dict = {}
    if filters is not None:
        for item in filters:
            column = item["column"]
            self.filters_dict[column] = item["values"]
    self.id_column = config.get("id_column")
    self.sample_path = config.get("sample_path", [])
    self.transforms = config.get("transform", [])

    # Storage filesystem configuration - only for S3 paths, not local paths
    self.filesystem = None
    storage_cfg = config.get("storage")
    if storage_cfg:
        # Use StorageConfig model to validate and access fields
        from dqm_ml_core.models.global_ import StorageConfig

        storage_config = StorageConfig.model_validate(storage_cfg)

        if storage_config.type == "s3":
            from dqm_ml_job.utils.s3 import get_s3_filesystem

            self.filesystem = get_s3_filesystem(storage_config)

get_selections() -> list[DataSelection]

Create one or more PandasDataSelection instances based on split config.

If split is configured, returns one selection per split value. Otherwise returns a single selection for the entire CSV file.

Returns:

Type Description
list[DataSelection]

A list of DataSelection instances.
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def get_selections(self) -> list[DataSelection]:
    """Create one or more PandasDataSelection instances based on split config.

    If split is configured, returns one selection per split value.
    Otherwise returns a single selection for the entire CSV file.

    Returns:
        A list of DataSelection instances.
    """
    if not self.split_by:
        return [
            PandasDataSelection(
                name=self.name,
                path=self.path,
                sample_path=self.sample_path,
                transforms=self.transforms,
                filters_dict=self.filters_dict,
            )
        ]

    # Determine split values
    values = self.split_values
    if values is None:
        # Auto-discover unique values from the CSV
        df = pd.read_csv(self.path, sep=",", usecols=[self.split_by])
        values = [str(v) for v in df[self.split_by].unique() if v is not None]
    else:
        # Expand wildcard patterns in values against available data
        if any(has_pattern(v) for v in values):
            df = pd.read_csv(self.path, sep=",", usecols=[self.split_by])
            available = [str(v) for v in df[self.split_by].unique() if v is not None]
            values = resolve_patterns(values, available)

    # Apply split.exclude (including wildcard patterns)
    if self.split and self.split.exclude:
        values = resolve_include_exclude(None, self.split.exclude, values)

    # Create one selection per value
    selections: list[DataSelection] = []
    for val in values:
        selection_name = f"{self.name}_{val}"
        merged_filters = (self.filters_dict or {}).copy()
        merged_filters[self.split_by] = val
        selections.append(
            PandasDataSelection(
                name=selection_name,
                path=self.path,
                sample_path=self.sample_path,
                transforms=self.transforms,
                filters_dict=merged_filters,
            )
        )
    return selections

PandasDataSelection

Bases: DataSelection

A selection of data from a CSV file loaded via Pandas.

This class represents data loaded from a CSV file and provides an iterator over PyArrow RecordBatches.

Attributes:

Name Type Description
name

Name identifier for this selection.
path

Path to the CSV file.
data DataFrame | None

The loaded pandas DataFrame.
sample_path

List of sample path configs describing column path prefixes.
transforms

List of transform configs (column cast operations).
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class PandasDataSelection(DataSelection):
    """A selection of data from a CSV file loaded via Pandas.

    This class represents data loaded from a CSV file and provides
    an iterator over PyArrow RecordBatches.

    Attributes:
        name: Name identifier for this selection.
        path: Path to the CSV file.
        data: The loaded pandas DataFrame.
        sample_path: List of sample path configs describing column path prefixes.
        transforms: List of transform configs (column cast operations).
    """

    def __init__(
        self,
        name: str,
        path: str,
        sample_path: list[dict[str, Any]] | None = None,
        transforms: list[dict[str, Any]] | None = None,
        filters_dict: dict[str, Any] | None = None,
    ):
        """Initialize a Pandas data selection.

        Args:
            name: Name identifier for this selection.
            path: Path to the CSV file.
            sample_path: List of sample path configs describing column path prefixes.
            transforms: List of transform configs (column cast operations).
            filters_dict: Column-value pairs to filter rows by.
        """
        self.name = name
        self.path = path
        self.sample_path = sample_path or []
        self.transforms = transforms or []
        self.filters_dict = filters_dict or {}
        self.data: pd.DataFrame | None = None

    @override
    def bootstrap(self, columns_list: list[str] | None = None) -> None:
        """Load the CSV file into memory as a pandas DataFrame.

        Args:
            columns_list: Unused, kept for API compatibility.
        """
        from dqm_ml_job.dataloaders.filters import build_filter_condition

        data = pd.read_csv(self.path, sep=",")
        assert isinstance(data, pd.DataFrame)
        self.data = data
        for col, val in self.filters_dict.items():
            condition = build_filter_condition(
                col,
                val,
                wildcard_fn=lambda c, vals: _match_wildcard(data[c], vals),
                isin_fn=lambda c, vals: data[c].isin(vals),
                equal_fn=lambda c, v: data[c] == v,
            )
            self.data = self.data[condition.reindex(self.data.index, fill_value=False)]

    def __len__(self) -> int:
        return len(self.data) if self.data is not None else 0

    @override
    def get_nb_batches(self) -> int:
        """Return the estimated number of batches (always 1 for CSV).

        Returns:
            1 if data is loaded, 0 otherwise.
        """
        return 1 if self.data is not None else 0

    @override
    def __iter__(self) -> Any:
        if self.data is not None:
            df = self.data.copy() if self.transforms else self.data
            _apply_pandas_transforms(df, self.transforms)
            yield pa.RecordBatch.from_pandas(df)

    @override
    def __repr__(self) -> str:
        return f"PandasSelection(name='{self.name}', path='{self.path}')"

data: pd.DataFrame | None = None instance-attribute

filters_dict = filters_dict or {} instance-attribute

name = name instance-attribute

path = path instance-attribute

sample_path = sample_path or [] instance-attribute

transforms = transforms or [] instance-attribute

__init__(name: str, path: str, sample_path: list[dict[str, Any]] | None = None, transforms: list[dict[str, Any]] | None = None, filters_dict: dict[str, Any] | None = None)

Initialize a Pandas data selection.

Parameters:

Name Type Description Default
name str

Name identifier for this selection.

 required
path str

Path to the CSV file.

 required
sample_path list[dict[str, Any]] | None

List of sample path configs describing column path prefixes.

 None
transforms list[dict[str, Any]] | None

List of transform configs (column cast operations).

 None
filters_dict dict[str, Any] | None

Column-value pairs to filter rows by.

 None
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    name: str,
    path: str,
    sample_path: list[dict[str, Any]] | None = None,
    transforms: list[dict[str, Any]] | None = None,
    filters_dict: dict[str, Any] | None = None,
):
    """Initialize a Pandas data selection.

    Args:
        name: Name identifier for this selection.
        path: Path to the CSV file.
        sample_path: List of sample path configs describing column path prefixes.
        transforms: List of transform configs (column cast operations).
        filters_dict: Column-value pairs to filter rows by.
    """
    self.name = name
    self.path = path
    self.sample_path = sample_path or []
    self.transforms = transforms or []
    self.filters_dict = filters_dict or {}
    self.data: pd.DataFrame | None = None

__iter__() -> Any

Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
136
137
138
139
140
141
@override
def __iter__(self) -> Any:
    if self.data is not None:
        df = self.data.copy() if self.transforms else self.data
        _apply_pandas_transforms(df, self.transforms)
        yield pa.RecordBatch.from_pandas(df)

__len__() -> int

Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
124
125
def __len__(self) -> int:
    return len(self.data) if self.data is not None else 0

__repr__() -> str

Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
143
144
145
@override
def __repr__(self) -> str:
    return f"PandasSelection(name='{self.name}', path='{self.path}')"

bootstrap(columns_list: list[str] | None = None) -> None

Load the CSV file into memory as a pandas DataFrame.

Parameters:

Name Type Description Default
columns_list list[str] | None

Unused, kept for API compatibility.

 None
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@override
def bootstrap(self, columns_list: list[str] | None = None) -> None:
    """Load the CSV file into memory as a pandas DataFrame.

    Args:
        columns_list: Unused, kept for API compatibility.
    """
    from dqm_ml_job.dataloaders.filters import build_filter_condition

    data = pd.read_csv(self.path, sep=",")
    assert isinstance(data, pd.DataFrame)
    self.data = data
    for col, val in self.filters_dict.items():
        condition = build_filter_condition(
            col,
            val,
            wildcard_fn=lambda c, vals: _match_wildcard(data[c], vals),
            isin_fn=lambda c, vals: data[c].isin(vals),
            equal_fn=lambda c, v: data[c] == v,
        )
        self.data = self.data[condition.reindex(self.data.index, fill_value=False)]

get_nb_batches() -> int

Return the estimated number of batches (always 1 for CSV).

Returns:

Type Description
int

1 if data is loaded, 0 otherwise.
Source code in packages/dqm-ml-job/src/dqm_ml_job/dataloaders/pandas.py 
127
128
129
130
131
132
133
134
@override
def get_nb_batches(self) -> int:
    """Return the estimated number of batches (always 1 for CSV).

    Returns:
        1 if data is loaded, 0 otherwise.
    """
    return 1 if self.data is not None else 0