Actualización de puntos de conexión de canalización a SDK v2

Artículo
09/01/2024

Una vez que tenga una canalización en funcionamiento, puede publicarla para que se ejecute con diferentes entradas. Esto se conoce como Canalizaciones publicadas.

¿Qué ha cambiado?

El punto de conexión de Batch propone una manera similar y más eficaz de controlar varios recursos que se ejecutan en una API duradera, por lo que la funcionalidad Canalizaciones publicadas se movió a implementaciones de componentes de canalización en puntos de conexión por lotes.

Los puntos de conexión por lotes desacoplan la interfaz (punto de conexión) de la implementación real (implementación) y permiten al usuario decidir qué implementación sirve para la implementación predeterminada del punto de conexión. Las implementaciones de componentes de canalización en puntos de conexión por lotes permiten a los usuarios implementar componentes de canalización en lugar de canalizaciones, lo que mejora el uso de recursos reutilizables para esas organizaciones que buscan simplificar su práctica de MLOps.

En la tabla siguiente se muestra una comparación de cada uno de los conceptos:

Concepto	SDK v1	SDK v2
Punto de conexión de REST de la canalización para la invocación	Punto de conexión de canalización	Punto de conexión por lotes
Versión específica de la canalización en el punto de conexión	Canalización publicada	Implementación de componentes de canalización
Argumentos de canalización en la invocación	Parámetro Pipeline	Entradas de trabajo
Trabajo generado a partir de una canalización publicada	Trabajo de canalización	Trabajo por lotes

Para obtener información sobre cómo crear la primera implementación de componentes de canalización, consulte Implementación de canalizaciones en puntos de conexión por lotes.

Mover a puntos de conexión por lotes

Use las instrucciones siguientes para obtener información sobre cómo mover de SDK v1 a SDK v2 mediante los conceptos de puntos de conexión por lotes.

Publicar una canalización

Compare cómo ha cambiado la publicación de una canalización de v1 a v2:

SDK v1
SDK v2

En primer lugar, es necesario obtener la canalización que queremos publicar:
```
pipeline1 = Pipeline(workspace=ws, steps=[step1, step2])
```

Podemos publicar la canalización de la siguiente manera:

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

En primer lugar, es necesario definir la canalización que queremos publicar.

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

    return {
        (..)
    }

Los puntos de conexión de Batch no implementan canalizaciones, sino componentes de canalización. Los componentes ofrecen una manera más confiable de tener el control de código fuente de los recursos que se implementan en un punto de conexión. Podemos convertir cualquier definición de canalización en un componente de canalización de la siguiente manera:
```
pipeline_component = pipeline().component
```
Como procedimiento recomendado, se recomienda registrar los componentes de canalización para que pueda mantener el control de versiones de ellos de forma centralizada dentro del área de trabajo o incluso los registros compartidos.
```
ml_client.components.create(pipeline_component)
```

A continuación, es necesario crear el punto de conexión que hospeda todas las implementaciones de canalización:

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)

Cree una implementación para el componente de canalización:

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)

Envío de un trabajo a un punto de conexión de canalización

SDK v1
SDK v2

Para llamar a la versión predeterminada de la canalización, puede 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 las entradas del trabajo si es necesario. Consulte Creación de trabajos y datos de entrada para puntos de conexión por lotes para obtener una explicación más detallada sobre cómo indicar entradas y salidas.

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

También puede enviar un trabajo a una versión específica:

SDK v1
SDK v2

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

En los puntos de conexión por lotes, las implementaciones no tienen versiones. Sin embargo, puede implementar varias versiones de componentes de canalización en el mismo punto de conexión. En este sentido, cada versión de canalización de v1 corresponde a una versión de componente de canalización diferente y su implementación correspondiente en el punto de conexión.

A continuación, puede implementar una implementación específica que se ejecuta en el punto de conexión si esa implementación ejecuta la versión en la que me interesa.

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

Obtención de todas las canalizaciones implementadas

SDK v1
SDK v2

all_pipelines = PublishedPipeline.get_all(ws)

En el código siguiente se enumeran todos los puntos de conexión existentes en el área de trabajo:

all_endpoints = ml_client.batch_endpoints.list()

Sin embargo, tenga en cuenta que los puntos de conexión por lotes pueden hospedar implementaciones que ponen en operación canalizaciones o modelos. Si desea obtener una lista de todas las implementaciones que hospedan canalizaciones, puede hacer lo siguiente:

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)

Mediante la API de REST

Puede crear trabajos a partir de los puntos de conexión mediante la API de REST de la dirección URL de invocación. Consulte los ejemplos siguientes para ver cómo ha cambiado la invocación de v1 a 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}
    }
)

Los puntos de conexión por lotes admiten varios tipos de entradas. En el ejemplo siguiente se muestra cómo indicar dos entradas diferentes de tipo string y numeric. Consulte Creación de trabajos y datos de entrada para puntos de conexión por lotes (REST) para obtener ejemplos más detallados:

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

Compartir a través de