Управление входными и выходными данными компонента и конвейера

В этой статье раскрываются следующие темы.

• Обзор входных и выходных данных в компоненте и конвейере
• Повышение уровня входных и выходных данных компонентов для конвейера входных и выходных данных
• Определение необязательных входных данных
• Настройка пути вывода
• Скачивание выходных данных
• Регистрация выходных данных в качестве именованного ресурса

Обзор входных и выходных данных

Машинное обучение Azure конвейеры поддерживают входные и выходные данные как на уровне компонента, так и на уровне конвейера.

На уровне компонента входные и выходные данные определяют интерфейс компонента. Выходные данные одного компонента можно использовать в качестве входных данных для другого компонента в одном родительском конвейере, что позволяет передавать данные или модели между компонентами. Эта взаимосоединение формирует граф, иллюстрируя поток данных в конвейере.

На уровне конвейера входные и выходные данные полезны для отправки заданий конвейера с различными входными данными или параметрами, которые управляют логикой обучения (например learning_rate). Они особенно полезны при вызове конвейера через конечную точку REST. Эти входные и выходные данные позволяют назначать различные значения входным данным конвейера или получать доступ к выходным данным заданий конвейера через конечную точку REST. Дополнительные сведения см. в статье "Создание заданий и входных данных" для конечной точки пакетной службы.

Типы входных и выходных данных

Следующие типы поддерживаются как выходные данные компонента или конвейера.

• типы данных; Проверьте типы данных в Машинное обучение Azure, чтобы узнать больше о типах данных.

  • uri_file
  • uri_folder
  • mltable

• Типы моделей.

  • mlflow_model
  • custom_model

Использование выходных данных или моделей по сути сериализует выходные данные и сохраняет их в виде файлов в расположении хранилища. В последующих шагах это расположение хранилища можно подключить, скачать или отправить в целевую файловую систему вычислений, что позволяет следующему шагу получить доступ к файлам во время выполнения задания.

Для этого процесса требуется исходный код компонента, сериализующий нужный выходной объект ( обычно хранящийся в памяти) в файлы. Например, можно сериализовать кадр данных Pandas в виде CSV-файла. Обратите внимание, что Машинное обучение Azure не определяет стандартные методы для сериализации объектов. В качестве пользователя вы можете выбрать предпочтительный метод сериализации объектов в файлы. После этого в нижнем компоненте можно независимо десериализировать и читать эти файлы. Ниже приведены несколько примеров для справки:

• В примере nyc_taxi_data_regression компонент подготовки имеет выходныеuri_folder данные типа. В исходном коде компонента он считывает CSV-файлы из входной папки, обрабатывает файлы и записывает обработанные CSV-файлы в выходную папку.
• В nyc_taxi_data_regression примере компонент обучения имеет выходные mlflow_model данные типа. В исходном коде компонента он сохраняет обученную модель с помощью mlflow.sklearn.save_model метода.

Помимо указанных выше типов данных или моделей, входные данные конвейера или компонента также могут быть следующими примитивными типами .

• string
• number
• integer
• boolean

В nyc_taxi_data_regression примере компонент обучения имеет входные number данныеtest_split_ratio.

Примечание.

Выходные данные примитивных типов не поддерживаются.

Путь и режим для входных и выходных данных

Для входных и выходных данных ресурса данных необходимо указать параметр, указывающий path на расположение данных. В этой таблице показаны различные расположения данных, поддерживаемые конвейером Машинное обучение Azure, а также примеры параметров пути:

Местонахождение	Примеры	Входные данные	Выходные данные
Путь к локальному компьютеру	./home/username/data/my_data	✓
Путь к общедоступному HTTP(S)-серверу	https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv	✓
Путь к службе хранилища Azure	wasbs://<container_name>@<account_name>.blob.core.windows.net/<path> abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>	Не рекомендуется, так как для чтения данных может потребоваться дополнительная конфигурация удостоверений.
Путь к хранилищу данных Машинное обучение Azure	azureml://datastores/<data_store_name>/paths/<path>	✓	✓
Путь к ресурсу данных	azureml:<my_data>:<version>	✓	✓

