Hantera indata och utdata för komponent och pipeline

I den här artikeln lär du dig:

• Översikt över indata och utdata i komponent och pipeline
• Så här höjer du upp komponentindata/utdata till pipelineindata/utdata
• Definiera valfria indata
• Så här anpassar du sökvägen för utdata
• Så här laddar du ned utdata
• Så här registrerar du utdata som namngiven tillgång

Översikt över indata och utdata

Azure Mašinsko učenje-pipelines stöder indata och utdata på både komponent- och pipelinenivå.

På komponentnivå definierar indata och utdata gränssnittet för en komponent. Utdata från en komponent kan användas 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 sammanlänkningen utgör ett diagram som illustrerar dataflödet i pipelinen.

På pipelinenivå är indata och utdata användbara för att skicka pipelinejobb med varierande dataindata eller parametrar som styr träningslogik (till exempel learning_rate). De är särskilt användbara när du anropar pipelinen via en REST-slutpunkt. Med dessa indata och utdata kan du tilldela olika värden till pipelineindata eller komma åt utdata från pipelinejobb via REST-slutpunkten. Mer information finns i Skapa jobb och indata för Batch-slutpunkt.

Typer av indata och utdata

Följande typer stöds som utdata från en komponent eller en pipeline.

• Datatyper. Kontrollera datatyper i Azure Mašinsko učenje om du vill veta mer om datatyper.

  • uri_file
  • uri_folder
  • mltable

• Modelltyper.

  • mlflow_model
  • custom_model

Användning av data- eller modellutdata serialiserar utdata och sparar dem som filer på en lagringsplats. I efterföljande steg kan den här lagringsplatsen monteras, laddas ned eller laddas upp till filsystemet för beräkningsmål, vilket gör att nästa steg kan komma åt filerna under jobbkörningen.

Den här processen kräver att komponentens källkod serialiserar det önskade utdataobjektet , vanligtvis lagrat i minnet, till filer. Du kan till exempel serialisera en Pandas-dataram som en CSV-fil. Observera att Azure Mašinsko učenje inte definierar några standardiserade metoder för objektserialisering. Som användare har du flexibiliteten att välja önskad metod för att serialisera objekt till filer. Därefter kan du i den underordnade komponenten deserialisera och läsa dessa filer separat. Här är några exempel på din referens:

• I det nyc_taxi_data_regression exemplet har förberedelsekomponenten enuri_folder typutdata. I komponentens källkod läser den csv-filerna från indatamappen, bearbetar filerna och skriver bearbetade CSV-filer till utdatamappen.
• I det nyc_taxi_data_regression exemplet har träningskomponenten en mlflow_model typutdata. I komponentens källkod sparar den tränade modellen med hjälp av mlflow.sklearn.save_model metoden.

Förutom ovanstående data- eller modelltyper kan pipeline- eller komponentindata också följa primitiva typer.

• string
• number
• integer
• boolean

I det nyc_taxi_data_regression exemplet har träningskomponenten en number indata med namnet test_split_ratio.

Kommentar

Primitiva typer av utdata stöds inte.

Sökväg och läge för dataindata/utdata

För indata/utdata för datatillgången måste du ange en path parameter som pekar på dataplatsen. Den här tabellen visar de olika dataplatser som Azure Mašinsko učenje pipeline stöder och visar även sökvägsparameterexempel:

Plats	Exempel	Indata	Utdata
En sökväg på den lokala datorn	./home/username/data/my_data	✓
En sökväg på en offentlig http-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> abfss://<file_system>@<account_name>.dfs.core.windows.net/<path>	Inte föreslaget eftersom det kan behöva extra identitetskonfiguration för att läsa data.
En sökväg på en Azure Mašinsko učenje Datastore	azureml://datastores/<data_store_name>/paths/<path>	✓	✓
En sökväg till en datatillgång	azureml:<my_data>:<version>	✓	✓

Kommentar

