Skip to content

dqm_ml_core.api

API modules for DQM ML Core.

This package contains the base API components for data metric processors, feature extractors, and gap processors.

__all__ = ['FeaturesProcessor', 'GapProcessor', 'MetricsProcessor', 'Processor'] module-attribute

FeaturesProcessor

Bases: Processor

Base class for all feature extraction processors.

Feature processors transform raw data into per-sample features (e.g., image luminosity, embeddings). The primary lifecycle method is compute_features, which produces a dict of feature arrays from each batch of raw data.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class FeaturesProcessor(Processor):
    """
    Base class for all feature extraction processors.

    Feature processors transform raw data into per-sample features
    (e.g., image luminosity, embeddings). The primary lifecycle method
    is ``compute_features``, which produces a dict of feature arrays
    from each batch of raw data.
    """

    def _resolve_output_name(self, col: str, feature_key: str) -> str:
        """Build the output column name with prefix and suffix applied.

        Args:
            col: The input column name.
            feature_key: The feature key to append after ``col_``.

        Returns:
            The fully qualified output column name.
        """
        from dqm_ml_core.models.columns import ColumnsConfig

        cfg: ColumnsConfig | None = getattr(self, "columns_config", None)
        base = f"{col}_{feature_key}"
        p = cfg.prefix if cfg else ""
        s = cfg.suffix if cfg else ""
        return f"{p}{base}{s}"

    def _check_image_fail_fast(self, exc: Exception, *error_attrs: str) -> None:
        if not (self.errors_config and self.errors_config.images):
            return
        image_errors = self.errors_config.images
        for attr in error_attrs:
            if getattr(image_errors, attr, None) == "fail_fast":
                raise exc

    def generated_features(self) -> list[str]:
        """
        Return the list of columns generated by this processor during feature extraction.

        Returns:
            A list of feature names.
        """
        outputs = getattr(self, "output_features", {})
        return list(outputs.values())

    def compute_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Transform a raw data batch into features.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping feature names to pyarrow Arrays.
        """
        features = {}

        available = batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        for col in cols:
            if col in prev_features:
                continue
            features[col] = batch.column(col)

        return features

compute_features(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Transform a raw data batch into features.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping feature names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def compute_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Transform a raw data batch into features.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping feature names to pyarrow Arrays.
    """
    features = {}

    available = batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    for col in cols:
        if col in prev_features:
            continue
        features[col] = batch.column(col)

    return features

generated_features() -> list[str]

Return the list of columns generated by this processor during feature extraction.

Returns:

Type Description
list[str]

A list of feature names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/features_processor.py 
49
50
51
52
53
54
55
56
57
def generated_features(self) -> list[str]:
    """
    Return the list of columns generated by this processor during feature extraction.

    Returns:
        A list of feature names.
    """
    outputs = getattr(self, "output_features", {})
    return list(outputs.values())

GapProcessor

Bases: Processor

Base class for all domain gap processors.

