Atualizar pontos de extremidade de pipeline para SDK v2

Artigo
09/01/2024

Depois de ter um pipeline instalado e em execução, você pode publicar um pipeline para que ele seja executado com entradas diferentes. Isso era conhecido como Pipelines Publicados.

O que mudou?

O Batch Endpoint propõe uma maneira semelhante, ainda mais poderosa, de lidar com vários ativos executados sob uma API durável, e é por isso que a funcionalidade Pipelines publicados foi movida para implantações de componentes Pipeline em pontos de extremidade em lote.

Os pontos de extremidade em lote separam a interface (ponto de extremidade) da implementação real (implantação) e permitem que o usuário decida qual implantação serve a implementação padrão do ponto de extremidade. As implantações de componentes de pipeline em pontos de extremidade em lote permitem que os usuários implantem componentes de pipeline em vez de pipelines, que fazem um melhor uso de ativos reutilizáveis para as organizações que procuram simplificar sua prática de MLOps.

A tabela a seguir mostra uma comparação de cada um dos conceitos:

Conceito	SDK v1	SDK v2
Ponto de extremidade REST do pipeline para invocação	Ponto de extremidade do pipeline	Ponto final em lote
Versão específica do pipeline sob o endpoint	Pipeline publicado	Implantação de componentes de pipeline
Argumentos da Pipeline sobre a invocação	Parâmetro do pipeline	Entradas de trabalho
Trabalho gerado a partir de um pipeline publicado	Trabalho de pipeline	Tarefa de lote

Para saber como criar sua primeira implantação de componente de pipeline, consulte Como implantar pipelines em pontos de extremidade em lote.

Movendo-se para pontos de extremidade de lote

Use as diretrizes a seguir para saber como mover do SDK v1 para o SDK v2 usando os conceitos em Pontos de extremidade em lote.

Publicar um pipeline

Compare como a publicação de um pipeline mudou de v1 para v2:

SDK v1
SDK v2

Primeiro, precisamos obter o pipeline que queremos publicar:
```
pipeline1 = Pipeline(workspace=ws, steps=[step1, step2])
```

Podemos publicar o pipeline da seguinte forma:

from azureml.pipeline.core import PipelineEndpoint

endpoint_name = "PipelineEndpointTest"
pipeline_endpoint = PipelineEndpoint.publish(
    workspace=ws, 
    name=endpoint_name,
    pipeline=pipeline, 
    description="A hello world endpoint for component deployments"
)

Primeiro, precisamos definir o pipeline que queremos publicar.

@pipeline()
def pipeline(input_data: Input(type=AssetTypes.URI_FOLDER)):
    (...)

    return {
        (..)
    }

Os pontos de extremidade em lote não implantam pipelines, mas componentes de pipeline. Os componentes propõem uma maneira mais confiável de ter controle do código-fonte dos ativos que estão sendo implantados sob um ponto de extremidade. Podemos converter qualquer definição de pipeline em um componente de pipeline da seguinte maneira:
```
pipeline_component = pipeline().component
```
Como prática recomendada, recomendamos registrar componentes de pipeline para que você possa manter o controle de versão deles de forma centralizada dentro do espaço de trabalho ou até mesmo nos registros compartilhados.
```
ml_client.components.create(pipeline_component)
```

Em seguida, precisamos criar o ponto de extremidade que hospeda todas as implantações de pipeline:

endpoint_name = "PipelineEndpointTest"
endpoint = BatchEndpoint(
    name=endpoint_name,
    description="A hello world endpoint for component deployments",
)

ml_client.batch_endpoints.begin_create_or_update(endpoint)

Crie uma implantação para o componente de pipeline:

deployment_name = "hello-batch-dpl"
deployment = BatchPipelineComponentDeployment(
    name=deployment_name,
    description="A hello world deployment with a single step.",
    endpoint_name=endpoint.name,
    component=pipeline_component
)

ml_client.batch_deployments.begin_create_or_update(deployment)

Enviar um trabalho para um ponto de extremidade de pipeline

SDK v1
SDK v2

Para chamar a versão padrão do pipeline, você pode usar:

pipeline_endpoint = PipelineEndpoint.get(workspace=ws, name="PipelineEndpointTest")
run_id = pipeline_endpoint.submit("PipelineEndpointExperiment")

job = ml_client.batch_endpoints.invoke(
    endpoint_name=batch_endpoint, 
)

Use inputs para indicar as entradas do trabalho, se necessário. Consulte Criar trabalhos e dados de entrada para pontos de extremidade em lote para obter uma explicação mais detalhada sobre como indicar entradas e saídas.

job = ml_client.batch_endpoints.invoke(
    endpoint_name=batch_endpoint,
    inputs={
        "input_data": Input(type=AssetTypes.URI_FOLDER, path="./my_local_data")
    }
)

Você também pode enviar um trabalho para uma versão específica:

SDK v1
SDK v2

run_id = pipeline_endpoint.submit(endpoint_name, pipeline_version="0")

Em pontos de extremidade em lote, as implantações não são versionadas. No entanto, você pode implantar várias versões de componentes de pipeline sob o mesmo ponto de extremidade. Nesse sentido, cada versão de pipeline na v1 corresponde a uma versão diferente do componente de pipeline e sua implantação correspondente sob o ponto de extremidade.

Em seguida, você pode implantar uma implantação específica em execução sob o ponto de extremidade se essa implantação executar a versão em que você está interessado.

job = ml_client.batch_endpoints.invoke(
    endpoint_name=endpoint_name, 
    deployment_name=deployment_name, 
)

all_pipelines = PublishedPipeline.get_all(ws)

O código a seguir lista todos os pontos de extremidade existentes no espaço de trabalho:

all_endpoints = ml_client.batch_endpoints.list()

No entanto, lembre-se de que os pontos de extremidade em lote podem hospedar implantações operacionalizando pipelines ou modelos. Se quiser obter uma lista de todas as implantações que hospedam pipelines, você pode fazer o seguinte:

all_deployments = []

for endpoint in all_endpoints:
    all_deployments.extend(ml_client.batch_deployments.list(endpoint_name=endpoint.name))

all_pipeline_deployments = filter(all_endpoints, lamdba x: x is BatchPipelineComponentDeployment)

Utilizar a API REST

Você pode criar trabalhos a partir dos pontos de extremidade usando a API REST da URL de invocação. Veja os exemplos a seguir para ver como a invocação foi alterada de v1 para v2.

SDK v1
SDK v2

pipeline_endpoint = PipelineEndpoint.get(workspace=ws, name=endpoint_name)
rest_endpoint = pipeline_endpoint.endpoint

response = requests.post(
    rest_endpoint, 
    headers=aad_token, 
    json={
        "ExperimentName": "PipelineEndpointExperiment",
        "RunSource": "API",
        "ParameterAssignments": {"argument1": "united", "argument2":45}
    }
)

Os pontos de extremidade em lote suportam vários tipos de entradas. O exemplo a seguir mostra como indicar duas entradas diferentes do tipo string e numeric. Consulte Criar trabalhos e dados de entrada para pontos de extremidade em lote (REST) para obter exemplos mais detalhados:

batch_endpoint = ml_client.batch_endpoints.get(endpoint_name)
rest_endpoint = batch_endpoint.invocation_url

response = requests.post(
    rest_endpoint, 
    headers=aad_token, 
    json={
        "properties": {
            "InputData": {
                "argument1": {
                    "JobInputType": "Literal",
                    "Value": "united"
                },
                "argument2": {
                    "JobInputType": "Literal",
                    "Value": 45
                }
            }
        }
    }
)

Partilhar via