Примечание.

Для входных и выходных данных в хранилище мы настоятельно рекомендуем использовать путь к хранилищу Машинное обучение Azure вместо прямого служба хранилища Azure пути. Путь к хранилищу данных поддерживается в различных типах заданий в конвейере.

Для ввода и вывода данных можно выбрать различные режимы (скачивание, подключение или отправка), чтобы определить способ доступа к данным в целевом объекте вычислений. В этой таблице показаны возможные режимы для различных сочетаний типов,ввода и вывода.

Тип	Ввод-вывод	upload	download	ro_mount	rw_mount	direct	eval_download	eval_mount
uri_folder	Входные данные		✓	✓		✓
uri_file	Входные данные		✓	✓		✓
mltable	Входные данные		✓	✓		✓	✓	✓
uri_folder	Выходные данные	✓			✓
uri_file	Выходные данные	✓			✓
mltable	Выходные данные	✓			✓	✓

Примечание.

В большинстве случаев рекомендуется использовать ro_mount или rw_mount режим. Дополнительные сведения о режиме см. в режимах ресурса данных.

Визуальное представление в Студия машинного обучения Azure

На следующих снимках экрана приведен пример отображения входных и выходных данных в задании конвейера в Студия машинного обучения Azure. Это конкретное задание с именем nyc-taxi-data-regressionможно найти в azureml-example.

На странице задания конвейера студии входные и выходные данные компонента типа данных/модели отображаются в виде небольшого круга в соответствующем компоненте, известном как порт ввода и вывода. Эти порты представляют поток данных в конвейере.

Выходные данные уровня конвейера отображаются в виде фиолетового поля для простой идентификации.

При наведении указателя мыши на порт ввода и вывода отображается тип.

Входные данные примитивных типов не отображаются на графе. Его можно найти на вкладке Параметры панели обзора задания конвейера (для входных данных уровня конвейера) или на панели компонентов (для входных данных на уровне компонента). На снимка экрана показана вкладка Параметры задания конвейера, которую можно открыть, выбрав ссылку "Обзор задания".

Если вы хотите проверка входные данные для компонента, дважды щелкните компонент, чтобы открыть панель компонентов.

Аналогичным образом при редактировании конвейера в конструкторе можно найти входные и выходные данные конвейера на панели интерфейса конвейера, а также входные и выходные данные компонента на панели компонента (триггер, дважды щелкнув компонент).

Повышение уровня входных и выходных данных компонентов на уровне конвейера

Повышение входных и выходных данных компонента на уровне конвейера позволяет перезаписать входные и выходные данные компонента при отправке задания конвейера. Кроме того, полезно активировать конвейер с помощью конечной точки REST.

Ниже приведены примеры для повышения входных и выходных данных компонентов до входных и выходных данных на уровне конвейера.

Azure CLI

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline
display_name: 1b_e2e_registered_components
description: E2E dummy train-score-eval pipeline with registered components

inputs:
  pipeline_job_training_max_epocs: 20
  pipeline_job_training_learning_rate: 1.8
  pipeline_job_learning_rate_schedule: 'time-based'

outputs: 
  pipeline_job_trained_model:
    mode: upload
  pipeline_job_scored_data:
    mode: upload
  pipeline_job_evaluation_report:
    mode: upload

settings:
 default_compute: azureml:cpu-cluster

