Správa vstupů a výstupů komponent a kanálu

V tomto článku se dozvíte:

• Přehled vstupů a výstupů v komponentách a kanálech
• Jak zvýšit úroveň vstupů a výstupů komponent na vstupy a výstupy kanálu
• Jak definovat volitelné vstupy
• Jak přizpůsobit cestu výstupu
• Stažení výstupů
• Postup registrace výstupů jako pojmenovaného prostředku

Přehled vstupů a výstupů

Kanály Azure Machine Učení podporují vstupy a výstupy na úrovni komponent i kanálů.

Na úrovni komponenty definují vstupy a výstupy rozhraní komponenty. Výstup jedné komponenty lze použít jako vstup pro jinou komponentu ve stejném nadřazeném kanálu, což umožňuje předávání dat nebo modelů mezi komponentami. Toto propojení tvoří graf, který znázorňuje tok dat v rámci kanálu.

Na úrovni kanálu jsou vstupy a výstupy užitečné pro odesílání úloh kanálu s různými datovými vstupy nebo parametry, které řídí logiku trénování (například learning_rate). Jsou zvlášť užitečné při vyvolání kanálu prostřednictvím koncového bodu REST. Tyto vstupy a výstupy umožňují přiřadit ke vstupu kanálu různé hodnoty nebo přistupovat k výstupu úloh kanálu prostřednictvím koncového bodu REST. Další informace najdete v tématu Vytváření úloh a vstupních dat pro koncový bod batch.

Typy vstupů a výstupů

Následující typy jsou podporovány jako výstupy komponenty nebo kanálu.

• Datové typy. Další informace o datových typech najdete ve službě Azure Machine Učení.

  • uri_file
  • uri_folder
  • mltable

• Typy modelů.

  • mlflow_model
  • custom_model

Použití výstupu dat nebo modelu v podstatě serializuje výstupy a uloží je jako soubory do umístění úložiště. V dalších krocích je možné toto umístění úložiště připojit, stáhnout nebo nahrát do cílového systému souborů výpočetního objektu a povolit dalšímu kroku přístup k souborům během provádění úlohy.

Tento proces vyžaduje, aby zdrojový kód komponenty serializoval požadovaný výstupní objekt – obvykle uložený v paměti – do souborů. Můžete například serializovat datový rámec pandas jako soubor CSV. Všimněte si, že Azure Machine Učení nedefinuje žádné standardizované metody pro serializaci objektů. Jako uživatel máte možnost zvolit preferovanou metodu serializace objektů do souborů. Potom můžete v podřízené komponentě nezávisle deserializovat a číst tyto soubory. Tady je několik příkladů pro referenci:

• V nyc_taxi_data_regression příkladu máuri_folder komponenta prep výstup typu. Ve zdrojovém kódu komponenty načte soubory CSV ze vstupní složky, zpracuje soubory a zapisuje zpracované soubory CSV do výstupní složky.
• V nyc_taxi_data_regression příkladu má mlflow_model komponenta train výstup typu. Ve zdrojovém kódu komponenty uloží trénovaný model pomocí mlflow.sklearn.save_model metody.

Kromě výše uvedených typů dat nebo modelů mohou vstupy kanálů nebo komponent být také následující primitivní typy.

• string
• number
• integer
• boolean

V nyc_taxi_data_regression příkladu má number komponenta trénu vstup s názvem test_split_ratio.

Poznámka:

Výstup primitivních typů není podporován.

Cesta a režim pro vstupy a výstupy dat

Pro vstup a výstup datového assetu je nutné zadat path parametr, který odkazuje na umístění dat. Tato tabulka ukazuje různá umístění dat, která kanál Azure Machine Učení podporuje, a také ukazuje příklady parametrů cesty:

Umístění	Příklady	Vstup	Výstup
Cesta na místním počítači	./home/username/data/my_data	✓
Cesta na veřejném serveru HTTP	https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv	✓
Cesta ve službě Azure Storage	wasbs://<container_name>@<account_name>.blob.core.windows.net/<path> abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>	Nenavrhuje se, protože ke čtení dat může potřebovat další konfiguraci identity.
Cesta na počítači Azure Učení datastore	azureml://datastores/<data_store_name>/paths/<path>	✓	✓
Cesta k datovému assetu	azureml:<my_data>:<version>	✓	✓

Poznámka:

Pro vstup a výstup v úložišti důrazně doporučujeme místo přímé cesty ke službě Azure Storage použít cestu úložiště dat azure machine Učení. Cesta k úložišti dat se podporuje napříč různými typy úloh v kanálu.

