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 is validated against JobConfig and must follow the v2 structure: - dataloaders: Contains loaders list and optional storage. - features: Optional interface with outputs and processors list. - metrics: Optional interface with outputs and processors list. - gap: Optional interface with outputs and processors list.

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 is validated against JobConfig and must follow the v2 structure:
    - dataloaders: Contains loaders list and optional storage.
    - features: Optional interface with outputs and processors list.
    - metrics: Optional interface with outputs and processors list.
    - gap: Optional interface with outputs and processors list.
    """
    if not config:
        raise ValueError("Job requires a configuration dictionary.")

    validated = JobConfig.model_validate(config)

    dataloaders_registry = PluginLoadedRegistry.get_dataloaders_registry()
    features_registry = PluginLoadedRegistry.get_features_registry()
    metrics_registry = PluginLoadedRegistry.get_metrics_registry()
    gap_registry = PluginLoadedRegistry.get_gap_registry()
    outputs_registry = PluginLoadedRegistry.get_outputwriter_registry()

    # Initialize dataloaders from list format
    dataloader_dicts = [loader.model_dump() for loader in validated.dataloaders.loaders]
    compute = _resolve_compute_config(validated)
    dl_storage = validated.dataloaders.storage.model_dump() if validated.dataloaders.storage else None
    for dl in dataloader_dicts:
        dl["threads"] = compute.threads
        if dl_storage and not dl.get("storage"):
            dl["storage"] = dl_storage
    dataloaders = _init_components_from_list(dataloader_dicts, dataloaders_registry, "dataloader")

    # Resolve storage config: interface override takes precedence over job-level
    def _resolve_storage(interface: Any) -> dict[str, Any] | None:
        if interface and interface.storage:
            result: dict[str, Any] = interface.storage.model_dump()
            return result
        if validated.storage:
            result = validated.storage.model_dump()
            return result
        return None

    # Initialize processors from all interfaces
    features_processors = _init_processors_from_interface(
        validated.features,
        features_registry,
        _resolve_storage(validated.features),
    )
    metrics_processors = _init_processors_from_interface(
        validated.metrics,
        metrics_registry,
        _resolve_storage(validated.metrics),
    )
    gap_processors = _init_processors_from_interface(
        validated.gap,
        gap_registry,
        _resolve_storage(validated.gap),
    )

    # Initialize output writers from interfaces
    features_output = _init_interface_outputs(
        validated.features,
        outputs_registry,
        "features",
        _resolve_storage(validated.features),
    )
    metrics_output = _init_interface_outputs(
        validated.metrics,
        outputs_registry,
        "metrics",
        _resolve_storage(validated.metrics),
    )
    delta_output = _init_interface_outputs(
        validated.gap,
        outputs_registry,
        "delta",
        _resolve_storage(validated.gap),
    )

    # Configure logging based on compute.log_level
    if compute.log_level:
        log_level = compute.log_level.upper()
        level = getattr(logging, log_level)
        logging.basicConfig(level=level)

    job = DatasetJob(
        dataloaders=dataloaders,
        features_processors=features_processors,
        metrics_processors=metrics_processors,
        gap_processors=gap_processors,
        features_output=features_output,
        progress_bar=compute.progress_bar,
        threads=compute.threads,
        errors_by_interface=_build_errors_by_interface(validated),
        compute_seed=compute.seed,
        compute_device=compute.device,
        compute_max_memory=compute.max_memory,
    )

    dataselection_metrics_list, delta_metrics_table = job.run()

    if metrics_output:
        metrics_output.write_metrics_dict(dataselection_metrics_list)
        _safe_flush(metrics_output)

    if delta_output and delta_metrics_table:
        delta_data = {col: delta_metrics_table.column(col) for col in delta_metrics_table.column_names}
        _enrich_delta_with_pairwise(validated, delta_data, delta_metrics_table)
        delta_output.write_table("delta", delta_data)
        _safe_flush(delta_output)