getindata / kedro-azureml Goto Github PK

View Code? Open in Web Editor NEW

36.0 12.0 17.0 3.99 MB

Kedro plugin to support running workflows on Microsoft Azure ML Pipelines

Home Page: https://kedro-azureml.readthedocs.io

License: Apache License 2.0

Python 100.00%

azure-ml kedro kedro-plugin machinelearning mlops

kedro-azureml's Introduction

Kedro Azure ML Pipelines plugin

We help companies turn their data into assets

About

Following plugin enables running Kedro pipelines on Azure ML Pipelines service.

We support 2 native Azure Machine Learning types of workflows:

For Data Scientists: fast, iterative development with code upload
For MLOps: stable, repeatable workflows with Docker

Documentation

For detailed documentation refer to https://kedro-azureml.readthedocs.io/

Usage guide

Usage: kedro azureml [OPTIONS] COMMAND [ARGS]...

Options:
  -e, --env TEXT  Environment to use.
  -h, --help      Show this message and exit.

Commands:
  compile  Compiles the pipeline into YAML format
  init     Creates basic configuration for Kedro AzureML plugin
  run      Runs the specified pipeline in Azure ML Pipelines

Quickstart

Follow quickstart section on kedro-azureml.readthedocs.io to get up to speed with plugin usage or watch the video below

kedro-azureml's People

Contributors

Stargazers

Watchers

Forkers

fdroessler tomasvanpottelbergh brucew2099 gustavo-advisia j0rd1smit asafalinadsg jakubigla jpoullet2000 ghazalli gabriel2409 datboi2001 dunnkers eliorc sfczekalski kinjaldand

kedro-azureml's Issues

Pass user input parameters to Azure pipeline at the run time

Hello Team!

I have some designing questions, and see if kedro-azureml is the right tool to use.

My team is building a long running simulation model written using Kedro. We have a ton of parameters in conf/base/. Our goal is to let users configure the parameters from UI (wepapp written in js), and to trigger the model with these parameters from UI. We heard that we can create REST API for Azure pipeline to run the model and pass input parameters to the model. If our team create the pipeline using kedro-azureml, can we pass different parameters at the run time?

I'm new to kedro and azure, so please correct me if I'm wrong.

FileNotFound Error for local runs with uri_files and AzureMLDataset

I manually created an azure data asset (basic csv uri_file) in Azure Machine learning and
I noticed a bug when using the following catalog for a local run (simple kedro run)

#catalog.yml
projects_train_raw:
    type: kedro_azureml.datasets.AzureMLAssetDataSet
    azureml_dataset: test
    root_dir: data/00_azurelocals/
    versioned: True
    dataset:
        type: pandas.CSVDataSet
        filepath: "dataset.csv"

I get the following error
FileNotFoundError: [Errno 2] No such file or directory: 'data/00_azurelocals/test/1/dataset.csv'

To investigate, I forked the repo and I noticed that AzureMLFileSystemMock and AzureMachineLearningFileSystem
have a slightly different behavior (see below)

from tests.conftest import AzureMLFileSystemMock
from azureml.fsspec import AzureMachineLearningFileSystem

azureml_ds_path = 'azureml://subscriptions/1234/resourcegroups/dummy_rg/workspaces/dummy_ws/datastores/some_datastore/paths/test_file/test.pickle'

mockfs = AzureMLFileSystemMock(azureml_ds_path)
amlfs = AzureMachineLearningFileSystem(azureml_ds_path)
mockfs_path_on_azure = mockfs._infer_storage_options(azureml_ds_path)[-1]
amlfs_path_on_azure = amlfs._infer_storage_options(azureml_ds_path)[-1]
print(mockfs_path_on_azure) # PosixPath(test_file)
print(amlfs_path_on_azure) # PosixPath(test_file/test.pickle)

print(amlfs.ls(amlfs_path_on_azure)) # empty list !

If you look at the definition of AzureMLFileSystemMock._infer_storage_options,
there is a condition on path_on_azure.suffix

def _infer_storage_options(self, uri):
    path_on_azure = Path(
        AzureMachineLearningFileSystem._infer_storage_options(uri)[-1]
    )
    if path_on_azure.suffix != "":
        path_on_azure = str(path_on_azure.parent)
    else:
        path_on_azure = str(path_on_azure)
    return [self._prefix / path_on_azure]

However, in a local run in a normal context (not test), in the AzureMLAssetDataSet,
for "uri_file", you don't set path_on_azure to the parent folder

def _load(self) -> Any:
    if self._download:
        ...
        fs = AzureMachineLearningFileSystem(azureml_ds.path)
        if azureml_ds.type == "uri_file":
            path_on_azure = fs._infer_storage_options(azureml_ds.path)[-1]
        elif azureml_ds.type == "uri_folder":
            ...
        else:
            ...

        # When you arrive at this line, in the test version, for a given uri file,
        # path_on_azure is set to the parent but not in the standard implementation.

        # the ls method will correctly list the files in the mock implementation
        # but not in the standard implementation where it will return an empty list
        for fpath in fs.ls(path_on_azure):
            ...
    return self._construct_dataset().load()

I was able to fix it by tweaking the _load function but it also requires modifying
AzureMLFileSystemMock. However there may be a better way to do it.

Fixes #69

Retrieving pipeline id in cicd pipelines

I am having a bit of trouble registering a model after launching a pipeline with kedro azureml.
Indeed, to find the actual id of the pipeline that trained the model, I currently take the most recent (with a few additional filters to make sure i get the correct one) but I don't think that is a great solution.

I noticed that the pipeline id is available in AzureMLPipelinesClient.run (it is just pipeline_job.name).

Is there an intelligent way to retrieve it?

I could imagine logging it to the output then do something like
pipeline_text=$(kedro azureml run ...)
and then filter pipeline_text to get the actual value but it seems very ugly.
Plus you would need to make sure your pipeline log level is set correctly so I don't think it is ideal.

Add option to enable cache via configuration

Right now, Azure ML built-in caching mechanism is disabled by the plugin explicitly.
We should add an option to enable it on specific nodes based on tags (in the same way as the resources are configured).

Proposed configuration azureml.yml:

azure:
  # rest of the config
  cache:
    __default__: false
    specific_tag: true

then, all nodes tagged with specific_tag should have is_deterministic flag set to true:

kedro-azureml/kedro_azureml/generator.py

Line 207 in 2e5836b

 is_deterministic=False, # TODO: allow setting this to true per node (e.g. by tags as for resources) 

Additional help/documentation for new users [help wanted]

Hi, I am a new user of your plugin. First let me thank you for putting together such a wonderful tool. I have pretty good experience with Kedro and deploying models to on-prem servers.

I am new to deploying to Azure and I was hoping for some advice/best practices on deploying using Kedro-Azureml.

I am able to get my ML pipeline running using the Docker deployment method. I am trying to setup a CI/CD for this pipeline with Azure DevOps and am running into some difficulties. Do you setup a remote repository in Azure Devops and setup an Azure DevOps pipeline that triggers kedro azureml run on push to the main branch? Does kedro-azureml plugin generate a YAML file that can be used with Azure DevOps for setting up the pipeline?

Any advice would be greatly appreciated thank you.

Evaluate integration with kedro-mlflow

Right now, there is no support for kedro-mlflow as it handles the mlflow run initialization on it's own via kedro hooks. In Azure ML Pipelines, at runtime, all of the MLflow environment variables are automatically set (pointing to Azure ML's built-in MLFlow).

Evaluate whether we can override some of the logic from kedro-mlflow to allow it to log to the Azure ML's native MLflow. If it's possible, implement the integration and provide a clear instructions how to use it.

Support versioning in AzureMLPipelineDataSet

It would be nice to support versioning of the underlying dataset in AzureMLPipelineDataSet when running the pipeline locally. We could dynamically disable the versioning then in the Azure ML runner.

Originally posted by @tomasvanpottelbergh in #43 (comment)

ValueError: Given configuration path either does not exist or is not a valid directory

I am using kedro-azureml==0.3.1 and facing an issue with executing a pipeline.
I have a set environment on my workspace

An I am using my own custom kedro starter, which uses Poetry for package management, so the Dockerfile for creating this environment looks like this

ARG BASE_IMAGE=python:3.8-buster
FROM $BASE_IMAGE

ARG AZURE_STORAGE_ACCOUNT_NAME
ARG AZURE_STORAGE_ACCOUNT_KEY

# install project requirements
ENV PYTHONFAULTHANDLER=1 \
    PYTHONUNBUFFERED=1 \
    PYTHONHASHSEED=random \
    PIP_NO_CACHE_DIR=off \
    PIP_DISABLE_PIP_VERSION_CHECK=on \
    PIP_DEFAULT_TIMEOUT=100 \
    PATH="${PATH}:/root/.local/bin/poetry:/home/kedro/src" \
    AZURE_STORAGE_ACCOUNT_NAME="$AZURE_STORAGE_ACCOUNT_NAME" \
    AZURE_STORAGE_ACCOUNT_KEY="$AZURE_STORAGE_ACCOUNT_KEY"

RUN curl -sSL https://install.python-poetry.org | python3 -
COPY poetry.lock pyproject.toml ./
RUN /root/.local/bin/poetry config virtualenvs.create false && \
    /root/.local/bin/poetry install --no-interaction --no-root --no-ansi && \
    rm poetry.lock pyproject.toml

If it is relevant, this Dockerfile, with the addition of configuring user and workdir, worked perfectly in previous kedro-azureml versions.

Now, when I execute using the azureml environment shown in the image, I get the following error

