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

1. En primer lugar, es necesario obtener la canalización que queremos publicar:

   pipeline1 = Pipeline(workspace=ws, steps=[step1, step2])
   

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

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

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

2. 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
   

3. 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)
   

4. 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)
   

5. 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)
   

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

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

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)

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