dqm_ml_job

DQM ML Job package for executing data quality assessment pipelines.

This package provides the core job execution framework for running data quality metric computations on datasets. It includes: - CLI entry points for running jobs from YAML configurations - Job orchestration for data loading, metric computation, and output writing - Data loaders for various file formats (Parquet, CSV) - Output writers for persisting results

Example

from dqm_ml_job.cli import run run({"config": {...}})

__all__ = ['ComputeDatasetFeatures'] module-attribute

__description__ = 'DQM ML Job - Data quality assessment pipeline execution' module-attribute

ComputeDatasetFeatures(config: dict[str, Any]) -> None

Execute a job from a validated configuration dictionary.

The config must contain: - dataloaders: Map of configurations for data sources. - metrics_processor: Map of configurations for quality metrics. - outputs: Map of configurations for results storage.

Source code in packages/dqm-ml-job/src/dqm_ml_job/cli.py

def run(config: dict[str, Any]) -> None:
    """
    Execute a job from a validated configuration dictionary.

    The config must contain:
    - dataloaders: Map of configurations for data sources.
    - metrics_processor: Map of configurations for quality metrics.
    - outputs: Map of configurations for results storage.
    """
    dataloaders_registry = PluginLoadedRegistry.get_dataloaders_registry()
    metrics_registry = PluginLoadedRegistry.get_metrics_registry()
    outputs_registry = PluginLoadedRegistry.get_outputwriter_registry()

    if not config:
        raise ValueError("Job requires a configuration dictionary.")

    # Load data loaders
    if "dataloaders" not in config or not isinstance(config["dataloaders"], dict):
        raise ValueError("'dataloaders' must be provided as a dictionary")

    dataloaders: dict[str, DataLoader] = _init_components(config["dataloaders"], dataloaders_registry, "dataloader")

    # Load metrics
    if "metrics_processor" not in config or not isinstance(config["metrics_processor"], dict):
        raise ValueError("'metrics_processor' must be provided as a dictionary")

    metrics: dict[str, DatametricProcessor] = _init_components(config["metrics_processor"], metrics_registry, "metric")

    if "compute_delta" in config:
        logger.warning("compute_delta' is deprecated and will be removed in future versions.")

    # Load output writers
    if "outputs" not in config or not isinstance(config["outputs"], dict):
        raise ValueError("'outputs' must be provided as a dictionary")

    metrics_output: OutputWriter | None = None
    features_output: OutputWriter | None = None
    delta_output: OutputWriter | None = None

    for key, output_config in config["outputs"].items():
        if output_config["type"] not in outputs_registry:
            raise ValueError(f"Output '{key}' must have a valid 'type' in {list(outputs_registry.keys())}")
        writer = outputs_registry[output_config["type"]](name=key, config=output_config)
        if key == "metrics":
            metrics_output = writer
        elif key == "delta_metrics":
            delta_output = writer
        elif key == "features":
            features_output = writer
        else:
            raise ValueError(f"Unsupported output key '{key}'. Only 'features' and 'metrics' are allowed.")

    progress_bar = config.get("progress_bar", True)

    job = DatasetJob(
        dataloaders=dataloaders, metrics=metrics, features_output=features_output, progress_bar=progress_bar
    )

    dataselection_metrics_list, delta_metrics_table = job.run()

    # If we have computed metrics to store
    if metrics_output:
        metrics_output.write_metrics_dict(dataselection_metrics_list)

    # If we have to compute delta metrics
    if delta_output and delta_metrics_table:
        delta_output.write_table("delta", delta_metrics_table)