파이프라인 엔드포인트를 SDK v2로 업그레이드

아티클
02/14/2024

파이프라인이 실행되고 나면 다른 입력으로 실행되도록 파이프라인을 게시할 수 있습니다. 이를 게시된 파이프라인이라고 합니다.

무엇이 변경되었습니까?

Batch 엔드포인트는 지속성 API에서 실행되는 여러 자산을 처리하는 보다 강력한 방법을 제안하므로 게시된 파이프라인 기능이 일괄 처리 엔드포인트에서 파이프라인 구성 요소 배포로 이동되었습니다.

일괄 처리 엔드포인트는 실제 구현(배포)에서 인터페이스(엔드포인트)를 분리하고 사용자가 엔드포인트의 기본 구현을 제공하는 배포를 결정할 수 있게 해줍니다. 일괄 처리 엔드포인트의 파이프라인 구성 요소 배포를 통해 사용자는 파이프라인 대신 파이프라인 구성 요소를 배포할 수 있으므로 MLOps 실행을 간소화하려는 조직에서 재사용 가능한 자산을 더 효과적으로 활용할 수 있습니다.

다음 표에서는 각 개념을 비교합니다.

개념	SDK v1	SDK v2
호출을 위한 파이프라인의 REST 엔드포인트	파이프라인 엔드포인트	일괄 처리 엔드포인트
엔드포인트 아래 파이프라인의 특정 버전	게시된 파이프라인	파이프라인 구성 요소 배포
호출에 대한 파이프라인의 인수	파이프라인 매개 변수	작업 입력
게시된 파이프라인에서 생성된 작업	파이프라인 작업	일괄 처리 작업

첫 번째 파이프라인 구성 요소 배포를 만드는 방법을 알아보려면 일괄 처리 엔드포인트에서 파이프라인을 배포하는 방법을 참조하세요.

일괄 처리 엔드포인트로 이동

일괄 처리 엔드포인트의 개념을 사용하여 SDK v1에서 SDK v2로 이동하는 방법을 알아보려면 다음 지침을 따릅니다.

파이프라인 게시

파이프라인 게시가 v1에서 v2로 어떻게 변경되었는지 비교합니다.

SDK v1
SDK v2

먼저 게시하려는 파이프라인을 가져와야 합니다.
```
pipeline1 = Pipeline(workspace=ws, steps=[step1, step2])
```

다음과 같이 파이프라인을 게시할 수 있습니다.

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

먼저 게시하려는 파이프라인을 정의해야 합니다.

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

    return {
        (..)
    }

일괄 처리 엔드포인트는 파이프라인이 아니라 파이프라인 구성 요소를 배포합니다. 구성 요소는 엔드포인트에서 배포되는 자산의 소스 제어를 위한 보다 안정적인 방법을 제안합니다. 다음과 같이 파이프라인 정의를 파이프라인 구성 요소로 변환할 수 있습니다.
```
pipeline_component = pipeline().component
```
가장 좋은 방법은 파이프라인 구성 요소를 등록하여 작업 영역이나 공유 레지스트리 내에서 중앙 집중식으로 버전 관리를 유지할 수 있도록 하는 것입니다.
```
ml_client.components.create(pipeline_component)
```

그런 다음 모든 파이프라인 배포를 호스팅하는 엔드포인트를 만들어야 합니다.

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)

파이프라인 구성 요소에 대한 배포를 만듭니다.

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)

파이프라인의 기본 버전을 호출하려면 다음을 사용할 수 있습니다.

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

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

필요한 경우 작업의 입력을 나타내는 데 사용합니다 inputs . 입력 및 출력을 나타내는 방법에 대한 자세한 설명은 일괄 처리 엔드포인트에 대한 작업 및 입력 데이터 만들기를 참조하세요.

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

특정 버전에 작업을 제출할 수도 있습니다.

SDK v1
SDK v2

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

일괄 처리 엔드포인트에서는 배포 버전이 지정되지 않습니다. 그러나 동일한 엔드포인트 아래에 여러 파이프라인 구성 요소 버전을 배포할 수 있습니다. 이러한 의미에서 v1의 각 파이프라인 버전은 다른 파이프라인 구성 요소 버전 및 엔드포인트의 해당 배포에 해당합니다.

그런 다음, 해당 배포가 관심 있는 버전을 실행하는 경우 엔드포인트에서 실행되는 특정 배포를 배포할 수 있습니다.

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

모든 파이프라인 배포

SDK v1
SDK v2

all_pipelines = PublishedPipeline.get_all(ws)

다음 코드는 작업 영역에 존재하는 모든 엔드포인트를 나열합니다.

all_endpoints = ml_client.batch_endpoints.list()

하지만 일괄 처리 엔드포인트는 파이프라인이나 모델을 운영하는 배포를 호스팅할 수 있다는 점에 유의해야 합니다. 파이프라인을 호스팅하는 모든 배포 목록을 가져오려면 다음과 같이 하면 됩니다.

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)

REST API 사용

호출 URL의 REST API를 사용하여 엔드포인트에서 작업을 만들 수 있습니다. 호출이 v1에서 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}
    }
)

일괄 처리 엔드포인트는 여러 입력 형식을 지원합니다. 다음 예제에서는 형식 string 및 numeric형식의 서로 다른 두 입력을 나타내는 방법을 보여 줍니다. 자세한 예제는 REST(일괄 처리 엔드포인트)에 대한 작업 및 입력 데이터 만들기를 참조하세요.

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