Gerenciar entradas e saídas de componente e pipeline

Artigo
10/20/2023

Neste artigo, você aprende:

Visão geral de entradas e saídas no componente e pipeline
Como promover entradas/saídas de componente 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 das saídas e entradas

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 grafo, ilustrando o fluxo de dados dentro do pipeline.

No nível do pipeline, entradas e saídas são úteis para enviar trabalhos de pipeline com diferentes entradas de dados 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 de trabalhos de pipeline por meio do ponto de extremidade REST. Para saber mais, confira Como criar trabalhos e dados de entrada para o ponto de extremidade do lote.

Tipos de entradas e saídas

Os tipos a seguir têm suporte como saídas de um componente ou de um 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 modelo.
- mlflow_model
- custom_model

Usando dados ou saída de modelo essencialmente serializando as saídas e salvando-as 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.

Esse 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 do Pandas como um arquivo CSV. Observe que o Azure Machine Learning 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 de nyc_taxi_data_regression, o componente de preparação tem uma saída de tipo uri_folder. 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 de nyc_taxi_data_regression, o componente de trem tem uma saída de tipomlflow_model. No código-fonte do componente, ele salva o modelo treinado usando o método mlflow.sklearn.save_model.

Além dos tipos de modelo ou dados acima, as entradas de pipeline ou componente também podem estar seguindo tipos primitivos.

string
number
integer
boolean

No exemplo de nyc_taxi_data_regression, componente de trem tem uma entrada number chamada test_split_ratio.

Observação

Não há suporte para a saída de tipos primitivos.

Caminho e modo para entradas/saídas de dados

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

Location	Exemplos	Entrada	Saída
Um caminho no computador local	`./home/username/data/my_data`	✓
Um caminho em um servidor https(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>` `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>`	✓	✓

Observação

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

Para entrada/saída de dados, você pode escolher entre vários modos (baixar, montar ou carregar) 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.

Tipo	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	✓			✓	✓

Observação

Na maioria dos casos, sugerimos usar ro_mount ou rw_mount modo. Para saber mais sobre o modo, consulte os modos de ativo de dados.

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

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

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

A saída de nível de 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 grafo. 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 de componentes (para entradas no nível do componente). A captura de tela a seguir mostra a guia Configurações de um trabalho de pipeline, que pode ser aberta 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 de componentes.

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

Como promover entradas e saídas de componente 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 ao enviar um trabalho de pipeline. Também será útil se você quiser disparar o pipeline usando o ponto de extremidade REST.

A seguir estão exemplos para promover entradas/saídas de componente para entradas/saídas de nível de pipeline.

CLI do Azure
SDK do Python

$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. Esse pipeline promove três entradas e três saídas para o nível do pipeline. Vamos tomar pipeline_job_training_max_epocs como exemplo. Ele é declarado em seção inputs no nível raiz, o que significa que é sua entrada no nível do pipeline. Na seção jobs -> train_job, a entrada nomeada max_epocs é referenciada como ${{parent.inputs.pipeline_job_training_max_epocs}}, o que indica que max_epocs da entrada de train_job faz referência à entrada de nível do pipeline pipeline_job_training_max_epocs. 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 notebook de ponta a ponta no repositório azureml-example

Estúdio

Você pode promover a entrada de um componente para entrada no nível do pipeline na página de criação do designer. Acesse o painel de configurações do componente clicando duas vezes no componente -> localize a entrada que quer 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) sempre que você enviar um trabalho de pipeline. No entanto, pode haver instâncias em que você precisa 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 em dois cenários abaixo:

Se você tiver uma entrada de tipo de modelo/dados opcional e não atribuir um valor a ela ao enviar o trabalho de pipeline, haverá um componente no pipeline que não tem 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 a dependência anterior estar 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 node2 estiver usando a entrada necessária do node1, ele não será executado se o node1 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 adotar a linha de comando com entradas. Veja a linha realçada no exemplo acima.

Observação

Não há suporte para a saída opcional.

No grafo de pipeline, as entradas opcionais do tipo Data/Model 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 elas não são obrigatórias.

Como personalizar o caminho de saída

Por padrão, a saída de um componente será armazenada em 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 for definido, será o armazenamento de blobs do workspace. O {name} é o nome do trabalho, que será resolvido no momento da execução do trabalho. O {output_name} é o nome de saída que o cliente definiu no YAML do componente.

Mas você também pode personalizar onde armazenar a saída definindo o caminho de uma saída. Veja a seguir o exemplo:

CLI do Azure
SDK do Python

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 o exemplo de componentes registrados. Você pode usar o comando a seguir 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 notebook de ponta a ponta pode ser encontrado no pipeline de build com command_component notebook de funções python decorado.

Como baixar a saída

Você pode baixar a saída ou a saída do pipeline de um componente 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 nos aprofundarmos no código, você precisa de uma maneira de referenciar seu workspace. Você cria ml_client um identificador para o workspace. Consulte Criar identificador para o workspace 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)

Baixar a saída do trabalho filho

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

CLI do Azure
SDK do Python

# 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 à saída. O ativo registrado pode ser listado em seu workspace por meio da interface do usuário/CLI/SDK do estúdio e também ser referenciado em seus trabalhos futuros.

Registrar saída de pipeline

CLI do Azure
SDK do Python

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 filho

CLI do Azure
SDK do Python

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"

Gerenciar entradas e saídas de componente e pipeline

Visão geral das saídas e entradas

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 e saídas de componente para o nível do pipeline

Estúdio

Entrada opcional

Como personalizar o caminho de saída

Como baixar a saída

Baixar a saída no nível do pipeline

Baixar a saída do trabalho filho

Como registrar a saída como ativo nomeado

Registrar saída de pipeline

Registrar a saída de um trabalho filho

Próximas etapas

Recursos adicionais