Pro vstup a výstup dat si můžete vybrat z různých režimů (stažení, připojení nebo nahrání) a definovat, jak se k datům přistupuje v cílovém výpočetním objektu. Tato tabulka ukazuje možné režimy pro různé kombinace typů,režimů/vstupu/výstupu.

Typ	Vstup a výstup	upload	download	ro_mount	rw_mount	direct	eval_download	eval_mount
uri_folder	Vstup		✓	✓		✓
uri_file	Vstup		✓	✓		✓
mltable	Vstup		✓	✓		✓	✓	✓
uri_folder	Výstup	✓			✓
uri_file	Výstup	✓			✓
mltable	Výstup	✓			✓	✓

Poznámka:

Ve většině případů doporučujeme použít ro_mount nebo rw_mount režim. Další informace o režimu najdete v režimech datových assetů.

Vizuální reprezentace v studio Azure Machine Learning

Následující snímky obrazovky poskytují příklad zobrazení vstupů a výstupů v úloze kanálu v studio Azure Machine Learning. Tuto konkrétní úlohu s názvem nyc-taxi-data-regressionnajdete v příkladu azureml.

Na stránce úlohy kanálu studia se vstupy a výstupy typu dat/modelu komponenty zobrazují jako malý kruh v odpovídající komponentě, označované jako vstupní/výstupní port. Tyto porty představují tok dat v kanálu.

Výstup na úrovni kanálu se zobrazí jako fialové pole pro snadnou identifikaci.

Při najetí myší na vstupní/výstupní port se zobrazí typ.

Vstupy primitivního typu se v grafu nezobrazí. Najdete ho na kartě Nastavení panelu přehledu úloh kanálu (pro vstupy na úrovni kanálu) nebo na panelu komponent (pro vstupy na úrovni komponenty). Následující snímek obrazovky ukazuje kartu Nastavení úlohy kanálu, kterou můžete otevřít výběrem odkazu Přehled úlohy.

Pokud chcete zkontrolovat vstupy pro komponentu, poklikejte na komponentu a otevřete panel komponent.

Podobně při úpravě kanálu v návrháři můžete najít vstupy a výstupy kanálu na panelu rozhraní kanálu a vstupy komponent&výstupy na panelu komponenty (trigger poklikáním na komponentu).

Jak zvýšit úroveň vstupů komponent a výstupů na úroveň kanálu

Zvýšení úrovně vstupu a výstupu komponenty na úroveň kanálu umožňuje při odesílání úlohy kanálu přepsat vstup a výstup komponenty. Je také užitečné, pokud chcete kanál aktivovat pomocí koncového bodu REST.

Následují příklady pro zvýšení úrovně vstupů a výstupů komponent na vstupy a výstupy na úrovni kanálu.

Azure CLI

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

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

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

settings:
 default_compute: azureml:cpu-cluster

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

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

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

Úplný příklad najdete v kanálu train-score-eval s registrovanými komponentami. Tento kanál podporuje tři vstupy a tři výstupy na úroveň kanálu. Podívejme se pipeline_job_training_max_epocs na příklad. Deklaruje se v inputs části na kořenové úrovni, což znamená vstup na úrovni kanálu. V jobs -> train_job oddílu se na vstup pojmenovaný max_epocs odkazuje jako ${{parent.inputs.pipeline_job_training_max_epocs}}, který označuje train_jobvstup max_epocs 's odkazy na vstup pipeline_job_training_max_epocsna úrovni kanálu . Podobně můžete zvýšit úroveň výstupu kanálu pomocí stejného schématu.

Python SDK

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

Příklad koncového poznámkového bloku v úložišti s příkladem azureml

Studio

Vstup komponenty můžete zvýšit na vstup na úrovni kanálu na stránce pro vytváření návrháře. Poklikáním na komponentu přejděte na panel nastavení komponenty –> najděte vstup, který chcete zvýšit , vyberte> tři tečky vpravo –> Vyberte přidat ke vstupu kanálu.

Volitelný vstup

Ve výchozím nastavení jsou všechny vstupy povinné a musí být přiřazeny hodnotě (nebo výchozí hodnotě) při každém odeslání úlohy kanálu. Mohou však existovat instance, ve kterých potřebujete volitelné vstupy. V takových případech máte při odesílání úlohy kanálu flexibilitu nepřiřazovat hodnotu vstupu.