ProjectMetadata(config_file=PosixPath('/mnt/azureml/cr/j/20935b5bfe9d4a9796036cb173eeffc0/exe/wd/pyproject.toml'), package_name='kedro_test', project_name='kedro-test', project_path=PosixPath('/mnt/azureml/cr/j/20935b5bfe9d4a9796036cb173eeffc0/exe/wd'), project_version='0.18.3', source_dir=PosixPath('/mnt/azureml/cr/j/20935b5bfe9d4a9796036cb173eeffc0/exe/wd/src'))
╭───────────────────── Traceback (most recent call last) ──────────────────────╮
│ /usr/local/bin/kedro:8 in │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/framework/cli/cli.py:211 in │
│ main │
│ │
│ 208 │ """ │
│ 209 │ _init_plugins() │
│ 210 │ cli_collection = KedroCLI(project_path=Path.cwd()) │
│ ❱ 211 │ cli_collection() │
│ 212 │
│ │
│ /usr/local/lib/python3.8/site-packages/click/core.py:1130 in call │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/framework/cli/cli.py:139 in │
│ main │
│ │
│ 136 │ │ ) │
│ 137 │ │ │
│ 138 │ │ try: │
│ ❱ 139 │ │ │ super().main( │
│ 140 │ │ │ │ args=args, │
│ 141 │ │ │ │ prog_name=prog_name, │
│ 142 │ │ │ │ complete_var=complete_var, │
│ │
│ /usr/local/lib/python3.8/site-packages/click/core.py:1055 in main │
│ │
│ /usr/local/lib/python3.8/site-packages/click/core.py:1657 in invoke │
│ │
│ /usr/local/lib/python3.8/site-packages/click/core.py:1657 in invoke │
│ │
│ /usr/local/lib/python3.8/site-packages/click/core.py:1404 in invoke │
│ │
│ /usr/local/lib/python3.8/site-packages/click/core.py:760 in invoke │
│ │
│ /usr/local/lib/python3.8/site-packages/click/decorators.py:38 in new_func │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro_azureml/cli.py:312 in execute │
│ │
│ 309 ): │
│ 310 │ # 1. Run kedro │
│ 311 │ parameters = parse_extra_params(params) │
│ ❱ 312 │ with KedroContextManager( │
│ 313 │ │ ctx.metadata.package_name, env=ctx.env, extra_params=parameter │
│ 314 │ ) as mgr: │
│ 315 │ │ runner = AzurePipelinesRunner() │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro_azureml/utils.py:33 in │
│ enter │
│ │
│ 30 │ │ return KedroAzureMLConfig.parse_obj(self.context.config_loader. │
│ 31 │ │
│ 32 │ def enter(self): │
│ ❱ 33 │ │ self.session = KedroSession.create( │
│ 34 │ │ │ self.package_name, env=self.env, extra_params=self.extra_pa │
│ 35 │ │ ) │
│ 36 │ │ self.context = self.session.load_context() │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/framework/session/session.py:18 │
│ 1 in create │
│ │
│ 178 │ │ session._store.update(session_data) │
│ 179 │ │ │
│ 180 │ │ # we need a ConfigLoader registered in order to be able to set │
│ ❱ 181 │ │ session._setup_logging() │
│ 182 │ │ return session │
│ 183 │ │
│ 184 │ def _get_logging_config(self) -> Dict[str, Any]: │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/framework/session/session.py:19 │
│ 8 in _setup_logging │
│ │
│ 195 │ def _setup_logging(self) -> None: │
│ 196 │ │ """Register logging specified in logging directory.""" │
│ 197 │ │ try: │
│ ❱ 198 │ │ │ logging_config = self._get_logging_config() │
│ 199 │ │ except MissingConfigException: │
│ 200 │ │ │ self._logger.debug( │
│ 201 │ │ │ │ "No project logging configuration loaded; " │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/framework/session/session.py:18 │
│ 5 in _get_logging_config │
│ │
│ 182 │ │ return session │
│ 183 │ │
│ 184 │ def _get_logging_config(self) -> Dict[str, Any]: │
│ ❱ 185 │ │ logging_config = self._get_config_loader().get( │
│ 186 │ │ │ "logging*", "logging*/", "/logging*" │
│ 187 │ │ ) │
│ 188 │ │ # turn relative paths in logging config into absolute path │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/config/config.py:101 in get │
│ │
│ 98 │ │ return _remove_duplicates(self._build_conf_paths()) │
│ 99 │ │
│ 100 │ def get(self, *patterns: str) -> Dict[str, Any]: │
│ ❱ 101 │ │ return _get_config_from_patterns( │
│ 102 │ │ │ conf_paths=self.conf_paths, patterns=list(patterns) │
│ 103 │ │ ) │
│ 104 │
│ │
│ /usr/local/lib/python3.8/site-packages/kedro/config/common.py:69 in │
│ _get_config_from_patterns │
│ │
│ 66 │ │
│ 67 │ for conf_path in conf_paths: │
│ 68 │ │ if not Path(conf_path).is_dir(): │
│ ❱ 69 │ │ │ raise ValueError( │
│ 70 │ │ │ │ f"Given configuration path either does not exist " │
│ 71 │ │ │ │ f"or is not a valid directory: {conf_path}" │
│ 72 │ │ │ ) │
╰──────────────────────────────────────────────────────────────────────────────╯
ValueError: Given configuration path either does not exist or is not a valid
directory: /mnt/azureml/cr/j/20935b5bfe9d4a9796036cb173eeffc0/exe/wd/conf/local

My .amlignore is empty if it's relevant

Current release candidate doesn't build on arm mac


Cannot install azureml-dataprep-native.

  - Installing azureml-dataprep-rslex (2.22.2): Failed

  RuntimeError

  Unable to find installation candidates for azureml-dataprep-rslex (2.22.2)

There are no wheels for Apple Silicon available for azureml v1, the solution recommended by Microsoft is to upgrade to azure sdk v2

Add subscription to config

Currently the Azure subscription ID needs to be provided with every invocation of the kedro azureml run command or needs to be set using the environment variable AZURE_SUBSCRIPTION_ID. I think it would be more consistent to add this to the configuration with an option to override via a CLI option.

Am I missing something or would a PR for this be welcome?

Azure Storage credentials for storing files from the data catalog

Hi there,

how would you access the Azure Storage from the data catalog assuming that the storage to be accessed is other than the default one called temporary_storage?
Is there any mechanism to achieve that? Thanks

Bug using AzureMLAssetDataset locally

When using the AzureMLAssetDataset it all works fine when deployed. However, I get an error locally when one pipeline outputs an AzureMLAssetDataset, and another pipeline tries to consume this asset. Here is a reproducible example:

The first pipeline:

from kedro.pipeline import Pipeline, node
import pandas as pd


def create_dataset():
    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
    return df


def create_pipeline(**kwargs) -> Pipeline:
    return Pipeline(
        nodes=[
            node(
                func=create_dataset,
                inputs=None,
                outputs="test_raw",
                name="create_test_raw",
            ),
        ],
    )

The second pipeline:

from kedro.pipeline import Pipeline, node


def create_pipeline(**kwargs) -> Pipeline:
    return Pipeline(
        nodes = [
            node(
                func=lambda x: x,
                inputs="test_raw",
                outputs="test_raw_copy",
                name="copy_test_raw"
            )
        ],
    )

and the catalog:

test_raw:
  type: kedro_azureml.datasets.AzureMLAssetDataset
  azureml_dataset: test_raw
  root_dir: data/00_azurelocals
  versioned: true
  dataset:
    type: pandas.CSVDataset
    filepath: test_raw.csv

test_raw_copy:
  type: kedro_azureml.datasets.AzureMLAssetDataset
  azureml_dataset: test_raw_copy
  root_dir: data/00_azurelocals
  versioned: true
  dataset:
    type: pandas.CSVDataset
    filepath: test_raw_copy.csv

When running the first pipeline locally with kedro run --pipeline test, it creates a local file at data/00_azurelocals/test_raw/local/test_raw.csv. Then when running the second pipeline with kedro run --pipeline copy_test, I get the following stack trace:

(enerfore-deployment) C:\Users\Robert.McLeod2\git_repos\ptx-ds-enerfore-deployment>kedro run --pipeline copy_test
[07/17/24 16:34:09] INFO     Kedro project ptx-ds-enerfore-deployment                                                                                                                                                                                session.py:365
[07/17/24 16:34:18]                                                                                                                                                                 
                    WARNING  Replacing dataset 'test_raw'                                                                                                                                                                                       data_catalog.py:606
                    WARNING  Replacing dataset 'test_raw_copy'                                                                                                                                                                                  data_catalog.py:606
                    INFO     Loading data from 'test_raw' (AzureMLAssetDataset)...                                                                                                                                                              data_catalog.py:502
Found the config file in: C:\Users\ROBERT~1.MCL\AppData\Local\Temp\tmpxxwei3q5\config.json
Found the config file in: C:\Users\ROBERT~1.MCL\AppData\Local\Temp\tmp5omkfo_i\config.json
[07/17/24 16:34:31] WARNING  No nodes ran. Repeat the previous command to attempt a new run.                                                                                                                                                          runner.py:213
Traceback (most recent call last):
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\_utils\_asset_utils.py", line 775, in _get_latest_version_from_container
    else container_operation.get(
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\core\tracing\decorator.py", line 94, in wrapper_use_tracer
    return func(*args, **kwargs)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\_restclient\v2023_04_01_preview\operations\_data_containers_operations.py", line 430, in get
    map_error(status_code=response.status_code, response=response, error_map=error_map)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\core\exceptions.py", line 161, in map_error
    raise error
azure.core.exceptions.ResourceNotFoundError: (UserError) test_raw container was not found.
Code: UserError
Message: test_raw container was not found.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\operations\_data_operations.py", line 265, in get
    return _resolve_label_to_asset(self, name, label)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\_utils\_asset_utils.py", line 1022, in _resolve_label_to_asset
    return resolver(name)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\operations\_data_operations.py", line 675, in _get_latest_version
    latest_version = _get_latest_version_from_container(
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\_utils\_asset_utils.py", line 795, in _get_latest_version_from_container
    raise ValidationException(
azure.ai.ml.exceptions.ValidationException: Asset test_raw does not exist in workspace azuremlworkspace.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\io\core.py", line 193, in load
    return self._load()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro_azureml\datasets\asset_dataset.py", line 188, in _load
    azureml_ds = self._get_azureml_dataset()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro_azureml\datasets\asset_dataset.py", line 182, in _get_azureml_dataset
    self._azureml_dataset, version=self.resolve_load_version()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\io\core.py", line 576, in resolve_load_version
    return self._fetch_latest_load_version()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\cachetools\__init__.py", line 799, in wrapper
    v = method(self, *args, **kwargs)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro_azureml\datasets\asset_dataset.py", line 175, in _fetch_latest_load_version
    return self._get_latest_version()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro_azureml\datasets\asset_dataset.py", line 169, in _get_latest_version
    return ml_client.data.get(self._azureml_dataset, label="latest").version
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\_telemetry\activity.py", line 292, in wrapper
    return f(*args, **kwargs)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\operations\_data_operations.py", line 279, in get
    log_and_raise_error(ex)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\azure\ai\ml\_exception_helper.py", line 337, in log_and_raise_error
    raise MlException(message=formatted_error, no_personal_data_message=formatted_error)
azure.ai.ml.exceptions.MlException:


1) Resource was not found.


Details:

(x) Asset test_raw does not exist in workspace azuremlworkspace.

Resolutions:
1) Double-check that the resource has been specified correctly and that you have access to it.
If using the CLI, you can also check the full log in debug mode for more details by adding --debug to the end of your command

Additional Resources: The easiest way to author a yaml specification file is using IntelliSense and auto-completion Azure ML VS code extension provides: https://code.visualstudio.com/docs/datascience/azure-machine-learning. To set up VS Code, visit https://docs.microsoft.com/azure/machine-learning/how-to-setup-vs-code


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\runpy.py", line 196, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\runpy.py", line 86, in _run_code
    exec(code, run_globals)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\Scripts\kedro.exe\__main__.py", line 7, in <module>
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\framework\cli\cli.py", line 211, in main
    cli_collection()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\click\core.py", line 1157, in __call__
    return self.main(*args, **kwargs)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\framework\cli\cli.py", line 139, in main
    super().main(
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\click\core.py", line 1078, in main
    rv = self.invoke(ctx)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\click\core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\click\core.py", line 1434, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\click\core.py", line 783, in invoke
    return __callback(*args, **kwargs)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\framework\cli\project.py", line 453, in run
    session.run(
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\framework\session\session.py", line 436, in run
    run_result = runner.run(
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\runner\runner.py", line 103, in run
    self._run(pipeline, catalog, hook_manager, session_id)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\framework\session\session.py", line 436, in run
    run_result = runner.run(
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\runner\runner.py", line 103, in run
    self._run(pipeline, catalog, hook_manager, session_id)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\runner\runner.py", line 103, in run
    self._run(pipeline, catalog, hook_manager, session_id)
    self._run(pipeline, catalog, hook_manager, session_id)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\runner\sequential_runner.py", line 70, in _run
    run_node(node, catalog, hook_manager, self._is_async, session_id)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\runner\runner.py", line 331, in run_node
    node = _run_node_sequential(node, catalog, hook_manager, session_id)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\runner\runner.py", line 414, in _run_node_sequential
    inputs[name] = catalog.load(name)
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\io\data_catalog.py", line 506, in load
    result = dataset.load()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\io\core.py", line 614, in load
    return super().load()
  File "C:\Users\Robert.McLeod2\AppData\Local\anaconda3\envs\enerfore-deployment\lib\site-packages\kedro\io\core.py", line 202, in load
    raise DatasetError(message) from exc
kedro.io.core.DatasetError: Failed while loading data from data set AzureMLAssetDataset(dataset_config={'filepath': test_raw.csv}, dataset_type=CSVDataset, filepath_arg=filepath, root_dir=data/00_azurelocals).



1) Resource was not found.


Details:

(x) Asset test_raw does not exist in workspace azuremlworkspace.

Resolutions:
1) Double-check that the resource has been specified correctly and that you have access to it.
If using the CLI, you can also check the full log in debug mode for more details by adding --debug to the end of your command

Additional Resources: The easiest way to author a yaml specification file is using IntelliSense and auto-completion Azure ML VS code extension provides: https://code.visualstudio.com/docs/datascience/azure-machine-learning. To set up VS Code, visit https://docs.microsoft.com/azure/machine-learning/how-to-setup-vs-code

So it seems like it is trying to find a version of the file on Azure, rather than using the local copy. When there is a version on Azure that exists, it puts the version number of the Dataset on azure in the directory path rather than local, i.e. it will look for a file at data/00_azurelocals/test_raw/4/test_raw.csv

I'm not sure why it is trying to find the dataset on Azure, but I would expect the behaviour would be to just look at the local files instead. This error only happens when using an AzureMLAssetDataset as an input locally. Any help is appreciated, thanks.

[help wanted] Best way to configure model for batch inference

Hello, I have been using kedro-azureml specifically for online inference and it has been really nice. So first let me say thank you for all that you do.

For my question, I am having a hard time figuring out how to use kedro-azureml to create an azureml pipeline that is used for batch predictions.

What I am trying to achieve are the following:

Read data directly from Azure DataLake something like "adl://path"
Specify the location of the output to either adl or to blob storage
use azureml to schedule this job to run on scheduler

My specific use case is that I always need to retrain the model (A survival model written in PYMC) before producing predictions so ideally I just want a kedro pipeline to be converted to an azureml pipeline where I can specify the input and output.

Are these functions already something that are available in kedro-azureml?

Parse params in Kedro-native way

Right now, additional run params are passed via JSON object in command line. Kedro uses different syntax --params param_key1:value1,param_key2:2.0 (https://kedro.readthedocs.io/en/stable/kedro_project_setup/configuration.html#specify-parameters-at-runtime). As per https://stackoverflow.com/a/70607923/1955346 it now also supports nested params, so let's use this native way of passing parameters to kedro azureml run and kedro azureml execute commands.

AzureMLPipelineDataSet does not play well with other datasets

@tomasvanpottelbergh this is related to the AzureMLPipelineDataSet you've implemented.

Issue

AzureMLPipelineDataSet does not work with i.e. AzureMLPandasDataSet when pipeline data passing is enabled.

Steps to reproduce:

Use spaceflights with the following catalog:

companies:
  type: pandas.CSVDataSet
  filepath: data/01_raw/companies.csv
  layer: raw

reviews:
  type: pandas.CSVDataSet
  filepath: data/01_raw/reviews.csv
  layer: raw

shuttles:
  type: pandas.ExcelDataSet
  filepath: data/01_raw/shuttles.xlsx
  layer: raw

# TODO: this does not work
preprocessed_companies:
  type: kedro_azureml.datasets.pandas_dataset.AzureMLPandasDataSet
  azureml_dataset: e2e_tests_preprocessed_companies

preprocessed_shuttles:
  type: kedro_azureml.datasets.AzureMLPipelineDataSet
  dataset:
    type: pandas.CSVDataSet
    filepath: "shuttles.csv"

Enable pipeline_data_passing feature in azureml.yml

azure:
    # ...
    pipeline_data_passing:
        enabled: true

Run the pipeline (I'm using docker flow, but it should not matter).

Expected result

Pipeline executes normally.

Actual result

Job fails as soon as the create_model_input_table_node starts.

Error Code: ScriptExecution.StreamAccess.NotFound
Native Error: error in streaming from input data sources
	StreamError(NotFound)
=> stream not found
	NotFound
Error Message: The requested stream was not found. Please make sure the request uri is correct.| session_id=fb6a7934-81ac-4947-b2d8-d39bdcc4fad8

I suspect that the pipeline_data_passing feature assumes that there will always be something in the --az-input locations, while that is not always the case in Kedro, as users CAN specify intermediate data anywhere in the catalog.

Add support for Azure ML Datasets

Azure ML Studio provides has a Dataset abstraction which allows to register datasets with some data, validate it's schema etc.
It would be nice to have a integration with this piece of Azure ML's SDK in form of custom Kedro DataSet to allow storing and retrieving the data directly from Azure. This would be a next level of integration with AML service.

https://docs.microsoft.com/en-us/azure/machine-learning/how-to-create-register-data-assets?tabs=Python-SDK

Proper way to persist interpolations

@marrrcin I am opening up this ticket as a follow-up to ticket #88

I was able to follow your helpful instructions and got the pipeline from Azure DevOps running. The issue I am having now is that while the pipeline will send the kedro pipeline to Azure ML. The interpolations using oc.env do not persist in Azure ML and hence the pipeline will fail.
This is what my azureml.yml looks like:

azure:
  # Azure subscription ID to use
  subscription_id: ${oc.env:AZURE_SUBSCRIPTION_ID}
  # Azure ML Experiment Name
  experiment_name: "kedro_azml"
  # Azure resource group to use
  resource_group: ${oc.env:AZURE_RESOURCE_GROUP}
  # Azure ML Workspace name
  workspace_name: ${oc.env:AZURE_WORKSPACE_NAME}
  # Azure ML Environment to use during pipeline execution
  environment_name: ~
  # Path to directory to upload, or null to disable code upload
  code_directory: ~

Where the environment variables are set in the Azure DevOps Pipeline.
Again, the pipeline succeeds and sends the job to Azure ML. However, those environment variables are not persisted so in Azure ML the job fails:

Should I be setting these variables in the docker image itself or on the compute cluster somehow or something else?
Any help will be very appreciated.

Type hint issue with distributed_job Framework

When using decorator

@distributed_job(Framework.PyTorch, num_nodes="params:num_nodes")

I have the following warning

Argument of type "str" cannot be assigned to parameter "framework" of type "Framework" in function "distributed_job"
  "str" is incompatible with "Framework"

I am not sure if the typing error is due to my config or if it is standard behavior. In any case, this can be easily fixed by making the Framework class inherit from Enum

from enum import Enum
...

class Framework(Enum):
    PyTorch = "PyTorch"
    TensorFlow = "TensorFlow"
    MPI = "MPI"

Handle outputs better

Right now, the "Azure ML Pipelines-native" outputs are left with dummy data just for the sake of connecting the nodes. Consider adding a special flag to the run (or config like use_native_outputs_for_intermediate_data) to output the intermediate data to those output paths instead of the Azure Blob Storage ones.

A great solution would be to also recognize the intermediate data types and serialize them into appropriate format (e.g. Pandas -> Parquet, Text -> Text, Numpy -> npz etc), this would probably allow to visualize the output in the Azure ML UI.

AttributeError because of a running order of after_context_created hooks when using kedro-telemetry

When using session.load_context(), an error is raised in AzureMLLocalRunHook's after_catalog_created function, if kedro-telemetry are installed and opted in.

Steps to reproduce:

import os
from kedro.framework.session import KedroSession
from kedro.framework.startup import bootstrap_project

metadata = bootstrap_project(os.getcwd())
with KedroSession.create(metadata.package_name, os.getcwd(), env="base") as session:
    context = session.load_context()

My understanding of it is that at the end of the load_context function, kedro does self._hook_manager.hook.after_context_created(context=context) which tries to apply all the after_context_created hooks for the context. There is a after_context_created hook in kedro-azureml that defines this self.azure_config, but the ones from kedro-telemetry is called before it, and this one has a line catalog = context.catalog, which requires to create a catalog. At the end of the creation of the catalog, kedro does self._hook_manager.hook.after_catalog_created(... to run the post-creation hooks of the catalog, for which kedro-azureml has one that tries to use dataset.azure_config = self.azure_config. All this happens inside the first hook from kedro-telemetry, which means that the after_context_created hook of kedro-azureml was not run yet, and so self.azure_config does not exist.

Full stack trace

╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ /mnt/batch/tasks/shared/LS_root/mounts/clusters/instance/code/Users/username/project_name.       │
│ /test_kedro.py:8 in <module>                                                                     │
│                                                                                                  │
│   5 metadata = bootstrap_project(os.getcwd())                                                    │
│   6                                                                                              │
│   7 with KedroSession.create(metadata.package_name, os.getcwd(), env="base") as session:         │
│ ❱ 8 │   context = session.load_context()                                                         │
│   9                                                                                              │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/kedro/framework/session/session.py:283 in    │
│ load_context                                                                                     │
│                                                                                                  │
│   280 │   │   │   extra_params=extra_params,                                                     │
│   281 │   │   │   hook_manager=self._hook_manager,                                               │
│   282 │   │   )                                                                                  │
│ ❱ 283 │   │   self._hook_manager.hook.after_context_created(context=context)                     │
│   284 │   │                                                                                      │
│   285 │   │   return context                                                                     │
│   286                                                                                            │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_hooks.py:265 in __call__             │
│                                                                                                  │
│   262 │   │   else:                                                                              │
│   263 │   │   │   firstresult = False                                                            │
│   264 │   │                                                                                      │
│ ❱ 265 │   │   return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)        │
│   266 │                                                                                          │
│   267 │   def call_historic(self, result_callback=None, kwargs=None):                            │
│   268 │   │   """Call the hook with given ``kwargs`` for all registered plugins and              │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_manager.py:80 in _hookexec           │
│                                                                                                  │
│    77 │   def _hookexec(self, hook_name, methods, kwargs, firstresult):                          │
│    78 │   │   # called from all hookcaller instances.                                            │
│    79 │   │   # enable_tracing will set its own wrapping function at self._inner_hookexec        │
│ ❱  80 │   │   return self._inner_hookexec(hook_name, methods, kwargs, firstresult)               │
│    81 │                                                                                          │
│    82 │   def register(self, plugin, name=None):                                                 │
│    83 │   │   """Register a plugin and return its canonical name or ``None`` if the name         │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_manager.py:327 in traced_hookexec    │
│                                                                                                  │
│   324 │   │   │   │   lambda: oldcall(hook_name, hook_impls, kwargs, firstresult)                │
│   325 │   │   │   )                                                                              │
│   326 │   │   │   after(outcome, hook_name, hook_impls, kwargs)                                  │
│ ❱ 327 │   │   │   return outcome.get_result()                                                    │
│   328 │   │                                                                                      │
│   329 │   │   self._inner_hookexec = traced_hookexec                                             │
│   330                                                                                            │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_result.py:60 in get_result           │
│                                                                                                  │
│   57 │   │   │   return self._result                                                             │
│   58 │   │   else:                                                                               │
│   59 │   │   │   ex = self._excinfo                                                              │
│ ❱ 60 │   │   │   raise ex[1].with_traceback(ex[2])                                               │
│   61                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_result.py:33 in from_call            │
│                                                                                                  │
│   30 │   │   __tracebackhide__ = True                                                            │
│   31 │   │   result = excinfo = None                                                             │
│   32 │   │   try:                                                                                │
│ ❱ 33 │   │   │   result = func()                                                                 │
│   34 │   │   except BaseException:                                                               │
│   35 │   │   │   excinfo = sys.exc_info()                                                        │
│   36                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_manager.py:324 in <lambda>           │
│                                                                                                  │
│   321 │   │   def traced_hookexec(hook_name, hook_impls, kwargs, firstresult):                   │
│   322 │   │   │   before(hook_name, hook_impls, kwargs)                                          │
│   323 │   │   │   outcome = _Result.from_call(                                                   │
│ ❱ 324 │   │   │   │   lambda: oldcall(hook_name, hook_impls, kwargs, firstresult)                │
│   325 │   │   │   )                                                                              │
│   326 │   │   │   after(outcome, hook_name, hook_impls, kwargs)                                  │
│   327 │   │   │   return outcome.get_result()                                                    │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_callers.py:60 in _multicall          │
│                                                                                                  │
│   57 │   │   │   except StopIteration:                                                           │
│   58 │   │   │   │   pass                                                                        │
│   59 │   │                                                                                       │
│ ❱ 60 │   │   return outcome.get_result()                                                         │
│   61                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_result.py:60 in get_result           │
│                                                                                                  │
│   57 │   │   │   return self._result                                                             │
│   58 │   │   else:                                                                               │
│   59 │   │   │   ex = self._excinfo                                                              │
│ ❱ 60 │   │   │   raise ex[1].with_traceback(ex[2])                                               │
│   61                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_callers.py:39 in _multicall          │
│                                                                                                  │
│   36 │   │   │   │   │   except StopIteration:                                                   │
│   37 │   │   │   │   │   │   _raise_wrapfail(gen, "did not yield")                               │
│   38 │   │   │   │   else:                                                                       │
│ ❱ 39 │   │   │   │   │   res = hook_impl.function(*args)                                         │
│   40 │   │   │   │   │   if res is not None:                                                     │
│   41 │   │   │   │   │   │   results.append(res)                                                 │
│   42 │   │   │   │   │   │   if firstresult:  # halt further impl calls                          │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/kedro_telemetry/plugin.py:128 in             │
│ after_context_created                                                                            │
│                                                                                                  │
│   125 │   │                                                                                      │
│   126 │   │   logger.debug("You have opted into product usage analytics.")                       │
│   127 │   │                                                                                      │
│ ❱ 128 │   │   catalog = context.catalog                                                          │
│   129 │   │   default_pipeline = pipelines.get("__default__")  # __default__                     │
│   130 │   │   hashed_username = _get_hashed_username()                                           │
│   131                                                                                            │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/kedro/framework/context/context.py:226 in    │
│ catalog                                                                                          │
│                                                                                                  │
│   223 │   │   │   KedroContextError: Incorrect ``DataCatalog`` registered for the project.       │
│   224 │   │                                                                                      │
│   225 │   │   """                                                                                │
│ ❱ 226 │   │   return self._get_catalog()                                                         │
│   227 │                                                                                          │
│   228 │   @property                                                                              │
│   229 │   def params(self) -> dict[str, Any]:                                                    │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/kedro/framework/context/context.py:287 in    │
│ _get_catalog                                                                                     │
│                                                                                                  │
│   284 │   │   feed_dict = self._get_feed_dict()                                                  │
│   285 │   │   catalog.add_feed_dict(feed_dict)                                                   │
│   286 │   │   _validate_transcoded_datasets(catalog)                                             │
│ ❱ 287 │   │   self._hook_manager.hook.after_catalog_created(                                     │
│   288 │   │   │   catalog=catalog,                                                               │
│   289 │   │   │   conf_catalog=conf_catalog,                                                     │
│   290 │   │   │   conf_creds=conf_creds,                                                         │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_hooks.py:265 in __call__             │
│                                                                                                  │
│   262 │   │   else:                                                                              │
│   263 │   │   │   firstresult = False                                                            │
│   264 │   │                                                                                      │
│ ❱ 265 │   │   return self._hookexec(self.name, self.get_hookimpls(), kwargs, firstresult)        │
│   266 │                                                                                          │
│   267 │   def call_historic(self, result_callback=None, kwargs=None):                            │
│   268 │   │   """Call the hook with given ``kwargs`` for all registered plugins and              │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_manager.py:80 in _hookexec           │
│                                                                                                  │
│    77 │   def _hookexec(self, hook_name, methods, kwargs, firstresult):                          │
│    78 │   │   # called from all hookcaller instances.                                            │
│    79 │   │   # enable_tracing will set its own wrapping function at self._inner_hookexec        │
│ ❱  80 │   │   return self._inner_hookexec(hook_name, methods, kwargs, firstresult)               │
│    81 │                                                                                          │
│    82 │   def register(self, plugin, name=None):                                                 │
│    83 │   │   """Register a plugin and return its canonical name or ``None`` if the name         │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_manager.py:327 in traced_hookexec    │
│                                                                                                  │
│   324 │   │   │   │   lambda: oldcall(hook_name, hook_impls, kwargs, firstresult)                │
│   325 │   │   │   )                                                                              │
│   326 │   │   │   after(outcome, hook_name, hook_impls, kwargs)                                  │
│ ❱ 327 │   │   │   return outcome.get_result()                                                    │
│   328 │   │                                                                                      │
│   329 │   │   self._inner_hookexec = traced_hookexec                                             │
│   330                                                                                            │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_result.py:60 in get_result           │
│                                                                                                  │
│   57 │   │   │   return self._result                                                             │
│   58 │   │   else:                                                                               │
│   59 │   │   │   ex = self._excinfo                                                              │
│ ❱ 60 │   │   │   raise ex[1].with_traceback(ex[2])                                               │
│   61                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_result.py:33 in from_call            │
│                                                                                                  │
│   30 │   │   __tracebackhide__ = True                                                            │
│   31 │   │   result = excinfo = None                                                             │
│   32 │   │   try:                                                                                │
│ ❱ 33 │   │   │   result = func()                                                                 │
│   34 │   │   except BaseException:                                                               │
│   35 │   │   │   excinfo = sys.exc_info()                                                        │
│   36                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_manager.py:324 in <lambda>           │
│                                                                                                  │
│   321 │   │   def traced_hookexec(hook_name, hook_impls, kwargs, firstresult):                   │
│   322 │   │   │   before(hook_name, hook_impls, kwargs)                                          │
│   323 │   │   │   outcome = _Result.from_call(                                                   │
│ ❱ 324 │   │   │   │   lambda: oldcall(hook_name, hook_impls, kwargs, firstresult)                │
│   325 │   │   │   )                                                                              │
│   326 │   │   │   after(outcome, hook_name, hook_impls, kwargs)                                  │
│   327 │   │   │   return outcome.get_result()                                                    │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_callers.py:60 in _multicall          │
│                                                                                                  │
│   57 │   │   │   except StopIteration:                                                           │
│   58 │   │   │   │   pass                                                                        │
│   59 │   │                                                                                       │
│ ❱ 60 │   │   return outcome.get_result()                                                         │
│   61                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_result.py:60 in get_result           │
│                                                                                                  │
│   57 │   │   │   return self._result                                                             │
│   58 │   │   else:                                                                               │
│   59 │   │   │   ex = self._excinfo                                                              │
│ ❱ 60 │   │   │   raise ex[1].with_traceback(ex[2])                                               │
│   61                                                                                             │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/pluggy/_callers.py:39 in _multicall          │
│                                                                                                  │
│   36 │   │   │   │   │   except StopIteration:                                                   │
│   37 │   │   │   │   │   │   _raise_wrapfail(gen, "did not yield")                               │
│   38 │   │   │   │   else:                                                                       │
│ ❱ 39 │   │   │   │   │   res = hook_impl.function(*args)                                         │
│   40 │   │   │   │   │   if res is not None:                                                     │
│   41 │   │   │   │   │   │   results.append(res)                                                 │
│   42 │   │   │   │   │   │   if firstresult:  # halt further impl calls                          │
│                                                                                                  │
│ /anaconda/envs/env_name/lib/python3.9/site-packages/kedro_azureml/hooks.py:23 in                 │
│ after_catalog_created                                                                            │
│                                                                                                  │
│   20 │   def after_catalog_created(self, catalog):                                               │
│   21 │   │   for dataset_name, dataset in catalog._data_sets.items():                            │
│   22 │   │   │   if isinstance(dataset, AzureMLAssetDataSet):                                    │
│ ❱ 23 │   │   │   │   dataset.azure_config = self.azure_config                                    │
│   24 │   │   │   │   catalog.add(dataset_name, dataset, replace=True)                            │
│   25 │                                                                                           │
│   26 │   @hook_impl                                                                              │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
AttributeError: 'AzureMLLocalRunHook' object has no attribute 'azure_config'

