Hantera indata och utdata för komponenter och pipelines

Följande pipeline höjer upp tre indata och tre utdata till pipelinenivån. Till exempel pipeline_job_training_max_epocs är indata på pipelinenivå eftersom de deklareras under inputs avsnittet på rotnivå.

Under train_job i avsnittet jobs refereras indata med namnet max_epocs som ${{parent.inputs.pipeline_job_training_max_epocs}}, vilket innebär att train_job"s-indata max_epocs refererar till pipelinenivåindata pipeline_job_training_max_epocs . Pipelineutdata höjs upp med hjälp av samma schema.

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

Du hittar det fullständiga exemplet på pipelinen train-score-eval med registrerade komponenter i Azure Machine Learning-exempellagringsplatsen .

I följande kodexempel definieras pipelinen nyc_taxi_data_regression . Pipelinen tar en indata, pipeline_job_input, och genererar sex utdata enligt definitionen i -instruktionen return . Pipelineutdata höjs upp från den underordnade komponenten med hjälp av schemat <step_name.outputs.output_name>, till exempel prepare_sample_data.outputs.prep_data.

Du hittar notebook-filen från slutpunkt till slutpunkt vid regression av taxidata i Azure Machine Learning-exempellagringsplatsen .

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

Du kan höja upp en komponents indata till pipelinenivåindata på studiodesignerns redigeringssida.

1. Öppna komponentens inställningspanel genom att dubbelklicka på komponenten.

2. Välj ... bredvid de indata som du vill höja upp.

3. Välj Lägg till i pipelineindata.

   

I följande kodexempel visas hur du definierar valfria indata. När indata anges som optional = truemåste du använda $[[]] för att använda kommandoradsindata, som i de markerade raderna i exemplet.

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

I ett pipelinediagram representerar prickade cirklar valfria indata för data eller modelltyper. Valfria indata för primitiva typer finns på fliken Inställningar . Till skillnad från obligatoriska indata har valfria indata ingen asterisk bredvid sig, vilket indikerar att de inte är obligatoriska.

Den pipeline.yml filen vid train-score-eval-pipelinen med exempel på registrerade komponenter definierar en pipeline som har tre utdata på pipelinenivå. Använd följande kommando för att ange anpassade utdatasökvägar för pipeline_job_trained_model utdata:

# 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  

Följande kod visar hur du anpassar utdatasökvägar och kommer från den Build-pipelinen med command_component-dekorerad Python-funktionsanteckningsboken :

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"

I Designer Pipeline-gränssnittet för en pipeline eller komponentpanelen för en komponent expanderar du Utdata på fliken Inställningar för att ange en anpassad sökväg.

Du kan ladda ned alla utdata från ett jobb eller ladda ned en specifik utdata.

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

Börja med att skapa och initiera ml_client som ett handtag för att referera till din arbetsyta. Mer information finns i Skapa ett handtag till arbetsytan.

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

Ladda ned alla utdata för ett jobb eller ladda ned ett specifikt utdata.

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

På fliken Utdata + loggar på sidan med jobbinformation:

• Om du vill ladda ned alla utdata väljer du Ladda ned alla på den översta menyn.
• Om du vill ladda ned ett specifikt utdata väljer du ... bredvid en fil och väljer Ladda ned på snabbmenyn.

Om du vill ladda ned utdata från en underordnad komponent anger du först alla underordnade jobb för ett pipelinejobb och använder sedan liknande kod för att ladda ned utdata.

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

Börja med att skapa och initiera ml_client som ett handtag för att referera till din arbetsyta. Mer information finns i Skapa ett handtag till arbetsytan.

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

Om du vill ladda ned utdata från en underordnad komponent anger du först alla underordnade jobb för ett pipelinejobb och använder sedan liknande kod för att ladda ned utdata.

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

På fliken Utdata + loggar i komponentpanelen för en komponent väljer du Ladda ned alla.

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
  }

yamle = register_pipeline_output()
display_name: register_node_outputter pipeline output
type: pipelinengs.default_compute = "azureml:cpu-cluster"
jobs:ine.outputs.component_out_path.name = 'pipeline_output'
  node:e.outputs.component_out_path.version = '1'
    type: command
    component: ../components/helloworld_component.yml
    inputs:I](#tab/ui)
      component_in_path:
        type: uri_fileb for a pipeline job, select a **Data asset** link under **Inputs** or **Outputs**. On the data asset page, select **Register**.
        path: 'https://dprepdata.blob.core.windows.net/demo/Titanic.csv'
    outputs:e="content" source="./media/how-to-manage-pipeline-input-output/register-output.png" alt-text="Screenshot showing how to register output from a pipeline job.":::
      component_out_path:
        type: uri_folder
        name: 'node_output'  # Define name and version to register a child job's output
        version: '1'nt output
settings:
  default_compute: azureml:cpu-cluster

# [Python SDK](#tab/python)_output
type: pipeline
```python
from azure.ai.ml import dsl, Output
    type: command
# Load component functionsts/helloworld_component.yml
components_dir = "./components/"
helloworld_component = load_component(source=f"{components_dir}/helloworld_component.yml")
        type: uri_file
@pipeline()h: 'https://dprepdata.blob.core.windows.net/demo/Titanic.csv'
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'))t
        version: '1'
  # 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"
```python
from azure.ai.ml import dsl, Output
# [Studio UI](#tab/ui)
# Load component functions
On the **Overview** tab for a component, select a **Data asset** link under **Inputs** or **Outputs**. On the data asset page, select **Register**.
helloworld_component = load_component(source=f"{components_dir}/helloworld_component.yml")
---
@pipeline()
## Related contentoutput():
  # Call component obj as function: apply given inputs & parameters to create a node in pipeline
- [YAML reference for pipeline job](./reference-yaml-job-pipeline.md)
- [How to debug pipeline failure](./how-to-debug-pipeline-failure.md)mo/Titanic.csv'))
- [Schedule a pipeline job](./how-to-schedule-pipeline-job.md)
- [Deploy a pipeline with batch endpoints (preview)](./how-to-use-batch-pipeline-deployments.md)
  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"

På fliken Översikt för en komponent väljer du en datatillgångslänk under Indata eller Utdata. På sidan datatillgång väljer du Registrera.