Hantera indata och utdata för komponenter och pipelines

Azure Machine Learning-pipelines stöder indata och utdata på både komponent- och pipelinenivå. Den här artikeln beskriver pipeline- och komponentindata och utdata och hur du hanterar dem.

På komponentnivå definierar indata och utdata komponentgränssnittet. Du kan använda utdata från en komponent som indata för en annan komponent i samma överordnade pipeline, vilket gör att data eller modeller kan skickas mellan komponenter. Den här sammankopplingen representerar dataflödet i pipelinen.

På pipelinenivå kan du använda indata och utdata för att skicka pipelinejobb med varierande dataindata eller parametrar, till exempel learning_rate. Indata och utdata är särskilt användbara när du anropar en pipeline via en REST-slutpunkt. Du kan tilldela olika värden till pipelineindata eller komma åt utdata från olika pipelinejobb. Mer information finns i Skapa jobb och indata för batchslutpunkter.

Indata- och utdatatyper

Följande typer stöds som både indata och utdata för komponenter eller pipelines:

• Datatyper. Mer information finns i Datatyper.

  • uri_file
  • uri_folder
  • mltable

• Modelltyper.

  • mlflow_model
  • custom_model

Följande primitiva typer stöds också endast för indata:

• Primitiva typer
  • string
  • number
  • integer
  • boolean

Primitiva typutdata stöds inte.

Exempel på indata och utdata

De här exemplen kommer från pipelinen NYC Taxi Data Regression i GitHub-lagringsplatsen för Azure Machine Learning-exempel.

• Träningskomponenten har en number indata med namnet test_split_ratio.
• Förberedelsekomponenten har en uri_folder typutdata. Komponentens källkod läser CSV-filerna från indatamappen, bearbetar filerna och skriver de bearbetade CSV-filerna till utdatamappen.
• Träningskomponenten har en mlflow_model typutdata. Komponentens källkod sparar den tränade modellen med hjälp av mlflow.sklearn.save_model metoden .

Serialisering av utdata

Med data- eller modellutdata serialiseras utdata och sparas som filer på en lagringsplats. Senare steg kan komma åt filerna under jobbkörningen genom att montera den här lagringsplatsen eller genom att ladda ned eller ladda upp filerna till beräkningsfilsystemet.

Komponentens källkod måste serialisera utdataobjektet, som vanligtvis lagras i minnet, till filer. Du kan till exempel serialisera en Pandas-dataram till en CSV-fil. Azure Machine Learning definierar inte några standardiserade metoder för objektserialisering. Du har flexibiliteten att välja de metoder som du föredrar för att serialisera objekt till filer. I den underordnade komponenten kan du välja hur du ska deserialisera och läsa dessa filer.

Indata- och utdatasökvägar för datatyp

För indata och utdata för datatillgången måste du ange en sökvägsparameter som pekar på dataplatsen. I följande tabell visas de dataplatser som stöds för Azure Machine Learning-pipelineindata och utdata, med path parameterexempel.

Plats	Indata	Utdata	Exempel
En sökväg på den lokala datorn	✓		./home/<username>/data/my_data
En sökväg på en offentlig http/s-server	✓		https://raw.githubusercontent.com/pandas-dev/pandas/main/doc/data/titanic.csv
En sökväg i Azure Storage	*		wasbs://<container_name>@<account_name>.blob.core.windows.net/<path> eller abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>
En sökväg i ett Azure Machine Learning-datalager	✓	✓	azureml://datastores/<data_store_name>/paths/<path>
En sökväg till en datatillgång	✓	✓	azureml:my_data:<version>

* Att använda Azure Storage direkt rekommenderas inte för indata, eftersom det kan behövas extra identitetskonfiguration för att läsa data. Det är bättre att använda Azure Machine Learning-datalagersökvägar, som stöds för olika typer av pipelinejobb.

Indata- och utdatalägen för datatyp

För datatypsindata och utdata kan du välja mellan flera lägen för nedladdning, uppladdning och montering för att definiera hur beräkningsmålet kommer åt data. I följande tabell visas de lägen som stöds för olika typer av indata och utdata.

Typ	upload	download	ro_mount	rw_mount	direct	eval_download	eval_mount
uri_folder inmatning		✓	✓		✓
uri_file inmatning		✓	✓		✓
mltable inmatning		✓	✓		✓	✓	✓
uri_folder utdata	✓			✓
uri_file utdata	✓			✓
mltable utdata	✓			✓	✓