För indata/utdata på lagring rekommenderar vi starkt att du använder Azure Mašinsko učenje datalagersökväg i stället för direkt Azure Storage-sökväg. Datalagersökvägen stöds för olika jobbtyper i pipeline.

För indata/utdata kan du välja mellan olika lägen (ladda ned, montera eller ladda upp) för att definiera hur data ska nås i beräkningsmålet. Den här tabellen visar möjliga lägen för olika kombinationer av typ/läge/indata/utdata.

Typ	Indata/utdata	upload	download	ro_mount	rw_mount	direct	eval_download	eval_mount
uri_folder	Indata		✓	✓		✓
uri_file	Indata		✓	✓		✓
mltable	Indata		✓	✓		✓	✓	✓
uri_folder	Utdata	✓			✓
uri_file	Utdata	✓			✓
mltable	Utdata	✓			✓	✓

Kommentar

I de flesta fall rekommenderar vi att du använder ro_mount eller rw_mount läge. Mer information om läget finns i datatillgångslägen.

Visuell representation i Azure Mašinsko učenje Studio

Följande skärmbilder ger ett exempel på hur indata och utdata visas i ett pipelinejobb i Azure Mašinsko učenje studio. Det här jobbet med namnet nyc-taxi-data-regression, finns i azureml-example.

På pipelinejobbsidan i studio visas data/modelltypen indata/utdata för en komponent som en liten cirkel i motsvarande komponent, som kallas indata-/utdataporten. Dessa portar representerar dataflödet i en pipeline.

Utdata på pipelinenivå visas som en lila ruta för enkel identifiering.

När du håller muspekaren på en in-/utdataport visas typen.

Indata av primitiv typ visas inte i diagrammet. Den finns på fliken Inställningar i översiktspanelen för pipelinejobb (för indata på pipelinenivå) eller komponentpanelen (för indata på komponentnivå). Följande skärmbild visar fliken Inställningar för ett pipelinejobb. Den kan öppnas genom att välja länken Jobböversikt .

Om du vill kontrollera indata för en komponent dubbelklickar du på komponenten för att öppna komponentpanelen.

När du redigerar en pipeline i designern hittar du på samma sätt pipelineindata och utdata i panelen Pipeline-gränssnitt och komponentens indata och utdata i komponentens panel (utlösare genom att dubbelklicka på komponenten).

Så här höjer du upp komponentindata och utdata till pipelinenivå

När du befordrar en komponents indata/utdata till pipelinenivå kan du skriva över komponentens indata/utdata när du skickar ett pipelinejobb. Det är också användbart om du vill utlösa pipelinen med hjälp av REST-slutpunkten.

Här följer exempel för att höja upp komponentindata/utdata till indata/utdata på pipelinenivå.

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

Det fullständiga exemplet finns i pipelinen train-score-eval med registrerade komponenter. Den här pipelinen höjer upp tre indata och tre utdata till pipelinenivå. Låt oss ta pipeline_job_training_max_epocs som exempel. Det deklareras under inputs avsnittet på rotnivån, vilket innebär att det är indata på pipelinenivå. Under jobs -> train_job avsnittet refereras indata med namnet max_epocs som ${{parent.inputs.pipeline_job_training_max_epocs}}, vilket anger train_job"s-indata max_epocs refererar till pipelinenivåindata pipeline_job_training_max_epocs. På samma sätt kan du höja upp pipelineutdata med samma schema.

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"

Exempel på en notebook-fil från slutpunkt till slutpunkt i azureml-example-lagringsplatsen

Studio

Du kan höja upp en komponents indata till pipelinenivåindata på designredigeringssidan. Gå till komponentens inställningspanel genom att dubbelklicka på komponenten –> leta upp de indata som du vill höja upp –> Välj de tre punkterna till höger –> Välj Lägg till i pipelineindata.

Valfria indata