Gap processors compute distribution shift between two datasets (e.g., MMD, FID, KL divergence). The primary lifecycle methods are select_features (per-batch column selection aware of previous features), compute_batch_metric (batch aggregation), compute (final dataset-level statistics), and compute_delta (pairwise comparison).

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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
class GapProcessor(Processor):
    """
    Base class for all domain gap processors.

    Gap processors compute distribution shift between two datasets
    (e.g., MMD, FID, KL divergence). The primary lifecycle methods are
    ``select_features`` (per-batch column selection aware of previous features),
    ``compute_batch_metric`` (batch aggregation), ``compute`` (final dataset-level
    statistics), and ``compute_delta`` (pairwise comparison).
    """

    def select_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Extract relevant columns from a batch, resolving patterns against
        both batch columns and previously computed upstream features.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping column names to pyarrow Arrays.
        """
        available = list(prev_features.keys()) + batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        features: dict[str, pa.Array] = {}
        for col in cols:
            if col in prev_features:
                continue
            if col not in batch.schema.names:
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue
            features[col] = batch.column(col)

        return features

    def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Aggregate features into intermediate statistics for the current batch.

        Args:
            features: Dictionary of feature arrays computed on the batch.

        Returns:
            A dictionary of aggregated statistics.
        """
        return {}

    def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
        """
        Perform the final dataset-level aggregation of batch statistics.

        Args:
            batch_metrics: The aggregated intermediate statistics from all batches.

        Returns:
            A dictionary containing the final dataset-level statistics.
        """
        # SonarQube raises a warning because batch_metrics is not used.
        # It is irrelevant because compute is implemented in child classes which use batch_metrics.
        return {}

    def compute_delta(self, source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
        """
        Compare metrics between two different dataselections.

        Args:
            source: Final metrics from the source dataselection.
            target: Final metrics from the target dataselection.

        Returns:
            A dictionary containing distance or difference scores.
        """
        return {}

compute(batch_metrics: dict[str, pa.Array]) -> dict[str, Any]

Perform the final dataset-level aggregation of batch statistics.

Parameters:

Name Type Description Default
batch_metrics dict[str, Array]

The aggregated intermediate statistics from all batches.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing the final dataset-level statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
    """
    Perform the final dataset-level aggregation of batch statistics.

    Args:
        batch_metrics: The aggregated intermediate statistics from all batches.

    Returns:
        A dictionary containing the final dataset-level statistics.
    """
    # SonarQube raises a warning because batch_metrics is not used.
    # It is irrelevant because compute is implemented in child classes which use batch_metrics.
    return {}

compute_batch_metric(features: dict[str, pa.Array]) -> dict[str, pa.Array]

Aggregate features into intermediate statistics for the current batch.

Parameters:

Name Type Description Default
features dict[str, Array]

Dictionary of feature arrays computed on the batch.

 required

Returns:

Type Description
dict[str, Array]

A dictionary of aggregated statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
59
60
61
62
63
64
65
66
67
68
69
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Aggregate features into intermediate statistics for the current batch.

    Args:
        features: Dictionary of feature arrays computed on the batch.

    Returns:
        A dictionary of aggregated statistics.
    """
    return {}

compute_delta(source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]

Compare metrics between two different dataselections.

Parameters:

Name Type Description Default
source dict[str, Any]

Final metrics from the source dataselection.

 required
target dict[str, Any]

Final metrics from the target dataselection.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing distance or difference scores.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
85
86
87
88
89
90
91
92
93
94
95
96
def compute_delta(self, source: dict[str, Any], target: dict[str, Any]) -> dict[str, Any]:
    """
    Compare metrics between two different dataselections.

    Args:
        source: Final metrics from the source dataselection.
        target: Final metrics from the target dataselection.

    Returns:
        A dictionary containing distance or difference scores.
    """
    return {}

select_features(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Extract relevant columns from a batch, resolving patterns against both batch columns and previously computed upstream features.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping column names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/gap_processor.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def select_features(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Extract relevant columns from a batch, resolving patterns against
    both batch columns and previously computed upstream features.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping column names to pyarrow Arrays.
    """
    available = list(prev_features.keys()) + batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    features: dict[str, pa.Array] = {}
    for col in cols:
        if col in prev_features:
            continue
        if col not in batch.schema.names:
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue
        features[col] = batch.column(col)

    return features

MetricsProcessor

Bases: Processor

Base class for all metric computation processors.

Metric processors compute dataset-level scores (e.g., completeness, diversity, representativeness). The primary lifecycle methods are select_columns (per-batch column selection), compute_batch_metric (batch aggregation), and compute (final dataset-level computation).

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 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
class MetricsProcessor(Processor):
    """
    Base class for all metric computation processors.

    Metric processors compute dataset-level scores (e.g., completeness,
    diversity, representativeness). The primary lifecycle methods are
    ``select_columns`` (per-batch column selection), ``compute_batch_metric``
    (batch aggregation), and ``compute`` (final dataset-level computation).
    """

    def generated_metrics(self) -> list[str]:
        """
        Return the names of the final metrics produced by this processor.

        Returns:
            A list of metric names.
        """
        outputs = getattr(self, "output_metrics", {})
        return list(outputs.values())

    def select_columns(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Select relevant columns from a raw batch for metric computation.

        Args:
            batch: The input pyarrow RecordBatch.
            prev_features: Features already computed by preceding processors.

        Returns:
            A dictionary mapping column names to pyarrow Arrays.
        """
        features = {}

        available = batch.schema.names
        cols = resolve_include_exclude(
            self.input_columns,
            self.exclude_columns,
            available,
        )

        for col in cols:
            if col in prev_features:
                continue

            if col not in available:
                if (
                    self.errors_config
                    and self.errors_config.tabular
                    and self.errors_config.tabular.on_missing_column == "fail_fast"
                ):
                    raise KeyError(f"Column '{col}' not found in batch")
                logger.warning(f"[{self.name}] column '{col}' not found in batch")
                continue
            features[col] = batch.column(col)

        return features

    def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
        """
        Aggregate features into intermediate statistics for the current batch.

        This method is critical for scalability. It should return a compact
        representation of the data (e.g., partial sums) that can be
        efficiently combined later.

        Args:
            features: Dictionary of feature arrays computed on the batch.

        Returns:
            A dictionary of aggregated statistics.
        """
        return {}

    def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
        """
        Perform the final dataset-level metric calculation.

        Args:
            batch_metrics: The aggregated intermediate statistics from all batches.

        Returns:
            A dictionary containing the final metrics.
        """
        # SonarQube raises a warning because batch_metrics is not used.
        # It is irrelevant because compute is implemented in child classes which use batch_metrics.
        return {}

compute(batch_metrics: dict[str, pa.Array]) -> dict[str, Any]

Perform the final dataset-level metric calculation.

Parameters:

Name Type Description Default
batch_metrics dict[str, Array]

The aggregated intermediate statistics from all batches.

 required

Returns:

Type Description
dict[str, Any]

A dictionary containing the final metrics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def compute(self, batch_metrics: dict[str, pa.Array]) -> dict[str, Any]:  # NOSONAR
    """
    Perform the final dataset-level metric calculation.

    Args:
        batch_metrics: The aggregated intermediate statistics from all batches.

    Returns:
        A dictionary containing the final metrics.
    """
    # SonarQube raises a warning because batch_metrics is not used.
    # It is irrelevant because compute is implemented in child classes which use batch_metrics.
    return {}

compute_batch_metric(features: dict[str, pa.Array]) -> dict[str, pa.Array]

Aggregate features into intermediate statistics for the current batch.

This method is critical for scalability. It should return a compact representation of the data (e.g., partial sums) that can be efficiently combined later.

Parameters:

Name Type Description Default
features dict[str, Array]

Dictionary of feature arrays computed on the batch.

 required

Returns:

Type Description
dict[str, Array]

A dictionary of aggregated statistics.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def compute_batch_metric(self, features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Aggregate features into intermediate statistics for the current batch.

    This method is critical for scalability. It should return a compact
    representation of the data (e.g., partial sums) that can be
    efficiently combined later.

    Args:
        features: Dictionary of feature arrays computed on the batch.

    Returns:
        A dictionary of aggregated statistics.
    """
    return {}

generated_metrics() -> list[str]

Return the names of the final metrics produced by this processor.

Returns:

Type Description
list[str]

A list of metric names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
28
29
30
31
32
33
34
35
36
def generated_metrics(self) -> list[str]:
    """
    Return the names of the final metrics produced by this processor.

    Returns:
        A list of metric names.
    """
    outputs = getattr(self, "output_metrics", {})
    return list(outputs.values())

select_columns(batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]

Select relevant columns from a raw batch for metric computation.

Parameters:

Name Type Description Default
batch RecordBatch

The input pyarrow RecordBatch.

 required
prev_features dict[str, Array]

Features already computed by preceding processors.

 required

Returns:

Type Description
dict[str, Array]

A dictionary mapping column names to pyarrow Arrays.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/metrics_processor.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def select_columns(self, batch: pa.RecordBatch, prev_features: dict[str, pa.Array]) -> dict[str, pa.Array]:
    """
    Select relevant columns from a raw batch for metric computation.

    Args:
        batch: The input pyarrow RecordBatch.
        prev_features: Features already computed by preceding processors.

    Returns:
        A dictionary mapping column names to pyarrow Arrays.
    """
    features = {}

    available = batch.schema.names
    cols = resolve_include_exclude(
        self.input_columns,
        self.exclude_columns,
        available,
    )

    for col in cols:
        if col in prev_features:
            continue

        if col not in available:
            if (
                self.errors_config
                and self.errors_config.tabular
                and self.errors_config.tabular.on_missing_column == "fail_fast"
            ):
                raise KeyError(f"Column '{col}' not found in batch")
            logger.warning(f"[{self.name}] column '{col}' not found in batch")
            continue
        features[col] = batch.column(col)

    return features

Processor

Base class for all Data Quality metrics, feature extractors, and gap processors.

Provides shared initialization, failure-rate checking, and column resolution. Lifecycle methods are defined in the appropriate subclass: :class:FeaturesProcessor, :class:MetricsProcessor, or :class:GapProcessor.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class Processor:
    """
    Base class for all Data Quality metrics, feature extractors, and gap processors.

    Provides shared initialization, failure-rate checking, and column resolution.
    Lifecycle methods are defined in the appropriate subclass:
    :class:`FeaturesProcessor`, :class:`MetricsProcessor`, or :class:`GapProcessor`.
    """

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

        Args:
            name: Unique name of the processor instance.
            config: Configuration dictionary (optional).
        """
        self.name = name
        config = config or {}
        self.storage_raw = config.pop("storage", None)
        self.config = config
        self.errors_config = None
        self.compute_device: str = "cpu"
        self.compute_seed: int | None = None
        self.current_path_prefix: dict[str, str] = {}
        self._failure_count = 0
        self._total_count = 0

        self.input_columns: list[str] = []
        self.exclude_columns: list[str] | None = None
        self.columns_config = None
        if "columns" in self.config and isinstance(self.config["columns"], dict):
            from dqm_ml_core.models.columns import ColumnsConfig

            self.columns_config = ColumnsConfig.model_validate(self.config["columns"])
            self.input_columns = self.columns_config.input
            self.exclude_columns = self.columns_config.exclude

    def _check_failure_rate(self) -> None:
        if self.errors_config and self.errors_config.max_failure_rate is not None and self._total_count > 0:
            failure_rate = self._failure_count / self._total_count
            if failure_rate > self.errors_config.max_failure_rate:
                raise RuntimeError(
                    f"Failure rate {failure_rate:.1%} exceeds max {self.errors_config.max_failure_rate:.1%}"
                )

    def needed_columns(self) -> list[str]:
        """
        Return the list of raw input columns required for processing.

        Returns:
            A list of column names.
        """
        return self.input_columns

    def reset(self) -> None:
        """Reset per-selection state between dataselections.

        Processors that cache per-selection state (e.g. histogram bin
        edges in RepresentativenessProcessor) MUST override this to
        clear that state.  Called by DatasetJob after each selection.
        """

columns_config = None instance-attribute

compute_device: str = 'cpu' instance-attribute

compute_seed: int | None = None instance-attribute

config = config instance-attribute

current_path_prefix: dict[str, str] = {} instance-attribute

errors_config = None instance-attribute

exclude_columns: list[str] | None = None instance-attribute

input_columns: list[str] = [] instance-attribute

name = name instance-attribute

storage_raw = config.pop('storage', None) instance-attribute

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

Initialize the processor.

Parameters:

Name Type Description Default
name str

Unique name of the processor instance.

 required
config dict[str, Any] | None

Configuration dictionary (optional).

 required
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(self, name: str, config: dict[str, Any] | None):
    """
    Initialize the processor.

    Args:
        name: Unique name of the processor instance.
        config: Configuration dictionary (optional).
    """
    self.name = name
    config = config or {}
    self.storage_raw = config.pop("storage", None)
    self.config = config
    self.errors_config = None
    self.compute_device: str = "cpu"
    self.compute_seed: int | None = None
    self.current_path_prefix: dict[str, str] = {}
    self._failure_count = 0
    self._total_count = 0

    self.input_columns: list[str] = []
    self.exclude_columns: list[str] | None = None
    self.columns_config = None
    if "columns" in self.config and isinstance(self.config["columns"], dict):
        from dqm_ml_core.models.columns import ColumnsConfig

        self.columns_config = ColumnsConfig.model_validate(self.config["columns"])
        self.input_columns = self.columns_config.input
        self.exclude_columns = self.columns_config.exclude

needed_columns() -> list[str]

Return the list of raw input columns required for processing.

Returns:

Type Description
list[str]

A list of column names.
Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
59
60
61
62
63
64
65
66
def needed_columns(self) -> list[str]:
    """
    Return the list of raw input columns required for processing.

    Returns:
        A list of column names.
    """
    return self.input_columns

reset() -> None

Reset per-selection state between dataselections.

Processors that cache per-selection state (e.g. histogram bin edges in RepresentativenessProcessor) MUST override this to clear that state. Called by DatasetJob after each selection.

Source code in packages/dqm-ml-core/src/dqm_ml_core/api/processor.py 
68
69
70
71
72
73
74
def reset(self) -> None:
    """Reset per-selection state between dataselections.

    Processors that cache per-selection state (e.g. histogram bin
    edges in RepresentativenessProcessor) MUST override this to
    clear that state.  Called by DatasetJob after each selection.
    """