Volitelný vstup může být užitečný v následujících dvou scénářích:

• Pokud máte volitelný vstup typu dat nebo modelu a při odesílání úlohy kanálu nepřiřazujete hodnotu, v kanálu bude komponenta, která neobsahuje předchozí závislost dat. Jinými slovy, vstupní port není propojený s žádnou komponentou ani datovým nebo modelovým uzlem. To způsobí, že služba kanálu vyvolá tuto komponentu přímo, a nečeká na to, až bude předchozí závislost připravená.

• Následující snímek obrazovky ukazuje jasný příklad druhého scénáře. Pokud nastavíte continue_on_step_failure = True kanál a máte druhý uzel (uzel2), který jako volitelný vstup používá výstup z prvního uzlu (node1), uzel2 se přesto spustí i v případě, že uzel1 selže. Pokud však uzel2 používá požadovaný vstup z uzlu1, nebude spuštěn, pokud uzel1 selže.

  

Tady jsou příklady, jak definovat volitelný vstup.

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

Když je vstup nastavený jako optional = true, musíte použít $[[]] k přijetí příkazového řádku se vstupy. Viz zvýrazněný řádek ve výše uvedeném příkladu.

Poznámka:

Nepovinný výstup není podporován.

V grafu kanálu jsou volitelné vstupy typu Data/Model reprezentovány tečkovaným kruhem. Volitelné vstupy primitivních typů se dají nacházet na kartě Nastavení. Na rozdíl od požadovaných vstupů nemají volitelné vstupy vedle nich hvězdičku, což znamená, že nejsou povinné.

Přizpůsobení výstupní cesty

Ve výchozím nastavení se výstup komponenty uloží do azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}složky . Výchozí {default_datastore} sada zákazníků úložiště dat pro kanál. Pokud není nastavené úložiště objektů blob pracovního prostoru. Jedná se {name} o název úlohy, který se přeloží v době provádění úlohy. Jedná se {output_name} o zákazníka s názvem výstupu definovaným v YAML komponenty.

Můžete ale také přizpůsobit umístění pro uložení výstupu definováním cesty výstupu. Následuje příklad:

Azure CLI

Definuje pipeline.yaml kanál, který má tři výstupy na úrovni kanálu. Úplný YAML najdete v kanálu train-score-eval s příkladem registrovaných komponent. K nastavení vlastní výstupní cesty pro pipeline_job_trained_modelvýstup můžete použít následující příkaz.

# 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

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
)

Příklad koncového poznámkového bloku najdete v kanálu buildu s command_component zdobeným poznámkovým blokem funkcí Pythonu.

Stažení výstupu

Výstup komponenty nebo výstup kanálu si můžete stáhnout podle následujícího příkladu.

Stažení výstupu na úrovni kanálu

Azure CLI

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

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

Python SDK

Než se pustíme do kódu, potřebujete způsob, jak odkazovat na váš pracovní prostor. Vytvoříte ml_client pro popisovač pracovního prostoru. Informace o inicializaci ml_clientnajdete v tématu Vytvoření popisovače pracovního prostoru.

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

Stažení výstupu podřízené úlohy

Pokud potřebujete stáhnout výstup podřízené úlohy (výstup komponenty, který není propagovaný na úroveň kanálu), měli byste nejprve zobrazit seznam všech podřízených entit úlohy kanálu a pak použít podobný kód ke stažení výstupu.

Azure CLI

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

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

Python SDK

Než se pustíme do kódu, potřebujete způsob, jak odkazovat na váš pracovní prostor. Vytvoříte ml_client pro popisovač pracovního prostoru. Informace o inicializaci ml_clientnajdete v tématu Vytvoření popisovače pracovního prostoru.

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

Postup registrace výstupu jako pojmenovaného prostředku

Výstup komponenty nebo kanálu můžete zaregistrovat jako pojmenovaný prostředek přiřazením name a version výstupem. Registrovaný prostředek může být v pracovním prostoru vypisován prostřednictvím uživatelského rozhraní studia, rozhraní příkazového řádku nebo sady SDK a také odkazovat v budoucích úlohách.

Registrace výstupu kanálu

Azure CLI

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

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'

Registrace výstupu podřízené úlohy

Azure CLI

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

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"

Další kroky

• Referenční informace k YAML pro úlohu kanálu
• Postup ladění selhání kanálu
• Naplánování úlohy kanálu
• Nasazení kanálu s využitím dávkových koncových bodů (Preview)