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. Este artigo descreve as 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 variáveis, como learning_rate. As 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, consulte 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 modelos.

  • mlflow_model
  • custom_model

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

• Tipos primitivos
  • string
  • number
  • integer
  • boolean

Não há suporte para saída de tipo primitivo.

Exemplos de entradas e saídas

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

• O componente de trem tem uma number entrada chamada test_split_ratio.
• O componente de preparação tem uma saída de uri_folder tipo. 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 do trem tem uma saída de mlflow_model tipo. O código-fonte do componente salva o modelo treinado usando o mlflow.sklearn.save_model método.

Serialização de saída

O uso de dados ou saídas de modelo 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 para o 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 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 com suporte para entradas e saídas de pipeline do Azure Machine Learning, com path exemplos de parâmetros.

Location	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>

* Usar o Armazenamento do Azure diretamente não é recomendado para entrada, porque ele pode precisar de 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 trabalho de pipeline.

Modos de entrada e saída do 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.

Type	upload	download	ro_mount	rw_mount	direct	eval_download	eval_mount
uri_folder entrada		✓	✓		✓
uri_file entrada		✓	✓		✓
mltable entrada		✓	✓		✓	✓	✓
uri_folder Saída	✓			✓
uri_file Saída	✓			✓
mltable Saída	✓			✓	✓

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

Entradas e saídas em gráficos de pipeline

Na página de trabalho de pipeline no estúdio do Azure Machine Learning, as entradas e saídas de 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 facilitar a identificação.

A captura de tela a seguir do gráfico de 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 no nível do pipeline ou no painel do componente para entradas no nível do componente. Para abrir o painel de componentes, clique duas vezes no componente no gráfico.

Quando você edita um pipeline no Studio Designer, as entradas e saídas do pipeline estão no painel de interface do pipeline e as entradas e saídas do componente estão no painel do componente.

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

Promover a entrada/saída de um componente para o nível de pipeline permite substituir a entrada/saída do componente quando você envia 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 no nível do componente para entradas/saídas no nível do 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 é entrada no nível do pipeline porque é declarada sob a inputs seção no nível raiz.

Na train_job seção , a jobs entrada nomeada max_epocs é referenciada como ${{parent.inputs.pipeline_job_training_max_epocs}}, o que significa que a train_jobentrada do faz referência à entrada no max_epocs nível pipeline_job_training_max_epocs do pipeline. 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 no pipeline train-score-eval com componentes registrados no repositório de exemplos do Azure Machine Learning.

Python SDK

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

Você pode encontrar o bloco de anotações 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 no nível do pipeline na página de criação do Studio Designer.

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 necessárias e devem ter um valor padrão ou ser atribuído um valor cada vez que você enviar um trabalho de pipeline. No entanto, você pode definir uma entrada opcional.

Nota

Não há suporte para saídas opcionais.

A definição de entradas opcionais pode ser útil em dois cenários:

• Se você definir uma entrada opcional de tipo de dados/modelo e não atribuir um valor a ela ao enviar o trabalho de pipeline, o componente de 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 aguardar uma dependência anterior.

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

  

Azure CLI / Python SDK

O exemplo de código a seguir mostra como definir entrada opcional. Quando a entrada é definida como optional = true, você deve usar $[[]] para abraçar as entradas de 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. As entradas opcionais de tipos primitivos estão na guia Configurações . Ao contrário das entradas necessá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 conjunto definido {default_datastore} para o pipeline, azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}. Se não estiver definido, o padrão será o armazenamento de blob do espaço de trabalho.

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

CLI do Azure

O arquivo pipeline.yml no pipeline train-score-eval com exemplo de componentes registrados define um pipeline que tem três saídas de nível de pipeline. Você pode usar o comando a seguir para definir caminhos de saída personalizados para a pipeline_job_trained_model saída.

# 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  

Python SDK

O código a seguir que demonstra como personalizar caminhos de saída é do pipeline Build com command_component bloco de anotações de função python decorado.

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 Pipeline do Designer para um pipeline ou no painel do componente para um componente, expanda Saídas na guia Configurações para especificar um Caminho personalizado.

Download de saídas

Você pode baixar saídas no nível de pipeline ou 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>

Python SDK

Primeiro, crie e inicialize ml_client como um identificador para fazer referência ao seu espaço de trabalho. Para obter mais informações, consulte 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
)

Faça o download de todas as saídas de um trabalho ou faça o download de 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 todas no menu superior.
• Para baixar uma saída específica, selecione ... ao lado de um arquivo e selecione Download 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, em seguida, 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>

Python SDK

Primeiro, crie e inicialize ml_client como um identificador para fazer referência ao seu espaço de trabalho. Para obter mais informações, consulte 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, em seguida, 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.

Registrar a saída como um ativo nomeado

Você pode registrar a saída de um componente ou pipeline como um ativo nomeado atribuindo um name e version à saída. O ativo registrado pode ser listado em 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 no nível do 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

Python SDK

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 de ativo de dados em Entradas ou Saídas. Na página do ativo de dados, selecione Registrar.

Registrar saída de 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

Python SDK

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 do ativo de dados, selecione Registrar.

Conteúdos relacionados

• Referência YAML para trabalho de pipeline
• Como depurar falha de pipeline
• Agendar um trabalho de pipeline
• Implantar um pipeline com pontos de extremidade em lote (visualização)