mlrun.MLRunResult

kedro_datasets_experimental.mlrun.MLRunResult

MLRunResult(
    key=None, flatten=False, load_args=None, save_args=None
)

Bases: MLRunAbstractDataset

Dataset for saving/loading scalar results (metrics) via MLRun.

Uses MLRun's log_result. Results are read from context.results.

load_args and save_args accept any arguments supported by the corresponding MLRun API for your MLRun version; see the MLRun documentation.

Examples:

Using the YAML API:

training_metrics:
  type: kedro_datasets_experimental.mlrun.MLRunResult
  key: metrics
  flatten: true

Using the Python API:

from kedro_datasets_experimental.mlrun import MLRunResult

dataset = MLRunResult(key="metrics", flatten=True)
dataset.save({"accuracy": 0.95, "loss": 0.05})
loaded = dataset.load()

Parameters:

• key (str | None, default: None ) –

  Result key for MLRun (defaults to catalog dataset name).

• flatten (bool, default: False ) –

  If True, flatten nested dictionaries to dot-notation keys.

  When enabled: - Each key is stored as a separate MLRun result. - Loading must be performed per key (e.g. from context.results).

• load_args (dict[str, Any] | None, default: None ) –

  Passed to MLRun when loading; see MLRun docs for your version.

• save_args (dict[str, Any] | None, default: None ) –

  Passed to log_result; see MLRun docs for your version.

Source code in kedro_datasets_experimental/mlrun/result.py

def __init__(
    self,
    key: str | None = None,
    flatten: bool = False,
    load_args: dict[str, Any] | None = None,
    save_args: dict[str, Any] | None = None,
) -> None:
    super().__init__(key=key, save_args=save_args, load_args=load_args)
    self._flatten = flatten

_flatten instance-attribute

_flatten = flatten

_flatten_dict

_flatten_dict(d, prefix='')

Flatten nested dict into dot-notation keys.

Source code in kedro_datasets_experimental/mlrun/result.py

def _flatten_dict(self, d: dict, prefix: str = "") -> dict:
    """Flatten nested dict into dot-notation keys."""
    items = {}
    for k, v in d.items():
        new_key = f"{prefix}.{k}" if prefix else k
        if isinstance(v, dict):
            items.update(self._flatten_dict(v, new_key))
        else:
            items[new_key] = v
    return items

load

load()

Source code in kedro_datasets_experimental/mlrun/result.py

def load(self) -> Any:
    # When flatten=True, results were saved per key; use context.results per key.
    return self._ctx_manager.context.results.get(self.key)

save

save(data)

Source code in kedro_datasets_experimental/mlrun/result.py

def save(self, data: Any) -> None:
    if self._flatten and isinstance(data, dict):
        # Each flattened key is logged as its own MLRun result (no single key holds
        # the full dict). When loading, read each key from context.results.
        flat = self._flatten_dict(data)
        for k, v in flat.items():
            self._ctx_manager.context.log_result(k, v)
    else:
        self._ctx_manager.context.log_result(self.key, data)