Gerenciar entradas e saídas para componentes e pipelines

Os pipelines do Azure Machine Learning dão suporte a entradas e saídas nos níveis de componente e pipeline. Esse artigo descreve entradas e saídas de pipeline e componentes e como gerenciá-las.

No nível do componente, as entradas e saídas definem a interface do componente. Você pode usar a saída de um componente como entrada para outro componente no mesmo pipeline pai, permitindo que dados ou modelos sejam passados ​​entre componentes. Essa interconectividade representa o fluxo de dados dentro do pipeline.

No nível do pipeline, você pode usar entradas e saídas para enviar trabalhos de pipeline com entradas ou parâmetros de dados variados, como learning_rate. Entradas e saídas são especialmente úteis quando você invoca um pipeline por meio de um ponto de extremidade REST. Você pode atribuir valores diferentes à entrada do pipeline ou acessar a saída de diferentes trabalhos do pipeline. Para obter mais informações, veja Criar trabalhos e dados de entrada para pontos de extremidade em lote.

Tipos de entrada e saída

Os seguintes tipos são suportados como entradas e saídas de componentes ou pipelines:

• Tipos de dados. Para obter mais informações, consulte Tipos de dados.

  • uri_file
  • uri_folder
  • mltable

• Tipos de modelo.

  • mlflow_model
  • custom_model

Os seguintes tipos primitivos também são suportados somente para entradas:

• Tipos primitivos
  • string
  • number
  • integer
  • boolean

A saída do tipo primitivo não é suportada.

Entradas e saídas de exemplo

Esses exemplos são do pipeline NYC Taxi Data Regression no repositório Exemplos de Azure Machine Learning GitHub.

• O componente de trem tem uma numberentrada chamadatest_split_ratio.
• O componente prep tem uma saída do tipo uri_folder. O código-fonte do componente lê os arquivos CSV da pasta de entrada, processa os arquivos e grava os arquivos CSV processados ​​na pasta de saída.
• O componente de trem tem uma saída do tipo mlflow_model. O código-fonte do componente salva o modelo treinado usando o método mlflow.sklearn.save_model.

Serialização de saída

O uso de saídas de dados ou modelos serializa as saídas e as salva como arquivos em um local de armazenamento. Etapas posteriores podem acessar os arquivos durante a execução do trabalho montando esse local de armazenamento ou baixando ou carregando os arquivos no sistema de arquivos de computação.

O código-fonte do componente deve serializar o objeto de saída, que geralmente é armazenado na memória, em arquivos. Por exemplo, você pode serializar um dataframe do pandas em um arquivo CSV. O Azure Machine Learning não define nenhum método padronizado para serialização de objetos. Você tem a flexibilidade de escolher seus métodos preferidos para serializar objetos em arquivos. No componente downstream, você pode escolher como desserializar e ler esses arquivos.

Caminhos de entrada e saída do tipo de dados

Para entradas e saídas de ativos de dados, você deve especificar um parâmetro de caminho que aponte para o local dos dados. A tabela a seguir mostra os locais de dados suportados para entradas e saídas do pipeline do Azure Machine Learning, com exemplos de parâmetros path.

Localidade	Entrada	Saída	Exemplo
Um caminho no computador local	✓		./home/<username>/data/my_data
Um caminho em um servidor http/s público	✓		https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv
Um caminho no Armazenamento do Azure	*		wasbs://<container_name>@<account_name>.blob.core.windows.net/<path> ou abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>
Um caminho em um armazenamento de dados do Azure Machine Learning	✓	✓	azureml://datastores/<data_store_name>/paths/<path>
Um caminho para um ativo de dados	✓	✓	azureml:my_data:<version>

* Não é recomendado usar o Armazenamento do Microsoft Azure diretamente para entrada, pois pode ser necessária configuração de identidade extra para ler os dados. É melhor usar caminhos de armazenamento de dados do Azure Machine Learning, que são suportados em vários tipos de trabalhos de pipeline.

Modos de entrada e saída de tipo de dados

Para entradas e saídas de tipo de dados, você pode escolher entre vários modos de download, upload e montagem para definir como o destino de computação acessa os dados. A tabela a seguir mostra os modos suportados para diferentes tipos de entradas e saídas.