jobs:
  train_job:
    type: command
    component: azureml:my_train@latest
    inputs:
      training_data: 
        type: uri_folder 
        path: ./data      
      max_epocs: ${{parent.inputs.pipeline_job_training_max_epocs}}
      learning_rate: ${{parent.inputs.pipeline_job_training_learning_rate}}
      learning_rate_schedule: ${{parent.inputs.pipeline_job_learning_rate_schedule}}
    outputs:
      model_output: ${{parent.outputs.pipeline_job_trained_model}}
    services:
      my_vscode:
        type: vs_code
      my_jupyter_lab:
        type: jupyter_lab
      my_tensorboard:
        type: tensor_board
        log_dir: "outputs/tblogs"
    #  my_ssh:
    #    type: tensor_board
    #    ssh_public_keys: <paste the entire pub key content>
    #    nodes: all # Use the `nodes` property to pick which node you want to enable interactive services on. If `nodes` are not selected, by default, interactive applications are only enabled on the head node.

  score_job:
    type: command
    component: azureml:my_score@latest
    inputs:
      model_input: ${{parent.jobs.train_job.outputs.model_output}}
      test_data: 
        type: uri_folder 
        path: ./data
    outputs:
      score_output: ${{parent.outputs.pipeline_job_scored_data}}

  evaluate_job:
    type: command
    component: azureml:my_eval@latest
    inputs:
      scoring_result: ${{parent.jobs.score_job.outputs.score_output}}
    outputs:
      eval_output: ${{parent.outputs.pipeline_job_evaluation_report}}

Полный пример можно найти в конвейере train-score-eval с зарегистрированными компонентами. Этот конвейер обеспечивает три входных и три выходных данных на уровне конвейера. Давайте рассмотрим pipeline_job_training_max_epocs как пример. Он объявлен в разделе inputs на корневом уровне, что означает входные данные на уровне конвейера. В jobs -> train_job разделе указано max_epocs${{parent.inputs.pipeline_job_training_max_epocs}}имя входных данных, которое указывает train_job, что входные ссылки на входные max_epocspipeline_job_training_max_epocsданные уровня конвейера. Аналогичным образом можно повысить выходные данные конвейера с помощью той же схемы.

Пакет SDK для Python

# import required libraries
from azure.identity import DefaultAzureCredential

from azure.ai.ml import MLClient, Input
from azure.ai.ml.dsl import pipeline
from azure.ai.ml import load_component

# Set your subscription, resource group and workspace name:
subscription_id = "<SUBSCRIPTION_ID>"
resource_group = "<RESOURCE_GROUP>"
workspace = "<AML_WORKSPACE_NAME>"

# connect to the AzureML workspace
ml_client = MLClient(
    DefaultAzureCredential(), subscription_id, resource_group, workspace
)

# define the dirtory that stores the input data 
parent_dir = ""

# Load components
prepare_data = load_component(source=parent_dir + "./prep.yml")
transform_data = load_component(source=parent_dir + "./transform.yml")
train_model = load_component(source=parent_dir + "./train.yml")
predict_result = load_component(source=parent_dir + "./predict.yml")
score_data = load_component(source=parent_dir + "./score.yml")

# Construct pipeline. 
# Below code snippet defines nyc_taxi_data_regression pipeline.
# The pipeline takes 1 input (pipeline_job_input) and generates 6 outputs as defined in return statement.
# The pipeline outputs are promoted from the child component using schema as <step_name.outputs.output_name>.
# for example `prepare_sample_data.outputs.prep_data`.  
@pipeline()
def nyc_taxi_data_regression(pipeline_job_input):
    """NYC taxi data regression example."""
    prepare_sample_data = prepare_data(raw_data=pipeline_job_input)
    transform_sample_data = transform_data(
        clean_data=prepare_sample_data.outputs.prep_data
    )
    train_with_sample_data = train_model(
        training_data=transform_sample_data.outputs.transformed_data
    )
    predict_with_sample_data = predict_result(
        model_input=train_with_sample_data.outputs.model_output,
        test_data=train_with_sample_data.outputs.test_data,
    )
    score_with_sample_data = score_data(
        predictions=predict_with_sample_data.outputs.predictions,
        model=train_with_sample_data.outputs.model_output,
    )
    return {
        "pipeline_job_prepped_data": prepare_sample_data.outputs.prep_data,
        "pipeline_job_transformed_data": transform_sample_data.outputs.transformed_data,
        "pipeline_job_trained_model": train_with_sample_data.outputs.model_output,
        "pipeline_job_test_data": train_with_sample_data.outputs.test_data,
        "pipeline_job_predictions": predict_with_sample_data.outputs.predictions,
        "pipeline_job_score_report": score_with_sample_data.outputs.score_report,
    }