Update Pydantic Dependency Version

The Pydantic dependency version in this package is quite out of date, with kedro-azureml 0.7.0 depending on pydantic<1.10.0 and >=1.9.1. This is causing issues when working with other packages which have updated their pydantic version to latest (now 2.7.0).

Could this dependency version be updated please?

Is there anyway to debug individual nodes in the pipeline?

I'm trying to recreate a project pipeline similar to the spaceflights tutorial you had on youtube, but I can't even get past the first node. I get this error and is a bit confused about where to start debugging it? In the documentation, I can't see where the plugin creates the azurml components or pipeline files so I can go in and debug what it is trying to do.

Pipeline failing when using MlflowModelLoggerDataSet as an input of a node

I noticed a small issue when integrating with kedro-mlflow, more specifically with kedro_mlflow.io.models.MlflowModelLoggerDataSet when we want to reuse them from one node to another.

Consider the following scenario:

#catalog.yml
mlflow_logger_model:
  type: kedro_mlflow.io.models.MlflowModelLoggerDataSet
  flavor: mlflow.sklearn

# node.py
import numpy as np

from sklearn.dummy import DummyClassifier
def create_mlflow_model_artifact():
    X = np.array([[1],[2]])
    y = np.array([0,1])
    model = DummyClassifier()
    model.fit(X,y)
    return model 