Tipo	upload	download	ro_mount	rw_mount	direct	eval_download	eval_mount
Entrada uri_folder		✓	✓		✓
Entrada uri_file		✓	✓		✓
Entrada mltable		✓	✓		✓	✓	✓
Saída uri_folder	✓			✓
Saída uri_file	✓			✓
Saída mltable	✓			✓	✓

Os modos ro_mount ou rw_mount são recomendados para a maioria dos casos. Para mais informações, veja Modos.

Entradas e saídas em gráficos de pipeline

Na página de trabalho do pipeline no Estúdio do Azure Machine Learning, as entradas e saídas dos componentes aparecem como pequenos círculos chamados portas de entrada/saída. Essas portas representam o fluxo de dados no pipeline. A saída do nível do pipeline é exibida em caixas roxas para fácil identificação.

A captura de tela a seguir do gráfico do pipeline NYC Taxi Data Regression mostra muitas entradas e saídas de componentes e pipelines.

Quando você passa o mouse sobre uma porta de entrada/saída, o tipo é exibido.

O gráfico de pipeline não exibe entradas de tipo primitivo. Essas entradas aparecem na guia Configurações do painel Visão geral do trabalho do pipeline para entradas em nível de pipeline ou no painel de componentes para entradas em nível de componente. Para abrir o painel de componentes, clique duas vezes no componente no gráfico.

Ao editar um pipeline no Studio Designer, as entradas e saídas do pipeline ficam no painel Interface do pipeline, e as entradas e saídas do componente ficam no painel do componente.

Promover entradas/saídas de componentes para o nível de pipeline

Promover a entrada/saída de um componente para o nível de pipeline permite que você substitua a entrada/saída do componente ao enviar um trabalho de pipeline. Essa capacidade é especialmente útil para acionar pipelines usando pontos de extremidade REST.

Os exemplos a seguir mostram como promover entradas/saídas em nível de componente para entradas/saídas em nível de pipeline.

CLI do Azure

O pipeline a seguir promove três entradas e três saídas para o nível do pipeline. Por exemplo, pipeline_job_training_max_epocs é uma entrada de nível de pipeline porque é declarada na seção inputs no nível raiz.

Em train_job na seção jobs, a entrada chamada max_epocs é referenciada como ${{parent.inputs.pipeline_job_training_max_epocs}}, o que significa que a entrada train_jobdo max_epocs faz referência à entrada do nível do pipeline pipeline_job_training_max_epocs. A saída do pipeline é promovida usando o mesmo esquema.

$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}}

Você pode encontrar o exemplo completo em pipeline train-score-eval com componentes registrados no repositório de exemplos do Azure Machine Learning.

SDK do Python

O exemplo de código a seguir define o pipeline nyc_taxi_data_regression. O pipeline recebe uma entrada, pipeline_job_input, e gera seis saídas, conforme definido na instrução return. As saídas do pipeline são promovidas do componente filho usando o esquema <step_name.outputs.output_name>, por exemplo prepare_sample_data.outputs.prep_data.

Você pode encontrar o notebook de ponta a ponta em NYC taxi data regression no repositório de exemplos do Azure Machine Learning.

# 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 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 directory 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
@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,
    }
# define pipeline job
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"

Interface do Usuário do Estúdio

Você pode promover a entrada de um componente para a entrada em nível de pipeline na página de criação do Designer do Studio.

1. Abra o painel de configurações do componente clicando duas vezes no componente.

2. Selecione ... ao lado da entrada que você deseja promover.

3. Selecione Adicionar à entrada do pipeline.

   

Definir entradas opcionais

Por padrão, todas as entradas são obrigatórias e devem ter um valor padrão ou receber um valor sempre que você enviar um trabalho de pipeline. No entanto, você pode definir uma entrada opcional.

Observação

Saídas opcionais não são suportadas.

Definir entradas opcionais pode ser útil em dois cenários:

• Se você definir uma entrada de tipo de dado/modelo opcional e não atribuir um valor a ela ao enviar o trabalho do pipeline, o componente do pipeline não terá essa dependência de dados. Se a porta de entrada do componente não estiver vinculada a nenhum componente ou nó de dados/modelo, o pipeline invocará o componente diretamente em vez de esperar por uma dependência anterior.

• Se você definir continue_on_step_failure = True para o pipeline, mas node2 usar a entrada necessária de node1, node2 não será executado se node1 falhar. Se node1 a entrada for opcional, node2 será executado mesmo se node1 falhar. O gráfico a seguir demonstra esse cenário.

  

