Skip to content

DQM-ML Core

Core package for DQM-ML V2 providing the foundational API and standard metrics for data quality assessment.

Installation

pip install dqm-ml-core

Note: dqm-ml-core provides Metrics Processors only — no CLI or job orchestration. Use directly via Python or with dqm-ml-job for YAML config execution.

Quick Start: Generate Synthetic Test Data

Create data/core_metrics.parquet with 1000 rows covering all three metric types — Completeness, Representativeness, and Diversity:

# generate_data.py
import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
from pathlib import Path

rng = np.random.default_rng(42)
Path("data").mkdir(exist_ok=True)

n = 1000
# Completeness: numeric columns with ~20% missing values
col_a = np.where(rng.random(n) < 0.2, None, rng.integers(0, 100, n))
col_b = np.where(rng.random(n) < 0.2, None, rng.integers(0, 100, n))

# Representativeness: normal distribution
feature = rng.normal(0, 1, n)

# Diversity: categorical with 5 imbalanced classes
categories = rng.choice(["A", "B", "C", "D", "E"], n, p=[0.4, 0.25, 0.15, 0.12, 0.08])

table = pa.table({"col_a": col_a, "col_b": col_b, "feature": feature, "category": categories})
pq.write_table(table, "data/core_metrics.parquet")
print(f"Generated {n} rows -> data/core_metrics.parquet")

python generate_data.py

Usage

Completeness Example

Note: See Quick Start: Generate Synthetic Test Data to generate data/core_metrics.parquet.

from dqm_ml_core import CompletenessProcessor, ProcessorRunner
import pandas as pd

# Load synthetic data from parquet (generated by Quick Start script)
df = pd.read_parquet("data/core_metrics.parquet")  # columns: col_a, col_b, feature, category

# Configure processor with columns to analyze
processor = CompletenessProcessor(
    name="my_check",
    config={
        "columns": {"input": ["col_a", "col_b"]},
        "include_per_column": True,
        "include_overall": True
    }
)

# Run using ProcessorRunner (high-level API)
runner = ProcessorRunner()
result = runner.run(df, [processor])

print(f"Completeness col_a: {result['completeness_col_a']}")
print(f"Completeness col_b: {result['completeness_col_b']}")
print(f"Overall Completeness: {result['completeness_overall']}")

Representativeness Example

Note: See Quick Start: Generate Synthetic Test Data to generate data/core_metrics.parquet.

from dqm_ml_core import RepresentativenessProcessor
import pyarrow as pa
import pandas as pd

# Load synthetic data from parquet (generated by Quick Start script)
df = pd.read_parquet("data/core_metrics.parquet")
batch = pa.record_batch([pa.array(df["feature"])], names=["feature"])

# Configure processor
processor = RepresentativenessProcessor(
    name="dist_check",
    config={
        "columns": {"input": ["feature"]},
        "distribution": "normal",
        "metrics": ["chi-square", "kolmogorov-smirnov"],
        "mean_std_estimation": "from_first_batch"
    }
)

# Process data through pipeline (direct API)
features = processor.select_columns(batch, prev_features={})
batch_metrics = processor.compute_batch_metric(features)
result = processor.compute(batch_metrics)

print(f"Chi-Square p-value: {result['feature_chi-square_p_value']}")
print(f"KS statistic: {result['feature_kolmogorov-smirnov_statistic']}")
print(f"KS p-value: {result['feature_kolmogorov-smirnov_p_value']}")

With dqm-ml-job

For running from a YAML config, install together with dqm-ml-job:

pip install dqm-ml-job dqm-ml-core

Note: See Quick Start: Generate Synthetic Test Data to generate data/core_metrics.parquet.

Create a YAML config file (e.g., config.yaml):

dataloaders:
  loaders:
    - name: core_data
      type: parquet
      path: data/core_metrics.parquet
      batch_size: 500

metrics:
  processors:
    - name: completeness
      type: completeness
      columns:
        input: ["col_a", "col_b"]
      include_per_column: true
      include_overall: true

    - name: representativeness
      type: representativeness
      columns:
        input: ["feature"]
      distribution: "normal"
      metrics: ["chi-square", "kolmogorov-smirnov"]
      mean_std_estimation: "from_first_batch"

    - name: diversity
      type: diversity
      columns:
        input: ["category"]
      metrics: ["shannon", "gini-simpson", "simpson"]

outputs:
  path: output/metrics.parquet

Execute from Python:

from dqm_ml_job.cli import execute

# Execute a data quality job from a YAML config
execute(["-p", "config.yaml"])

Or from the command line:

python -m dqm_ml_job.cli -p config.yaml

Core Concepts

Three Processor Interfaces

DQM-ML V2 defines three distinct processor interfaces, each with its own base class:

Interface Base Class Purpose
Metrics MetricsProcessor Compute aggregated metric scores from data (Completeness, Representativeness, Diversity)
Features FeaturesProcessor Extract feature columns from data (Visual Features, Embeddings)
Gap GapProcessor Compute pairwise distances between selections (Domain Gap)

All three inherit from a common Processor base class (dqm_ml_core.api.processor:16) which provides: - __init__, _check_failure_rate, _check_image_fail_fast, needed_columns(), reset()

MetricsProcessor

Extends Processor. Implement: - generated_metrics() → list[str] — output metric names - select_columns(batch, prev_features) → dict[str, pa.Array] — select columns (optional, default in base) - compute_batch_metric(features) → dict[str, pa.Array] — batch statistics - compute(batch_metrics) → dict[str, Any] — final scores

FeaturesProcessor

Extends Processor. Implement: - generated_features() → list[str] — output feature column names - compute_features(batch, prev_features) → dict[str, pa.Array] — new feature columns - needed_columns() → list[str] — input columns needed (optional, default: input_columns)

GapProcessor

Extends Processor. Implement: - select_features(batch, prev_features) → dict[str, pa.Array] — retrieve embeddings - compute_batch_metric(features) → dict[str, pa.Array] — batch statistics - compute(batch_metrics) → dict[str, Any] — final scores - compute_delta(source, target) → dict[str, Any] — pairwise distances

Included Metrics

Metric Description
Completeness Analyzes null/missing values in your dataset
Representativeness Statistical distribution analysis (Chi-Square, KS, Shannon Entropy, GRTE)
Diversity Measures category distribution spread (Simpson, Gini-Simpson, Shannon, Richness)

For Developers

To create a new Metrics Processor:

  1. Subclass dqm_ml_core.api.metrics_processor.MetricsProcessor.
  2. Implement generated_metrics(), select_columns() (optional), compute_batch_metric(), and compute().
  3. Register in [project.entry-points."dqm_ml.metrics"] in pyproject.toml.

To create a Features Processor or Gap Processor, use the respective base classes in dqm_ml_core.api.features_processor and dqm_ml_core.api.gap_processor.

Reference implementations: - CompletenessProcessor — simple streaming metric - RepresentativenessProcessor — statistical tests - DiversityProcessor — value-count accumulation

Dependencies

DQM-ML is modular. For core metrics:

# Minimal: use as library only
pip install dqm-ml-core

# For YAML config execution
pip install dqm-ml-job dqm-ml-core

# Full stack with all metrics
pip install dqm-ml-job dqm-ml-core dqm-ml-images dqm-ml-pytorch

See Also