def dummy_predict(model):
    X = np.array([[1],[2]])
    model.predict(X)

#pipeline.py
from kedro.pipeline import Pipeline, node
from .nodes import dummy_predict, create_mlflow_model_artifact


def create_pipeline(**kwargs) -> Pipeline:
    return Pipeline(
        nodes=[
            node(
                func=create_mlflow_model_artifact,
                inputs=None,
                outputs="mlflow_logger_model",
                name="create_mlflow_model_artifact",
            ),
            node(
                func=dummy_predict,
                inputs="mlflow_logger_model",
                outputs=None,
                name="dummy_predict",
            ),
        ]
    )

In this scenario, the dummy classifier is logged to mlflow but can not be used by the second node dummy_predict

DatasetError: Failed while loading data from data set 
MlflowModelLoggerDataSet(artifact_path=model, flavor=mlflow.sklearn, 
load_args={}, save_args={}).
The following failures occurred while downloading one or more artifacts from 
................

We could save the model as an MlflowArtifactDataSet and provide a path instead:

#catalog.yml
mlflow_logger_model:
    type: kedro_mlflow.io.artifacts.MlflowArtifactDataSet
    data_set:
        type: kedro_mlflow.io.models.MlflowModelSaverDataSet
        flavor: mlflow.sklearn
        filepath: data/06_models/sklearn_model