CLI do Azure / SDK do Python

O exemplo de código a seguir mostra como definir uma entrada opcional. Quando a entrada é definida como optional = true, você deve usar $[[]] para abraçar as entradas da linha de comando, como nas linhas destacadas do exemplo.

$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.5/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}}

Interface do Usuário do Estúdio

Em um gráfico de pipeline, círculos pontilhados representam entradas opcionais de dados ou tipos de modelo. Entradas opcionais de tipos primitivos estão na aba Configurações. Ao contrário das entradas obrigatórias, as entradas opcionais não têm um asterisco ao lado delas, indicando que não são obrigatórias.

Personalizar caminhos de saída

Por padrão, a saída do componente é armazenada no {default_datastore} que você definiu para o pipeline, azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}. Se não for definido, o padrão será o armazenamento de blobs do espaço de trabalho.

O trabalho {name} é resolvido no momento da execução do trabalho e {output_name} é o nome que você definiu no componente YAML. Você pode personalizar onde armazenar a saída definindo um caminho de saída.

CLI do Azure

O arquivo pipeline.yml em train-score-eval pipeline com exemplo de componentes registrados define um pipeline que tem três saídas de nível de pipeline. Você pode usar o seguinte comando para definir caminhos de saída personalizados para a saída 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 do Python

O código a seguir, que demonstra como personalizar os caminhos de saída, é do notebook pipeline de build com command_component Decorated Python Function.

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
)

Interface do Usuário do Estúdio

Na interface Designer Pipeline para um pipeline, ou no painel de componentes para um componente, expanda Saídas na guia Configurações para especificar um Caminho personalizado.

Baixar saídas

Você pode baixar saídas no nível do pipeline ou do componente.

Baixar saídas de nível de pipeline

CLI do Azure

Você pode baixar todas as saídas de um trabalho ou baixar uma saída específica.

# 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 a 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 do Python

Primeiro, crie e inicialize ml_client como um identificador para referenciar seu espaço de trabalho. Para obter mais informações, veja Criar um identificador para o espaço de trabalho.

# 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
)

Baixe todas as saídas de um trabalho ou baixe uma saída específica.

# 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)

Interface do Usuário do Estúdio

Na guia Saídas + logs da página de detalhes do trabalho:

• Para baixar todas as saídas, selecione Baixar tudo no menu superior.
• Para baixar uma saída específica, selecione ... ao lado de um arquivo e selecione Baixar no menu de contexto.

Baixar saídas de componentes

CLI do Azure

Para baixar as saídas de um componente filho, primeiro liste todos os trabalhos filho de um trabalho de pipeline e depois use um código semelhante para baixar as saídas.

# 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 the desired 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 do Python

Primeiro, crie e inicialize ml_client como um identificador para referenciar seu espaço de trabalho. Para obter mais informações, veja Criar um identificador para o espaço de trabalho.

# 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
)

Para baixar as saídas de um componente filho, primeiro liste todos os trabalhos filho de um trabalho de pipeline e depois use um código semelhante para baixar as saídas.

# 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)

Interface do Usuário do Estúdio

Na guia Saídas + logs do painel de componentes de um componente, selecione Baixar tudo.

Registre a saída como um ativo nomeado

Você pode registrar a saída de um componente ou pipeline como um ativo nomeado atribuindo name e version à saída. O ativo registrado pode ser listado no seu espaço de trabalho por meio da interface do usuário do estúdio, CLI ou SDK e pode ser referenciado em trabalhos futuros do espaço de trabalho.

Registrar saída de nível de pipeline

CLI do Azure

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 do 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'

Interface do Usuário do Estúdio

Na guia Visão geral de um trabalho de pipeline, selecione um link Ativo de dados em Entradas ou Saídas. Na página de ativos de dados, selecione Registrar.

Registro de saída do componente

CLI do Azure

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 do 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"

Interface do Usuário do Estúdio

Na guia Visão geral de um componente, selecione um link Ativo de dados em Entradas ou Saídas. Na página de ativos de dados, selecione Registrar.

Conteúdo relacionado

• Referência YAML para trabalho de pipeline
• Como depurar a falha do pipeline
• Agendar um trabalho de pipeline
• Implantar um pipeline com pontos de extremidade em lotes (versão prévia)