# 
pipeline_job = nyc_taxi_data_regression(
    Input(type="uri_folder", path=parent_dir + "./data/")
)
# demo how to change pipeline output settings
pipeline_job.outputs.pipeline_job_prepped_data.mode = "rw_mount"

# set pipeline level compute
pipeline_job.settings.default_compute = "cpu-cluster"
# set pipeline level datastore
pipeline_job.settings.default_datastore = "workspaceblobstore"

Пример сквозной записной книжки в репозитории azureml-example

Студия

Входные данные компонента можно повысить до входных данных на уровне конвейера на странице разработки конструктора. Перейдите на панель параметров компонента, дважды щелкнув компонент .> Найдите входные данные, которые вы хотите повысить.> Выберите три точки справа .> Выберите "Добавить в входные данные конвейера".

Необязательные входные данные

По умолчанию все входные данные являются обязательными и должны быть назначены значение (или значение по умолчанию) при каждом отправке задания конвейера. Однако могут быть экземпляры, в которых требуются необязательные входные данные. В таких случаях у вас есть гибкость, чтобы не назначать входные данные при отправке задания конвейера.

Необязательные входные данные могут быть полезны в следующих двух сценариях:

• Если у вас есть необязательные входные данные или тип модели и не назначаете ему значение при отправке задания конвейера, в конвейере будет компонент, который не имеет предыдущей зависимости данных. Другими словами, входной порт не связан с каким-либо компонентом или узлом модели или данными. Это приводит к тому, что служба конвейера напрямую вызывает этот компонент, а не ожидает готовности предыдущей зависимости.

• Снимок экрана ниже содержит четкий пример второго сценария. Если для конвейера установлен continue_on_step_failure = True второй узел (node2), который использует выходные данные первого узла (node1) в качестве необязательных входных данных, node2 по-прежнему будет выполняться, даже если узел1 завершается ошибкой. Однако если узел 2 использует необходимые входные данные из node1, он не будет выполнен, если узел1 завершается ошибкой.

  

Ниже приведены примеры определения необязательных входных данных.

$schema: https://azuremlschemas.azureedge.net/latest/commandComponent.schema.json
name: train_data_component_cli
display_name: train_data
description: A example train component
tags:
  author: azureml-sdk-team
type: command
inputs:
  training_data: 
    type: uri_folder
  max_epocs:
    type: integer
    optional: true
  learning_rate: 
    type: number
    default: 0.01
    optional: true
  learning_rate_schedule: 
    type: string
    default: time-based
    optional: true
outputs:
  model_output:
    type: uri_folder
code: ./train_src
environment: azureml://registries/azureml/environments/sklearn-1.0/labels/latest
command: >-
  python train.py 
  --training_data ${{inputs.training_data}} 
  $[[--max_epocs ${{inputs.max_epocs}}]]
  $[[--learning_rate ${{inputs.learning_rate}}]]
  $[[--learning_rate_schedule ${{inputs.learning_rate_schedule}}]]
  --model_output ${{outputs.model_output}}

Если входные данные заданы как optional = true, необходимо использовать для $[[]] применения командной строки с входными данными. См. выделенные строки в приведенном выше примере.

Примечание.

Необязательные выходные данные не поддерживаются.

В графе конвейера необязательные входные данные типа Data/Model представлены точками круга. Необязательные входные данные примитивных типов можно найти на вкладке Параметры. В отличие от обязательных входных данных, необязательные входные данные не имеют звездочки рядом с ними, что означает, что они не являются обязательными.

Настройка выходного пути

По умолчанию выходные данные компонента будут храниться в azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}. Клиент {default_datastore} хранилища данных по умолчанию настроен для конвейера. Если не задано хранилище BLOB-объектов рабочей области. Это {name} имя задания, которое будет разрешено во время выполнения задания. Это {output_name} имя выходных данных, определенное в компоненте YAML.

Но вы также можете настроить место хранения выходных данных, определив путь к выходным данным. Ниже приведен пример:

Azure CLI