Lägena ro_mount eller rw_mount rekommenderas i de flesta fall. Mer information finns i Lägen.

Indata och utdata i pipelinediagram

På sidan pipelinejobb i Azure Machine Learning-studio visas komponentindata och utdata som små cirklar som kallas indata-/utdataportar. Dessa portar representerar dataflödet i pipelinen. Utdata på pipelinenivå visas i lila rutor för enkel identifiering.

Följande skärmbild från pipelinediagrammet NYC Taxi Data Regression visar många komponent- och pipelineindata och utdata.

När du hovra över en in-/utdataport visas typen.

Pipelinediagrammet visar inte primitiva typindata. Dessa indata visas på fliken Inställningar i panelen för pipelinejobböversikt för indata på pipelinenivå eller komponentpanelen för indata på komponentnivå. Om du vill öppna komponentpanelen dubbelklickar du på komponenten i diagrammet.

När du redigerar en pipeline i studiodesignern finns pipelineindata och utdata i panelen Pipeline-gränssnitt , och komponentindata och utdata finns i komponentpanelen.

Flytta upp komponentindata/utdata till pipelinenivå

När du befordrar en komponents indata/utdata till pipelinenivån kan du skriva över komponentens indata/utdata när du skickar ett pipelinejobb. Den här möjligheten är särskilt användbar för att utlösa pipelines med hjälp av REST-slutpunkter.

I följande exempel visas hur du höjer upp indata/utdata på komponentnivå till indata/utdata på pipelinenivå.

Azure CLI

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 .

Python SDK

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"

Studio-användargränssnitt

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.

   

Definiera valfria indata

Som standard krävs alla indata och måste antingen ha ett standardvärde eller tilldelas ett värde varje gång du skickar ett pipelinejobb. Du kan dock definiera valfria indata.

Kommentar

Valfria utdata stöds inte.

Det kan vara användbart att ange valfria indata i två scenarier:

• Om du definierar en valfri indata för data/modelltyp och inte tilldelar ett värde till den när du skickar pipelinejobbet, saknar pipelinekomponenten det databeroendet. Om komponentens indataport inte är länkad till någon komponent eller data/modellnod anropar pipelinen komponenten direkt i stället för att vänta på ett tidigare beroende.

• Om du anger continue_on_step_failure = True för pipelinen men node2 använder nödvändiga indata från node1, node2 körs inte om node1 det misslyckas. Om node1 indata är valfria node2 körs även om node1 det misslyckas. Följande diagram visar det här scenariot.

  

Azure CLI/Python SDK

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

Studio-användargränssnitt

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.

Anpassa utdatasökvägar

Som standard lagras komponentutdata i den {default_datastore} du anger för pipelinen, azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}. Om den inte har angetts är standardinställningen bloblagring för arbetsytor.

Jobbet {name} löses vid jobbkörningen och {output_name} är det namn som du definierade i komponenten YAML. Du kan anpassa var utdata ska lagras genom att definiera en utdatasökväg.

Azure CLI

Den pipeline.yml filen vid train-score-eval-pipelinen med exempel på registrerade komponenter definierar en pipeline som har tre utdata på pipelinenivå. Du kan använda 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  

Python SDK

Följande kod som visar hur du anpassar utdatasökvägar kommer från build-pipelinen med command_component dekorerad python-funktionsanteckningsbok .

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
)

Studio-användargränssnitt

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.

Ladda ned utdata

Du kan ladda ned utdata på pipeline- eller komponentnivå.

Ladda ned utdata på pipelinenivå

Azure CLI

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>

Python SDK

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)

Studio-användargränssnitt

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.

Ladda ned komponentutdata

Azure CLI

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>

Python SDK

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)

Studio-användargränssnitt

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

Registrera utdata som en namngiven tillgång

Du kan registrera utdata för en komponent eller pipeline som en namngiven tillgång genom att tilldela en name och version till utdata. Den registrerade tillgången kan visas i din arbetsyta via studiogränssnittet, CLI eller SDK och kan refereras till i framtida arbetsytejobb.

Registrera utdata på pipelinenivå

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'

Studio-användargränssnitt

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

Registrera komponentutdata

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"

Studio-användargränssnitt

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.

Relaterat innehåll

• YAML-referens för pipelinejobb
• Felsöka pipelinefel
• Schemalägga ett pipelinejobb
• Distribuera en pipeline med batchslutpunkter (förhandsversion)