Som standard krävs alla indata och måste tilldelas ett värde (eller ett standardvärde) varje gång du skickar ett pipelinejobb. Det kan dock finnas instanser där du behöver valfria indata. I sådana fall har du flexibiliteten att inte tilldela ett värde till indata när du skickar ett pipelinejobb.

Valfria indata kan vara användbara i följande två scenarier:

• Om du har en valfri inmatning av data/modelltyp och inte tilldelar ett värde till den när du skickar pipelinejobbet, kommer det att finnas en komponent i pipelinen som saknar ett tidigare databeroende. Med andra ord är indataporten inte länkad till någon komponent eller data/modellnod. Detta gör att pipelinetjänsten anropar den här komponenten direkt i stället för att vänta på att det föregående beroendet ska vara klart.

• Nedan visas ett tydligt exempel på det andra scenariot. Om du anger continue_on_step_failure = True för pipelinen och har en andra nod (node2) som använder utdata från den första noden (nod1) som valfri indata, körs node2 fortfarande även om node1 misslyckas. Men om node2 använder nödvändiga indata från node1 körs den inte om node1 misslyckas.

  

Här följer exempel på hur du definierar valfria indata.

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

När indata anges som optional = truemåste du använda $[[]] för att omfamna kommandoraden med indata. Se markerad rad i exemplet ovan.

Kommentar

Valfria utdata stöds inte.

I pipelinediagrammet representeras valfria indata av data-/modelltypen av en prickad cirkel. Valfria indata för primitiva typer kan finnas under fliken Inställningar . Till skillnad från obligatoriska indata har valfria indata inte någon asterisk bredvid sig, vilket betyder att de inte är obligatoriska.

Så här anpassar du utdatasökvägen

Som standard lagras utdata för en komponent i azureml://datastores/${{default_datastore}}/paths/${{name}}/${{output_name}}. {default_datastore} är standardkunduppsättningen för datalager för pipelinen. Om den inte har angetts är det bloblagring för arbetsytor. {name} är jobbnamnet, som kommer att matchas vid jobbkörningen. {output_name} är det utdatanamn som kunden har definierat i komponenten YAML.

Men du kan också anpassa var utdata ska lagras genom att definiera sökvägen för utdata. Följande är exempel:

Azure CLI

pipeline.yaml Definierar en pipeline som har tre utdata på pipelinenivå. Fullständig YAML finns i pipelinen train-score-eval med exempel på registrerade komponenter. Du kan använda följande kommando för att ange en anpassad utdatasökväg för pipeline_job_trained_modelutdata.

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

Exempel på notebook-filer från slutpunkt till slutpunkt finns i Skapa pipeline med command_component dekorerad python-funktionsanteckningsbok.

Så här laddar du ned utdata

Du kan ladda ned en komponents utdata eller pipelineutdata enligt exemplet nedan.

Ladda ned utdata på pipelinenivå

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

Innan vi går in i koden behöver du ett sätt att referera till din arbetsyta. Du skapar ml_client för en referens till arbetsytan. Information om hur du initierar ml_clientfinns i Skapa referens till arbetsyta.

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

Ladda ned underordnat jobbs utdata

När du behöver ladda ned utdata från ett underordnat jobb (ett komponentutdata som inte höjs upp till pipelinenivå) bör du först lista alla underordnade jobbentiteter för ett pipelinejobb och sedan använda liknande kod för att ladda ned utdata.

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

Innan vi går in i koden behöver du ett sätt att referera till din arbetsyta. Du skapar ml_client för en referens till arbetsytan. Information om hur du initierar ml_clientfinns i Skapa referens till arbetsyta.

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

Registrera utdata som namngiven tillgång

Du kan registrera utdata för en komponent eller pipeline som namngiven tillgång genom att name tilldela och version till utdata. Den registrerade tillgången kan vara en lista på din arbetsyta via Studio UI/CLI/SDK och även refereras till i dina framtida jobb.

Registrera pipelineutdata

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'

Registrera ett underordnat jobbs utdata

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"

Nästa steg

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