It worked when I launched the pipeline with this configuration. However I suspect it only worked because I had first ran the pipeline locally and the sklearn_model folder existed on my local machine before launching the pipeline.

Maybe the best way is to consider that everything that is tracked by mlflow should not be reused as an input in the pipeline.

kedro-azureml can't find environment version

Hi, I'm using the new version 0.3.1 using the suggested azureml environments feature.
I have successfully completed the steps for creating a docker image and creating an environment.
In my workspace under environment, I can see the created environment

Now, it is important to mention, I am not using VSCode, so I don't have the AzureML plugin installed, rather I am using PyCharm.

When trying to execute kedro azureml run (subscription ID is set as an environment variable), I get the error

ValidationException: Failed to extract version when parsing asset kedrotesting-environment of type environments as arm id. Version must be provided.

When trying to execute kedro azureml run --aml_env kedrotesting-environment, I am getting the same error.

Only when I specify the environment version as well does it work, e.g. kedro azureml run --aml_env kedrotesting-environment:1 works fine.

I assume this is not intended, and the expected behavior is that the latest version of the default environment will be used

Attaching the full stacktrace

stacktrace.txt

How to correctly create uri_file data assets ?

I am trying to save a data asset as a uri_file and the dataset is incorrectly saved as a uri_folder when I launch kedro azureml run

I have the following catalog:

projects_train_raw_local:
  type: pandas.CSVDataSet
  filepath: data/01_raw/dataset.csv

projects_train_raw:
    type: kedro_azureml.datasets.AzureMLAssetDataSet
    azureml_dataset: projects_train_raw
    root_dir: data/00_azurelocals/ 
    versioned: True
    azureml_type: uri_file
    dataset:
        type: pandas.CSVDataSet
        filepath: "dataset.csv"

and the following pipeline which just opens the local file and saves it

def create_pipeline(**kwargs) -> Pipeline:
    return Pipeline(
        nodes=[
            node(
                func=lambda x: x,
                inputs="projects_train_raw_local",
                outputs="projects_train_raw",
                name="create_train_dataasset",
            )
        ]
    )

I expected a new data asset to be created on azure as an uri_file. However, i get the following info on azure

It seems my file is not saved correctly, which seems to correspond to this part in cli.py if I am not mistaken

    # 2. Save dummy outputs
    # In distributed computing, it will only happen on nodes with rank 0
    if not pipeline_data_passing and is_distributed_master_node():
        for data_path in azure_outputs.values():
            (Path(data_path) / "output.txt").write_text("#getindata")
    else:
        logger.info("Skipping saving Azure outputs on non-master distributed nodes")

How can I correctly create a uri_file data asset ?

Enable use of Azure ML Environments and code upload

Thanks for creating this plugin!

When evaluating it for usage in our projects, I was a bit surprised by the approach to use a Docker image with the code baked in, instead of the Azure ML Environments (or Docker images) with code upload on job submission.

The current approach works fine for production deployments, but it doesn't support the experimentation workflow very well. I modified the plugin locally to support submitting jobs using Environments and code snapshotting. Would this change be welcome as a PR? Or would you like to keep this current approach or have both options?

Quickstart example leads to ServiceError on Azure ML

I followed the instructions + video on website but I receive the following error when I submit the pipeline to Azure ML using the command

kedro azureml run

The following is the error that I receive:

Failed to execute command group with error Container `9749aaa32e0c42b189c056da2890de0c-execution-wrapper` failed with status code `1` and it was not possible to extract the structured error Container `9749aaa32e0c42b189c056da2890de0c-execution-wrapper` exited with code 1 due to error None and we couldn't read the error due to GetErrorFromContainerFailed { last_stderr: Some("exec /mnt/azureml/cr/j/9749aaa32e0c42b189c056da2890de0c/cap/lifecycler/wd/execution-wrapper: no such file or directory\n") }.

I am able to execute the pipeline from the image I have built using the following command

kedro docker run

Any help would be greatly appreciated.

Error when running example

Hi thanks for releasing this plugin, I was working on something similar when I came across this on the kedro discord. I was just running the Quickstart and it seems like everything is working until actually running the job. It gets submitted alright but then on the first node I get the following error:

[2022-08-06 11:16:48Z] Job failed, job RunId is xxxxxxxxxxxxxxx. Error: {"Error"\n{"Code":"ServiceError","Severity":null,"Message":"AzureMLCompute job failed.
OrchestrateJobError: Failed to execute command group with error Got CommonRuntimeJobError from executing command: 
Docker container `xxxxxxxx-execution-wrapper` failed with status code `1` due to response error: 
None","MessageFormat":null,"MessageParameters":null,"ReferenceCode":null,"DetailsUri":null,"Target":null,"Details":
[],"InnerError":null,"DebugInfo":null,"AdditionalInfo":null},"Correlation":
{"operation":"xxxxxxx","request":"xxxxx"},"Environment":"switzerlandnorth","Location":"switzerlandnorth","
Time":"2022-08-06T11:16:48.1243444+00:00","ComponentName":"globaljobdispatcher"}

Have you come across this? Is it a configuration mistake on my side in the resource group, storage account or container registry? Or did I forgot something else?

`kedro azureml run` does not work on an AzureML compute instance

I cannot get the getting started guide in the docs to work on an AzureML compute instance.

I have authenticated using: azure login --use-device-code.
When I run the command kedro azureml run I get the following error:

DefaultAzureCredential failed to retrieve a token from the included credentials.                                        
chained.py:100
Attempted credentials:                                                                                                                
EnvironmentCredential: EnvironmentCredential authentication is unavailable. Environment variables are not fully configured.

So it starts to redirect me to an interactive login, but this does not work on a compute instance since the redirection port is not accessible.

Make all Azure-related config overridable via CLI

For consistency it should be possible to override the resource group and workspace name when using the CLI, just like it is possible to override the subscription ID now.

Originally posted by @tomasvanpottelbergh in #41 (comment)

What would be the best way to handle secrets using the plugin? Is it possible to set up a keyvault in the azureml.yml file?

" No such file or directory: 'python'" error when I follow the upload quick start

I have been training the quick start guide.
However, when I follow the "code upload flow" step 9.1 I keep encountering the error: AssetException: Error with code: [Errno 2] No such file or directory: 'python'.
I have doubled check every step in the tutorial but I cannot get it to work. Any ideas about what is going wrong?

Additional info:
I'm using Python 3.9.10.

Full stack trace:

╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_operation_orchestrator.py:257 in _get_code_asset_arm_id                             │
│                                                                                                  │
│   254 │   │   try:                                                                               │
│   255 │   │   │   self._validate_datastore_name(code_asset.path)                                 │
│   256 │   │   │   if register_asset:                                                             │
│ ❱ 257 │   │   │   │   code_asset = self._code_assets.create_or_update(code_asset)                │
│   258 │   │   │   │   return code_asset.id                                                       │
│   259 │   │   │   uploaded_code_asset, _ = _check_and_upload_path(                               │
│   260 │   │   │   │   artifact=code_asset,                                                       │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_code_operations.py:142 in create_or_update                                          │
│                                                                                                  │
│   139 │   │   │   │   │   │   no_personal_data_message=CHANGED_ASSET_PATH_MSG_NO_PERSONAL_DATA   │
│   140 │   │   │   │   │   │   error_category=ErrorCategory.USER_ERROR,                           │
│   141 │   │   │   │   │   )                                                                      │
│ ❱ 142 │   │   │   raise ex                                                                       │
│   143 │                                                                                          │
│   144 │   # @monitor_with_activity(logger, "Code.Get", ActivityType.PUBLICAPI)                   │
│   145 │   def get(self, name: str, version: str) -> Code:                                        │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_code_operations.py:95 in create_or_update                                           │
│                                                                                                  │
│    92 │   │   │   │   │   registry=self._registry_name,                                          │
│    93 │   │   │   │   │   body=get_asset_body_for_registry_storage(self._registry_name, "codes   │
│    94 │   │   │   │   )                                                                          │
│ ❱  95 │   │   │   code, _ = _check_and_upload_path(                                              │
│    96 │   │   │   │   artifact=code, asset_operations=self, sas_uri=sas_uri, artifact_type=Err   │
│    97 │   │   │   )                                                                              │
│    98                                                                                            │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_artifacts/_artifact_utilities.py:401 in _check_and_upload_path                                 │
│                                                                                                  │
│   398 │   │   )                                                                                  │
│   399 │   │   if not path.is_absolute():                                                         │
│   400 │   │   │   path = Path(artifact.base_path, path).resolve()                                │
│ ❱ 401 │   │   uploaded_artifact = _upload_to_datastore(                                          │
│   402 │   │   │   asset_operations._operation_scope,                                             │
│   403 │   │   │   asset_operations._datastore_operation,                                         │
│   404 │   │   │   path,                                                                          │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_artifacts/_artifact_utilities.py:299 in _upload_to_datastore                                   │
│                                                                                                  │
│   296 │   │   ignore_file = get_ignore_file(path)                                                │
│   297 │   if not asset_hash:                                                                     │
│   298 │   │   asset_hash = get_object_hash(path, ignore_file)                                    │
│ ❱ 299 │   artifact = upload_artifact(                                                            │
│   300 │   │   str(path),                                                                         │
│   301 │   │   datastore_operation,                                                               │
│   302 │   │   operation_scope,                                                                   │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_artifacts/_artifact_utilities.py:178 in upload_artifact                                        │
│                                                                                                  │
│   175 │   │   datastore_info = get_datastore_info(datastore_operation, datastore_name)           │
│   176 │   │   storage_client = get_storage_client(**datastore_info)                              │
│   177 │                                                                                          │
│ ❱ 178 │   artifact_info = storage_client.upload(                                                 │
│   179 │   │   local_path,                                                                        │
│   180 │   │   asset_hash=asset_hash,                                                             │
│   181 │   │   show_progress=show_progress,                                                       │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_artifacts/_blob_storage_helper.py:91 in upload                                                 │
│                                                                                                  │
│    88 │   │   │   msg = Fore.GREEN + f"Uploading {formatted_path}"                               │
│    89 │   │   │                                                                                  │
│    90 │   │   │   # warn if large file (> 100 MB)                                                │
│ ❱  91 │   │   │   file_size, _ = get_directory_size(source)                                      │
│    92 │   │   │   file_size_in_mb = file_size / 10**6                                            │
│    93 │   │   │   if file_size_in_mb > 100:                                                      │
│    94 │   │   │   │   module_logger.warning(FILE_SIZE_WARNING)                                   │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_utils/_asset_utils.py:414 in get_directory_size                                                │
│                                                                                                  │
│   411 │   │   │   if not os.path.islink(full_path):                                              │
│   412 │   │   │   │   path_size = os.path.getsize(full_path)                                     │
│   413 │   │   │   else:                                                                          │
│ ❱ 414 │   │   │   │   path_size = os.path.getsize(                                               │
│   415 │   │   │   │   │   os.readlink(convert_windows_path_to_unix(full_path))                   │
│   416 │   │   │   │   )  # ensure we're counting the size of the linked file                     │
│   417 │   │   │   size_list[full_path] = path_size                                               │
│                                                                                                  │
│ /Users/jordismit/.pyenv/versions/3.9.10/lib/python3.9/genericpath.py:50 in getsize               │
│                                                                                                  │
│    47                                                                                            │
│    48 def getsize(filename):                                                                     │
│    49 │   """Return the size of a file, reported by os.stat()."""                                │
│ ❱  50 │   return os.stat(filename).st_size                                                       │
│    51                                                                                            │
│    52                                                                                            │
│    53 def getmtime(filename):                                                                    │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
FileNotFoundError: [Errno 2] No such file or directory: 'python'

During handling of the above exception, another exception occurred:

╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/bin/kedro:8 in <module>                 │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/kedro/frame │
│ work/cli/cli.py:211 in main                                                                      │
│                                                                                                  │
│   208 │   """                                                                                    │
│   209 │   _init_plugins()                                                                        │
│   210 │   cli_collection = KedroCLI(project_path=Path.cwd())                                     │
│ ❱ 211 │   cli_collection()                                                                       │
│   212                                                                                            │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/core. │
│ py:1130 in __call__                                                                              │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/kedro/frame │
│ work/cli/cli.py:139 in main                                                                      │
│                                                                                                  │
│   136 │   │   )                                                                                  │
│   137 │   │                                                                                      │
│   138 │   │   try:                                                                               │
│ ❱ 139 │   │   │   super().main(                                                                  │
│   140 │   │   │   │   args=args,                                                                 │
│   141 │   │   │   │   prog_name=prog_name,                                                       │
│   142 │   │   │   │   complete_var=complete_var,                                                 │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/core. │
│ py:1055 in main                                                                                  │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/core. │
│ py:1657 in invoke                                                                                │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/core. │
│ py:1657 in invoke                                                                                │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/core. │
│ py:1404 in invoke                                                                                │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/core. │
│ py:760 in invoke                                                                                 │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/decor │
│ ators.py:38 in new_func                                                                          │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/click/decor │
│ ators.py:26 in new_func                                                                          │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/kedro_azure │
│ ml/cli.py:184 in run                                                                             │
│                                                                                                  │
│   181 │   ):                                                                                     │
│   182 │   │   az_client = AzureMLPipelinesClient(az_pipeline, subscription_id)                   │
│   183 │   │                                                                                      │
│ ❱ 184 │   │   is_ok = az_client.run(                                                             │
│   185 │   │   │   mgr.plugin_config.azure,                                                       │
│   186 │   │   │   wait_for_completion,                                                           │
│   187 │   │   │   lambda job: click.echo(job.studio_url),                                        │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/kedro_azure │
│ ml/client.py:65 in run                                                                           │
│                                                                                                  │
│   62 │   │   │   │   f"max instances: {cluster.max_instances})"                                  │
│   63 │   │   │   )                                                                               │
│   64 │   │   │                                                                                   │
│ ❱ 65 │   │   │   pipeline_job = ml_client.jobs.create_or_update(                                 │
│   66 │   │   │   │   self.azure_pipeline,                                                        │
│   67 │   │   │   │   experiment_name=config.experiment_name,                                     │
│   68 │   │   │   │   compute=cluster,                                                            │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/core/ │
│ tracing/decorator.py:78 in wrapper_use_tracer                                                    │
│                                                                                                  │
│   75 │   │   │                                                                                   │
│   76 │   │   │   span_impl_type = settings.tracing_implementation()                              │
│   77 │   │   │   if span_impl_type is None:                                                      │
│ ❱ 78 │   │   │   │   return func(*args, **kwargs)                                                │
│   79 │   │   │                                                                                   │
│   80 │   │   │   # Merge span is parameter is set, but only if no explicit parent are passed     │
│   81 │   │   │   if merge_span and not passed_in_parent:                                         │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_job_operations.py:562 in create_or_update                                           │
│                                                                                                  │
│    559 │   │   │   if isinstance(ex, (ValidationException, SchemaValidationError)):              │
│    560 │   │   │   │   log_and_raise_error(ex)                                                   │
│    561 │   │   │   else:                                                                         │
│ ❱  562 │   │   │   │   raise ex                                                                  │
│    563 │                                                                                         │
│    564 │   def _archive_or_restore(self, name: str, is_archived: bool):                          │
│    565 │   │   job_object = self._get_job(name)                                                  │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_job_operations.py:498 in create_or_update                                           │
│                                                                                                  │
│    495 │   │   │   │   self._validate(job, raise_on_failure=True)                                │
│    496 │   │   │                                                                                 │
│    497 │   │   │   # Create all dependent resources                                              │
│ ❱  498 │   │   │   self._resolve_arm_id_or_upload_dependencies(job)                              │
│    499 │   │   │                                                                                 │
│    500 │   │   │   git_props = get_git_properties()                                              │
│    501 │   │   │   # Do not add git props if they already exist in job properties.               │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_job_operations.py:842 in _resolve_arm_id_or_upload_dependencies                     │
│                                                                                                  │
│    839 │   │   :rtype: Job                                                                       │
│    840 │   │   """                                                                               │
│    841 │   │                                                                                     │
│ ❱  842 │   │   self._resolve_arm_id_or_azureml_id(job, self._orchestrators.get_asset_arm_id)     │
│    843 │   │                                                                                     │
│    844 │   │   if isinstance(job, PipelineJob):                                                  │
│    845 │   │   │   # Resolve top-level inputs                                                    │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_job_operations.py:1064 in _resolve_arm_id_or_azureml_id                             │
│                                                                                                  │
│   1061 │   │   elif isinstance(job, AutoMLJob):                                                  │
│   1062 │   │   │   job = self._resolve_arm_id_for_automl_job(job, resolver, inside_pipeline=Fal  │
│   1063 │   │   elif isinstance(job, PipelineJob):                                                │
│ ❱ 1064 │   │   │   job = self._resolve_arm_id_for_pipeline_job(job, resolver)                    │
│   1065 │   │   else:                                                                             │
│   1066 │   │   │   msg = f"Non supported job type: {type(job)}"                                  │
│   1067 │   │   │   raise ValidationException(                                                    │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_job_operations.py:1179 in _resolve_arm_id_for_pipeline_job                          │
│                                                                                                  │
│   1176 │   │                                                                                     │
│   1177 │   │   # Process each component job                                                      │
│   1178 │   │   try:                                                                              │
│ ❱ 1179 │   │   │   self._component_operations._resolve_dependencies_for_pipeline_component_jobs  │
│   1180 │   │   │   │   pipeline_job.component, resolver                                          │
│   1181 │   │   │   )                                                                             │
│   1182 │   │   except ComponentException as e:                                                   │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_component_operations.py:739 in _resolve_dependencies_for_pipeline_component_jobs    │
│                                                                                                  │
│   736 │   │   │   │   │   │   error_category=ErrorCategory.USER_ERROR,                           │
│   737 │   │   │   │   │   )                                                                      │
│   738 │   │   │                                                                                  │
│ ❱ 739 │   │   │   component_cache.resolve_nodes()                                                │
│   740                                                                                            │
│   741                                                                                            │
│   742 def _refine_component(component_func: types.FunctionType) -> Component:                    │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_utils/_cache_utils.py:305 in resolve_nodes                                                     │
│                                                                                                  │
│   302 │   │   # subgraph will be skipped on register_node_for_lazy_resolution when resolving s   │
│   303 │   │   _node_resolution_lock.acquire()                                                    │
│   304 │   │   try:                                                                               │
│ ❱ 305 │   │   │   self._resolve_nodes()                                                          │
│   306 │   │   finally:                                                                           │
│   307 │   │   │   # release lock even if exception happens                                       │
│   308 │   │   │   _node_resolution_lock.release()                                                │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_utils/_cache_utils.py:269 in _resolve_nodes                                                    │
│                                                                                                  │
│   266 │   │   if is_on_disk_cache_enabled():                                                     │
│   267 │   │   │   cache_contents_to_resolve = self._resolve_cache_contents_from_disk(cache_con   │
│   268 │   │                                                                                      │
│ ❱ 269 │   │   self._resolve_cache_contents(cache_contents_to_resolve, resolver=self._resolver)   │
│   270 │   │                                                                                      │
│   271 │   │   self._fill_back_component_to_nodes(dict_of_nodes_to_resolve)                       │
│   272                                                                                            │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /_utils/_cache_utils.py:197 in _resolve_cache_contents                                           │
│                                                                                                  │
│   194 │   │   # multiprocessing need to dump input objects before starting a new process, whic   │
│   195 │   │   # on _AttrDict for now, so put off the concurrent resolution                       │
│   196 │   │   for cache_content in cache_contents_to_resolve:                                    │
│ ❱ 197 │   │   │   cache_content.arm_id = resolver(                                               │
│   198 │   │   │   │   cache_content.component_ref,                                               │
│   199 │   │   │   │   azureml_type=AzureMLResourceType.COMPONENT                                 │
│   200 │   │   │   )                                                                              │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_operation_orchestrator.py:224 in get_asset_arm_id                                   │
│                                                                                                  │
│   221 │   │   │   │   elif azureml_type == AzureMLResourceType.DATA and isinstance(asset, Data   │
│   222 │   │   │   │   │   result = self._get_data_arm_id(asset, register_asset=register_asset)   │
│   223 │   │   │   │   elif azureml_type == AzureMLResourceType.COMPONENT and isinstance(asset,   │
│ ❱ 224 │   │   │   │   │   result = self._get_component_arm_id(asset)                             │
│   225 │   │   │   │   else:                                                                      │
│   226 │   │   │   │   │   msg = "Unsupported azureml type {} for asset: {}"                      │
│   227 │   │   │   │   │   raise ValidationException(                                             │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_operation_orchestrator.py:350 in _get_component_arm_id                              │
│                                                                                                  │
│   347 │   │   via remote call, register the component if necessary, and FILL BACK the            │
│   348 │   │   arm id to component to reduce remote call."""                                      │
│   349 │   │   if not component.id:                                                               │
│ ❱ 350 │   │   │   component._id = self._component.create_or_update(                              │
│   351 │   │   │   │   component, is_anonymous=True, show_progress=self._operation_config.show_   │
│   352 │   │   │   ).id                                                                           │
│   353 │   │   return component.id                                                                │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_component_operations.py:310 in create_or_update                                     │
│                                                                                                  │
│   307 │   │   │   self._validate(component, raise_on_failure=True)                               │
│   308 │   │                                                                                      │
│   309 │   │   # Create all dependent resources                                                   │
│ ❱ 310 │   │   self._resolve_arm_id_or_upload_dependencies(component)                             │
│   311 │   │                                                                                      │
│   312 │   │   name, version = component._get_rest_name_version()                                 │
│   313 │   │   rest_component_resource = component._to_rest_object()                              │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_component_operations.py:481 in _resolve_arm_id_or_upload_dependencies               │
│                                                                                                  │
│   478 │   │   ).get_asset_arm_id                                                                 │
│   479 │   │                                                                                      │
│   480 │   │   # resolve component's code                                                         │
│ ❱ 481 │   │   _try_resolve_code_for_component(component=component, get_arm_id_and_fill_back=ge   │
│   482 │   │   # resolve component's environment                                                  │
│   483 │   │   self._try_resolve_environment_for_component(                                       │
│   484 │   │   │   component=component,                                                           │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_component_operations.py:820 in _try_resolve_code_for_component                      │
│                                                                                                  │
│   817 │   with component._resolve_local_code() as code:                                          │
│   818 │   │   if code is None:                                                                   │
│   819 │   │   │   return                                                                         │
│ ❱ 820 │   │   component.code = get_arm_id_and_fill_back(code, azureml_type=AzureMLResourceType   │
│   821                                                                                            │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_operation_orchestrator.py:216 in get_asset_arm_id                                   │
│                                                                                                  │
│   213 │   │   │   try:                                                                           │
│   214 │   │   │   │   # TODO: once the asset redesign is finished, this logic can be replaced    │
│   215 │   │   │   │   if azureml_type == AzureMLResourceType.CODE and isinstance(asset, Code):   │
│ ❱ 216 │   │   │   │   │   result = self._get_code_asset_arm_id(asset, register_asset=register_   │
│   217 │   │   │   │   elif azureml_type == AzureMLResourceType.ENVIRONMENT and isinstance(asse   │
│   218 │   │   │   │   │   result = self._get_environment_arm_id(asset, register_asset=register   │
│   219 │   │   │   │   elif azureml_type == AzureMLResourceType.MODEL and isinstance(asset, Mod   │
│                                                                                                  │
│ /Users/jordismit/projects/spaceflights-code-upload/.venv/lib/python3.9/site-packages/azure/ai/ml │
│ /operations/_operation_orchestrator.py:275 in _get_code_asset_arm_id                             │
│                                                                                                  │
│   272 │   │   except (MlException, HttpResponseError) as e:                                      │
│   273 │   │   │   raise e                                                                        │
│   274 │   │   except Exception as e:                                                             │
│ ❱ 275 │   │   │   raise AssetException(                                                          │
│   276 │   │   │   │   message=f"Error with code: {e}",                                           │
│   277 │   │   │   │   target=ErrorTarget.ASSET,                                                  │
│   278 │   │   │   │   no_personal_data_message="Error getting code asset",                       │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
AssetException: Error with code: [Errno 2] No such file or directory: 'python'

Result of pip freeze:

``` adal==1.2.7 adlfs==2023.1.0 aiofiles==22.1.0 aiohttp==3.8.3 aiosignal==1.3.1 anyconfig==0.10.1 anyio==3.6.2 appnope==0.1.3 argon2-cffi==21.3.0 argon2-cffi-bindings==21.2.0 arrow==1.2.3 async-timeout==4.0.2 attrs==22.2.0 azure-ai-ml==1.3.0 azure-common==1.1.28 azure-core==1.26.2 azure-datalake-store==0.0.52 azure-identity==1.12.0 azure-mgmt-core==1.3.2 azure-storage-blob==12.14.1 azure-storage-file-datalake==12.9.1 azure-storage-file-share==12.10.1 Babel==2.11.0 backcall==0.2.0 backoff==2.2.1 beautifulsoup4==4.11.2 binaryornot==0.4.4 black==22.12.0 bleach==6.0.0 build==0.10.0 cachetools==4.2.4 certifi==2022.12.7 cffi==1.15.1 chardet==5.1.0 charset-normalizer==2.1.1 click==8.1.3 cloudpickle==2.2.1 colorama==0.4.6 comm==0.1.2 commonmark==0.9.1 cookiecutter==2.1.1 coverage==7.1.0 cryptography==39.0.0 debugpy==1.6.6 decorator==5.1.1 defusedxml==0.7.1 dynaconf==3.1.11 et-xmlfile==1.1.0 fastapi==0.85.2 fastjsonschema==2.16.2 flake8==4.0.1 frozenlist==1.3.3 fsspec==2022.11.0 gitdb==4.0.10 GitPython==3.1.30 graphql-core==3.2.3 h11==0.14.0 httptools==0.5.0 idna==3.4 importlib-metadata==6.0.0 importlib-resources==5.10.2 iniconfig==2.0.0 ipykernel==6.20.2 ipython==7.34.0 ipython-genutils==0.2.0 ipywidgets==8.0.4 isodate==0.6.1 isort==5.12.0 jedi==0.18.2 Jinja2==3.1.2 jinja2-time==0.2.0 jmespath==0.10.0 joblib==1.2.0 json5==0.9.11 jsonschema==4.17.3 jupyter==1.0.0 jupyter-console==6.4.4 jupyter-server==1.23.5 jupyter_client==8.0.2 jupyter_core==5.2.0 jupyterlab==3.5.3 jupyterlab-pygments==0.2.2 jupyterlab-widgets==3.0.5 jupyterlab_server==2.15.2 kedro==0.18.4 kedro-azureml==0.3.4 kedro-datasets==1.0.1 kedro-docker==0.3.1 kedro-viz==5.2.1 MarkupSafe==2.1.2 marshmallow==3.19.0 matplotlib-inline==0.1.6 mccabe==0.6.1 mistune==2.0.4 msal==1.21.0 msal-extensions==1.0.0 msrest==0.7.1 multidict==6.0.4 mypy-extensions==0.4.3 nbclassic==0.5.1 nbclient==0.7.2 nbconvert==7.2.9 nbformat==5.7.3 nbstripout==0.6.1 nest-asyncio==1.5.6 networkx==3.0 notebook==6.5.2 notebook_shim==0.2.2 numpy==1.24.1 oauthlib==3.2.2 openpyxl==3.1.0 orjson==3.8.5 packaging==23.0 pandas==1.5.3 pandocfilters==1.5.0 parso==0.8.3 pathspec==0.11.0 pexpect==4.8.0 pickleshare==0.7.5 pip-tools==6.12.2 platformdirs==2.6.2 plotly==5.13.0 pluggy==1.0.0 portalocker==2.7.0 prometheus-client==0.16.0 prompt-toolkit==3.0.36 psutil==5.9.4 ptyprocess==0.7.0 py==1.11.0 pyarrow==6.0.1 pycodestyle==2.8.0 pycparser==2.21 pydantic==1.9.2 pydash==5.1.2 pyflakes==2.4.0 Pygments==2.14.0 PyJWT==2.6.0 pyproject_hooks==1.0.0 pyrsistent==0.19.3 pytest==6.2.5 pytest-cov==3.0.0 pytest-mock==1.13.0 python-dateutil==2.8.2 python-dotenv==0.21.1 python-slugify==8.0.0 pytoolconfig==1.2.5 pytz==2022.7.1 PyYAML==6.0 pyzmq==25.0.0 qtconsole==5.4.0 QtPy==2.3.0 requests==2.28.2 requests-oauthlib==1.3.1 rich==12.6.0 rope==1.5.1 scikit-learn==1.2.1 scipy==1.10.0 semver==2.13.0 Send2Trash==1.8.0 six==1.16.0 smmap==5.0.0 sniffio==1.3.0 soupsieve==2.3.2.post1 SQLAlchemy==1.4.46 starlette==0.20.4 strawberry-graphql==0.155.3 strictyaml==1.6.2 tenacity==8.1.0 terminado==0.17.1 text-unidecode==1.3 threadpoolctl==3.1.0 tinycss2==1.2.1 toml==0.10.2 tomli==2.0.1 toposort==1.9 tornado==6.2 tqdm==4.64.1 traitlets==5.9.0 typing_extensions==4.4.0 urllib3==1.26.14 uvicorn==0.20.0 uvloop==0.17.0 watchfiles==0.18.1 watchgod==0.8.2 wcwidth==0.2.6 webencodings==0.5.1 websocket-client==1.5.0 websockets==10.4 widgetsnbextension==4.0.5 yarl==1.8.2 zipp==3.12.0

</details>

AzureMLPipelineDataSet preview of CSV does not work

@tomasvanpottelbergh this is related to the AzureMLPipelineDataSet you've implemented.

Issue

When saving intermediate data as pandas.CSV with pipeline_data_passing feature enabled, the data is indeed saved properly, but the preview does not work - it shows up as a base64-encoded blob (See the screenshot).
I'm suspecting that something has changed in the Azure's SDK that ends up with the file being recognized as blob rather that CSV.

Entry in the catalog I'm using:

preprocessed_shuttles:
  type: kedro_azureml.datasets.AzureMLPipelineDataSet
  dataset:
    type: pandas.CSVDataSet
    filepath: "shuttles.csv"

Preview:

Upgrade azure pipelines SDK to v2

Update Azure pipelines SDK to version 2.

Reference docs - https://learn.microsoft.com/en-us/azure/machine-learning/migrate-to-v2-execution-pipeline?view=azureml-api-2

AzureMLPipelineDataSet not compatible with pipeline_ml_factory method from kedro-mlflow

The pipeline_ml_factory method in kedro-mlflow is a useful method to store artifacts (transformers, models) automatically (using kedro-mlflow hook). However, this method calls the method extract_pipeline_artifacts which requires the _filepath attribute to be available (see here).
AzureMLPipelineDataSet class does not provide this attribute.
Wouldn't it be possible to add it to the class attributes?
Do you have any other suggestion to store the Mlflow Pipeline ?

SonarCloud error after merging node/compute feature

@fdroessler
our SonarCloud code analysis detected the following issue after merging your PR: #13 .

please fix it in a separate PR.

Add support for load-version/load-versions

I run multiple time the same pipeline on AzureML (multiple experiments testing different parameters at the same time). My pipeline is as follow :

Unversionned dataset is used as input in function x, which generates a versionned dataset
That versionned dataset is used as input in function y

Another way of looking at it, it looks like this :
unversionned dataset -> function x -> versionned dataset -> function y

I have the following timeline :

Starts Run A, node x
Starts Run B, node x
Run B node x finishes, saves a versionned dataset
Run B node y starts with versionned dataset from step 3 (as expected)
Run A node x finishes, saves a versionned dataset
Run A node y starts with versionned dataset from step 3 (unexpected, should start with versionned dataset generated at step 5)

The issue seems to stem from how the version ID is generated/transfered. Instead of having a single version ID for the run, and transfering that version ID from node to node, each node generates its own version ID (or fetch the latest one). That bug seems to happen with kedro also, but is easily fixed if I specify "--load-versions" in the CLI (which I can't with kedro azureml). If I was able to specify --load-versions, I believe this bug would disappear (or I could manage to work around it). Right now, I can't run the same pipeline in parallel with different parameters, I'd need to create a new pipeline with a different name (which doesn't make sense at all, I'm just running multiple experiment in parallel).

Consider porting grouping feature from kedro-vertexai

Adding as a reminder of things to do/consider

/home/kedro_docker: No such file or directory

Hello, team!

I am closely follow instructions for the docker image flow in https://kedro-azureml.readthedocs.io. I was able to create an Azure pipeline, but it failed with the error in std_log.txt: /bin/bash: line 1: cd: /home/kedro_docker: No such file or directory. This is the working directory configured in conf/base/azureml.yml, and it’s generated by kedro-azureml. I am not sure why it pointed a non-existing directory, and how to make it works.

distributed_job decorator will not accept the shm_size (shared memory parameter)

@distributed_job(Framework.MPI, num_nodes=2, processes_per_node=1)

When using MPI framework and using more nodes in azureml the shared memory (shm_size) is only 2gb by default,
In my usecase, It is giving the memory issue.

I want to increase the shm_size, which was not possible with kedro azureml

But we can pass the shm_size, If we use normal azureml sdk v2 without kedro, while creating the job in the job schema
attaching the example job schema

If I want to have the same in kedro azureml, What is the way?

Quick Start Guide Code Flow Does Not Seem to Work

I can't seem to figure out this quick start guide. I have spent several hours troubleshooting and figuring out why the code flow does not work per the guide and I ran into several issues along the way. I believe most of my issues currently stem from the azure-cli not being specified, but I think Azure also changed how they parse schemas on their end? So far I have ran into these issues:

The guide doesn't mention you need azure-cli installed and what version of it. I tried latest version and azure-cli~=2.50 but I stull run into the same bug.
Azure CLI did not seem to come with the az ml extension by default at least when I installed the latest version. I had to add it in with an az extensions add command.
az ml requires specifying either --workspace ( and --resource-group), or --registry-name when creating an aml environment, even though the guide does not mention any of that.
After getting though all of that, I was finally stopped in my tracks after running into this bug when I initiate a kedro azureml run. The bug seems to stem from a parsing error where a "/" is missing from the "environment_name" created during initialization. However, adding a "/" to the environment name does not remedy the bug.

Am I missing something?

I was not able to use the PartitionedDataSet as the AzureMLPipelineDataset

In my usecase, one of the node returns a PartitionedDataSet that dataset need to be accessed by the following node. But I was able use AzureMLPipelineDataset with single dataframe but not PartitionedDataSet.
As I want to use it for the parallel_processing...

downloaded_station_data:
type: PartitionedDataset
path: data/01_raw/downloaded_station_data
dataset: pandas.CSVDataset
filename_suffix: .csv

I tried this way:
downloaded_station_data:
type: kedro_azureml.datasets.AzureMLPipelineDataset
dataset:
type: PartitionedDataset
path: data/01_raw/downloaded_station_data
dataset: pandas.CSVDataset
filename_suffix: .csv

This need to be used as the azureml pipeline data using AzureMLPipelineDataset.
I was getting the error as [DatasetError: filepath]

If anyone has the solution please help me...

Docker build fail with useradd: UID 0 is not unique

Hi Guys, hope anyone can help on that. When I follow the quick start to build docker, it fails with below error:

Step 7/14 : RUN groupadd -f -g ${KEDRO_GID} -o kedro_group && useradd -d /home/kedro -s /bin/bash -g ${KEDRO_GID} -u ${KEDRO_UID} kedro ---> Running in cd4c32215333 **useradd: UID 0 is not unique** The command '/bin/sh -c groupadd -f -g ${KEDRO_GID} -o kedro_group && useradd -d /home/kedro -s /bin/bash -g ${KEDRO_GID} -u ${KEDRO_UID} kedro' returned a non-zero code: 4

BTW, I'm using root account in Cent OS to run the command.

Thanks!

getindata / kedro-azureml Goto Github PK

kedro-azureml's Introduction

Kedro Azure ML Pipelines plugin

We help companies turn their data into assets

About

Documentation

Usage guide

Quickstart

kedro-azureml's People

Contributors

Stargazers

Watchers

Forkers

kedro-azureml's Issues

Issue

Steps to reproduce:

Expected result

Actual result

Steps to reproduce:

Full stack trace

Issue

Recommend Projects

Recommend Topics

Recommend Org