Gerenciar entradas e saídas de componentes e pipelines

Artigo
10/18/2023

Neste artigo você aprende:

Visão geral de entradas e saídas em componentes e pipeline
Como promover entradas/saídas de componentes para entradas/saídas de pipeline
Como definir entradas opcionais
Como personalizar o caminho de saídas
Como baixar saídas
Como registrar saídas como ativo nomeado

Visão geral de entradas e saídas

Os pipelines do Azure Machine Learning dão suporte a entradas e saídas nos níveis de componente e pipeline.

No nível do componente, as entradas e saídas definem a interface de um componente. A saída de um componente pode ser usada como entrada para outro componente no mesmo pipeline pai, permitindo que dados ou modelos sejam passados entre componentes. Essa interconectividade forma um gráfico, ilustrando o fluxo de dados dentro do pipeline.

No nível do pipeline, as entradas e saídas são úteis para enviar trabalhos de pipeline com entradas de dados variáveis ou parâmetros que controlam a lógica de treinamento (por exemplo learning_rate). Eles são especialmente úteis ao invocar o pipeline por meio de um ponto de extremidade REST. Essas entradas e saídas permitem atribuir valores diferentes à entrada do pipeline ou acessar a saída dos trabalhos do pipeline por meio do ponto de extremidade REST. Para saber mais, consulte Criando trabalhos e dados de entrada para o ponto de extremidade em lote.

Tipos de entradas e saídas

Os seguintes tipos são suportados como saídas de um componente ou pipeline.

Tipos de dados. Verifique os tipos de dados no Azure Machine Learning para saber mais sobre os tipos de dados.
- uri_file
- uri_folder
- mltable
Tipos de modelos.
- mlflow_model
- custom_model

Usando dados ou saída de modelo essencialmente serializando as saídas e salvá-las como arquivos em um local de armazenamento. Nas etapas subsequentes, esse local de armazenamento pode ser montado, baixado ou carregado no sistema de arquivos de destino de computação, permitindo que a próxima etapa acesse os arquivos durante a execução do trabalho.

Este processo requer o código-fonte do componente serializando o objeto de saída desejado - geralmente armazenado na memória - em arquivos. Por exemplo, você pode serializar um dataframe pandas como um arquivo CSV. Observe que o Aprendizado de Máquina do Azure não define nenhum método padronizado para serialização de objetos. Como usuário, você tem a flexibilidade de escolher seu método preferido para serializar objetos em arquivos. Depois disso, no componente downstream, você pode desserializar e ler esses arquivos de forma independente. Aqui estão alguns exemplos para sua referência:

No exemplo nyc_taxi_data_regression, o componente de preparação tem umauri_folder saída de tipo. No código-fonte do componente, ele lê os arquivos csv da pasta de entrada, processa os arquivos e grava arquivos CSV processados na pasta de saída.
No exemplo nyc_taxi_data_regression, o componente de trem tem uma saída de mlflow_model tipo. No código-fonte do componente, ele salva o modelo treinado usando mlflow.sklearn.save_model o método.

Além dos tipos de dados ou modelos acima, as entradas de pipeline ou componente também podem seguir tipos primitivos.

string
number
integer
boolean

No exemplo nyc_taxi_data_regression, o componente train tem uma number entrada chamada test_split_ratio.

Nota

A saída de tipos primitivos não é suportada.

Caminho e modo para entradas/saídas de dados

Para entrada/saída de ativos de dados, você deve especificar um path parâmetro que aponte para o local dos dados. Esta tabela mostra os diferentes locais de dados suportados pelo pipeline do Azure Machine Learning e também mostra exemplos de parâmetros de caminho:

Localização	Exemplos	Entrada	Saída
Um caminho no computador local	`./home/username/data/my_data`	✓
Um caminho em um servidor http(s) público(s)	`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>` `abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>`	Não sugerido porque pode precisar de configuração de identidade extra para ler os dados.
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>`	✓	✓

Nota

Para entrada/saída no armazenamento, sugerimos usar o caminho do armazenamento de dados do Azure Machine Learning em vez do caminho direto do Armazenamento do Azure. O caminho do armazenamento de dados é suportado em vários tipos de trabalho no pipeline.

Para entrada/saída de dados, você pode escolher entre vários modos (download, montagem ou upload) para definir como os dados são acessados no destino de computação. Esta tabela mostra os modos possíveis para diferentes combinações de tipo/modo/entrada/saída.

Type	Entrada/Saída	`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	✓			✓	✓

Nota

Na maioria dos casos, sugerimos usar ro_mount ou rw_mount moderar. Para saber mais sobre o modo, consulte Modos de ativos de dados.

Representação visual no estúdio do Azure Machine Learning

As capturas de tela a seguir fornecem um exemplo de como as entradas e saídas são exibidas em um trabalho de pipeline no estúdio do Azure Machine Learning. Este trabalho específico, chamado nyc-taxi-data-regression, pode ser encontrado em azureml-example.

Na página de trabalho de pipeline do estúdio, as entradas/saídas de tipo de dados/modelo de um componente são mostradas como um pequeno círculo no componente correspondente, conhecido como porta de entrada/saída. Essas portas representam o fluxo de dados em um pipeline.

A saída do nível do pipeline é exibida como uma caixa roxa para facilitar a identificação.

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

As entradas de tipo primitivo não serão exibidas no gráfico. Ele pode ser encontrado na guia Configurações do painel de visão geral do trabalho de pipeline (para entradas no nível do pipeline) ou no painel do componente (para entradas no nível do componente). A captura de tela a seguir mostra a guia Configurações de um trabalho de pipeline, ele pode ser aberto selecionando o link Visão geral do trabalho.

Se você quiser verificar as entradas de um componente, clique duas vezes no componente para abrir o painel do componente.

Da mesma forma, ao editar um pipeline no designer, você pode encontrar as entradas e saídas do pipeline no painel de interface do pipeline e as entradas e saídas do componente no painel do componente (acionar clicando duas vezes no componente).

Como promover entradas de componentes e saídas para o nível de pipeline

Promover a entrada/saída de um componente para o nível do pipeline permite que você substitua a entrada/saída do componente ao enviar um trabalho de pipeline. Também é útil se você quiser acionar o pipeline usando o ponto de extremidade REST.

Seguem-se exemplos para promover entradas/saídas de componentes para entradas/saídas ao nível de pipeline.

CLI do Azure
Python SDK

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

O exemplo completo pode ser encontrado no pipeline train-score-eval com componentes registrados. Este pipeline promove três entradas e três saídas ao nível do pipeline. Tomemos pipeline_job_training_max_epocs como exemplo. Ele é declarado em inputs seção no nível raiz, o que significa sua entrada de nível de pipeline. Na jobs -> train_job seção, a entrada nomeada max_epocs é referenciada como ${{parent.inputs.pipeline_job_training_max_epocs}}, o que indica que a train_jobentrada do faz referência à entrada pipeline_job_training_max_epocsmax_epocs no nível do pipeline. Da mesma forma, você pode promover a saída do pipeline usando o mesmo esquema.

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

O exemplo de bloco de anotações de ponta a ponta no repositório azureml-example

Studio

Você pode promover a entrada de um componente para a entrada no nível do pipeline na página de criação do designer. Vá para o painel de configuração do componente clicando duas vezes no componente - encontre a entrada que você gostaria de promover - Selecione os três pontos à direita ->>> Selecione Adicionar à entrada do pipeline.

Entrada opcional

Por padrão, todas as entradas são necessárias e devem receber um valor (ou um valor padrão) cada vez que você envia um trabalho de pipeline. No entanto, pode haver casos em que você precise de entradas opcionais. Nesses casos, você tem a flexibilidade de não atribuir um valor à entrada ao enviar um trabalho de pipeline.

A entrada opcional pode ser útil nos dois cenários abaixo:

Se você tiver uma entrada opcional de tipo de dados/modelo e não atribuir um valor a ela ao enviar o trabalho de pipeline, haverá um componente no pipeline que não possui uma dependência de dados anterior. Em outras palavras, a porta de entrada não está vinculada a nenhum componente ou nó de dados/modelo. Isso faz com que o serviço de pipeline invoque esse componente diretamente, em vez de aguardar que a dependência anterior esteja pronta.
A captura de tela abaixo fornece um exemplo claro do segundo cenário. Se você definir continue_on_step_failure = True para o pipeline e tiver um segundo nó (node2) que usa a saída do primeiro nó (node1) como uma entrada opcional, o node2 ainda será executado mesmo se o node1 falhar. No entanto, se o nó2 estiver usando a entrada necessária do nó1, ele não será executado se o nó1 falhar.

A seguir estão exemplos sobre como definir a entrada opcional.

$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
version: 9
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}}

Quando a entrada é definida como optional = true, você precisa usar $[[]] para abraçar a linha de comando com entradas. Veja a linha destacada no exemplo acima.

Nota

A saída opcional não é suportada.

No gráfico de pipeline, as entradas opcionais do tipo Dados/Modelo são representadas por um círculo pontilhado. Entradas opcionais de tipos primitivos podem ser localizadas na guia Configurações . Ao contrário das entradas necessárias, as entradas opcionais não têm um asterisco ao lado delas, significando que não são obrigatórias.

Como personalizar o caminho de saída

Por padrão, a saída de um componente será armazenada no azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}. O {default_datastore} é o conjunto de clientes de armazenamento de dados padrão para o pipeline. Se não estiver definido, é armazenamento de blob de espaço de trabalho. O {name} é o nome do trabalho, que será resolvido no momento da execução do trabalho. O {output_name} é o nome de saída do cliente definido no componente YAML.

Mas você também pode personalizar onde armazenar a saída definindo o caminho de uma saída. Seguem-se alguns exemplos:

CLI do Azure
Python SDK

O pipeline.yaml define um pipeline que tem três saídas de nível de pipeline. O YAML completo pode ser encontrado no pipeline train-score-eval com exemplo de componentes registrados. Você pode usar o comando seguinte para definir o caminho de saída personalizado para a pipeline_job_trained_modelsaí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

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
)

O exemplo de bloco de anotações de ponta a ponta pode ser encontrado em Build pipeline com command_component notebook de função python decorado.

Como fazer o download da saída

Você pode baixar a saída de um componente ou a saída do pipeline seguindo o exemplo abaixo.

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

Antes de mergulharmos no código, você precisa de uma maneira de referenciar seu espaço de trabalho. Você cria ml_client para um identificador para o espaço de trabalho. Consulte Criar identificador para espaço de trabalho para inicializar 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)

Baixe a saída do trabalho infantil

Quando precisar baixar a saída de um trabalho filho (uma saída de componente que não promove para o nível de pipeline), você deve primeiro listar todas as entidades de trabalho filho de um trabalho de pipeline e, em seguida, usar um código semelhante para baixar a saída.

CLI do Azure
Python SDK

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

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

Como registrar a saída como ativo nomeado

Você pode registrar a saída de um componente ou pipeline como ativo nomeado atribuindo name e version para a saída. O ativo registrado pode ser listado em seu espaço de trabalho através do estúdio UI/CLI/SDK e também ser referenciado em seus trabalhos futuros.

Registrar saída de pipeline

CLI do Azure
Python SDK

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

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'

Registrar a saída de um trabalho infantil

CLI do Azure
Python SDK

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

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"

Share via

Gerenciar entradas e saídas de componentes e pipelines

Visão geral de entradas e saídas

Tipos de entradas e saídas

Caminho e modo para entradas/saídas de dados

Representação visual no estúdio do Azure Machine Learning

Como promover entradas de componentes e saídas para o nível de pipeline

Studio

Entrada opcional

Como personalizar o caminho de saída

Como fazer o download da saída

Baixar saída de nível de pipeline

Baixe a saída do trabalho infantil

Como registrar a saída como ativo nomeado

Registrar saída de pipeline

Registrar a saída de um trabalho infantil

Próximos passos

Recursos adicionais