Skip to content

dqm_ml_core.api.metrics_processor

Metrics processor base class.

This module contains the MetricsProcessor base class that all metric computation processors must inherit from.

logger = logging.getLogger(__name__) module-attribute

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