Определяет pipeline.yaml конвейер с тремя выходными данными уровня конвейера. Полный YAML можно найти в конвейере train-score-eval с примером зарегистрированных компонентов. Для задания пользовательского пути вывода для выходных pipeline_job_trained_modelданных можно использовать следующую команду.

# define the custom output path using datastore uri
# add relative path to your blob container after "azureml://datastores/<datastore_name>/paths"
output_path="azureml://datastores/{datastore_name}/paths/{relative_path_of_container}"  

# create job and define path using --outputs.<outputname>
az ml job create -f ./pipeline.yml --set outputs.pipeline_job_trained_model.path=$output_path  

Пакет SDK для Python

cluster_name = "cpu-cluster"
custom_path = "azureml://datastores/workspaceblobstore/paths/custom_path/${{name}}/"

# define a pipeline with component
@pipeline(default_compute=cluster_name)
def pipeline_with_python_function_components(input_data, test_data, learning_rate):
    """E2E dummy train-score-eval pipeline with components defined via python function components"""

    # Call component obj as function: apply given inputs & parameters to create a node in pipeline
    train_with_sample_data = train_model(
        training_data=input_data, max_epochs=5, learning_rate=learning_rate
    )
    score_with_sample_data = score_data(
        model_input=train_with_sample_data.outputs.model_output,
        test_data=test_data,
        model_file=train_with_sample_data.outputs.output,
    )
    # example how to change path of output on step level,
    # please note if the output is promoted to pipeline level you need to change path in pipeline job level
    score_with_sample_data.outputs.score_output = Output(
        type="uri_folder", mode="rw_mount", path=custom_path
    )
    eval_with_sample_data = eval_model(
        scoring_result=score_with_sample_data.outputs.score_output,
        scoring_file=score_with_sample_data.outputs.output,
    )

    # Return: pipeline outputs
    return {
        "eval_output": eval_with_sample_data.outputs.eval_output,
        "model_output": train_with_sample_data.outputs.model_output,
    }


pipeline_job = pipeline_with_python_function_components(
    input_data=Input(
        path="wasbs://demo@dprepdata.blob.core.windows.net/Titanic.csv", type="uri_file"
    ),
    test_data=Input(
        path="wasbs://demo@dprepdata.blob.core.windows.net/Titanic.csv", type="uri_file"
    ),
    learning_rate=0.1,
)
# example how to change path of output on pipeline level
pipeline_job.outputs.model_output = Output(
    type="uri_folder", mode="rw_mount", path=custom_path
)

Полный пример записной книжки можно найти в конвейере сборки с command_component украшенной записной книжкой функции Python.

Скачивание выходных данных

Вы можете скачать выходные данные компонента или выходные данные конвейера ниже.

Скачивание выходных данных уровня конвейера

Azure CLI

# Download all the outputs of the job
az ml job download --all -n <JOB_NAME> -g <RESOURCE_GROUP_NAME> -w <WORKSPACE_NAME> --subscription <SUBSCRIPTION_ID>

# Download specific output
az ml job download --output-name <OUTPUT_PORT_NAME> -n <JOB_NAME> -g <RESOURCE_GROUP_NAME> -w <WORKSPACE_NAME> --subscription <SUBSCRIPTION_ID>

Пакет SDK для Python

Прежде чем мы рассмотрим код, вам потребуется способ ссылаться на рабочую область. Вы создаете ml_client дескриптор для рабочей области. Сведения о создании дескриптора рабочей области для инициализации ml_client.

# Download all the outputs of the job
output = client.jobs.download(name=job.name, download_path=tmp_path, all=True)

# Download specific output
output = client.jobs.download(name=job.name, download_path=tmp_path, output_name=output_port_name)

Скачивание выходных данных дочернего задания

Когда необходимо скачать выходные данные дочернего задания (выходные данные компонента, которые не способствуют уровню конвейера), сначала следует перечислить все дочерние сущности задания конвейера, а затем использовать аналогичный код для скачивания выходных данных.

Azure CLI

