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:

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

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

• Formal and Core Concepts for definitions of Sample, Metric, Data Selection, Batch, and related terminology.
• Metrics Documentation
• API Reference