# List all child jobs in the job and print job details in table format
az ml job list --parent-job-name <JOB_NAME> -g <RESOURCE_GROUP_NAME> -w <WORKSPACE_NAME> --subscription <SUBSCRIPTION_ID> -o table

# Select needed child job name to download output
az ml job download --all -n <JOB_NAME> -g <RESOURCE_GROUP_NAME> -w <WORKSPACE_NAME> --subscription <SUBSCRIPTION_ID>

Пакет SDK для Python

Прежде чем мы рассмотрим код, вам потребуется способ ссылаться на рабочую область. Вы создаете ml_client дескриптор для рабочей области. Сведения о создании дескриптора рабочей области для инициализации ml_client.

# List all child jobs in the job
child_jobs = client.jobs.list(parent_job_name=job.name)
# Traverse and download all the outputs of child job
for child_job in child_jobs:
    client.jobs.download(name=child_job.name, all=True)

Регистрация выходных данных в качестве именованного ресурса

Вы можете зарегистрировать выходные данные компонента или конвейера в качестве именованного ресурса, назначив name и version выходные данные. Зарегистрированный ресурс может быть указан в рабочей области с помощью пользовательского интерфейса студии или интерфейса командной строки или пакета SDK, а также ссылаться на ваши будущие задания.

Регистрация выходных данных конвейера

Azure CLI

display_name: register_pipeline_output
type: pipeline
jobs:
  node:
    type: command
    inputs:
      component_in_path:
        type: uri_file
        path: https://dprepdata.blob.core.windows.net/demo/Titanic.csv
    component: ../components/helloworld_component.yml
    outputs:
      component_out_path: ${{parent.outputs.component_out_path}}
outputs:
  component_out_path:
    type: mltable
    name: pipeline_output  # Define name and version to register pipeline output
    version: '1'
settings:
  default_compute: azureml:cpu-cluster

Пакет SDK для Python

from azure.ai.ml import dsl, Output

# Load component functions
components_dir = "./components/"
helloworld_component = load_component(source=f"{components_dir}/helloworld_component.yml")

@pipeline()
def register_pipeline_output():
  # Call component obj as function: apply given inputs & parameters to create a node in pipeline
  node = helloworld_component(component_in_path=Input(
    type='uri_file', path='https://dprepdata.blob.core.windows.net/demo/Titanic.csv'))

  return {
      'component_out_path': node.outputs.component_out_path
  }

pipeline = register_pipeline_output()
# Define name and version to register pipeline output
pipeline.settings.default_compute = "azureml:cpu-cluster"
pipeline.outputs.component_out_path.name = 'pipeline_output'
pipeline.outputs.component_out_path.version = '1'

Регистрация выходных данных дочернего задания

Azure CLI

display_name: register_node_output
type: pipeline
jobs:
  node:
    type: command
    component: ../components/helloworld_component.yml
    inputs:
      component_in_path:
        type: uri_file
        path: 'https://dprepdata.blob.core.windows.net/demo/Titanic.csv'
    outputs:
      component_out_path:
        type: uri_folder
        name: 'node_output'  # Define name and version to register a child job's output
        version: '1'
settings:
  default_compute: azureml:cpu-cluster

Пакет SDK для Python

from azure.ai.ml import dsl, Output

# Load component functions
components_dir = "./components/"
helloworld_component = load_component(source=f"{components_dir}/helloworld_component.yml")

@pipeline()
def register_node_output():
  # Call component obj as function: apply given inputs & parameters to create a node in pipeline
  node = helloworld_component(component_in_path=Input(
    type='uri_file', path='https://dprepdata.blob.core.windows.net/demo/Titanic.csv'))

  # Define name and version to register node output
  node.outputs.component_out_path.name = 'node_output'
  node.outputs.component_out_path.version = '1'

pipeline = register_node_output()
pipeline.settings.default_compute = "azureml:cpu-cluster"

Следующие шаги

• Справочник по YAML для задания конвейера
• Отладка сбоя конвейера
• Планирование задания конвейера
• Развертывание конвейера с помощью пакетных конечных точек (предварительная версия)