パイプライン エンドポイントを SDK v2 にアップグレードする

パイプラインを起動して実行すると、パイプラインを発行することができるため、さまざまな入力で実行されます。 これは公開済みパイプラインと呼ばれていたものです。

どのような点が変わりましたか?

Batch Endpoint では、永続的な API で実行されている複数の資産を処理するための同様の強力な方法が提案されています。そのため、発行されたパイプライン機能はバッチ エンドポイントのパイプライン コンポーネント デプロイに移動されました。

バッチ エンドポイントは、インターフェイス (エンドポイント) を実際の実装 (デプロイ) から切り離し、エンドポイントの既定の実装を提供するデプロイをユーザーが決定できるようにします。 バッチ エンドポイントでパイプライン コンポーネントのデプロイを使用すると、ユーザーはパイプラインではなくパイプライン コンポーネントをデプロイできるようになります。これにより、MLOps プラクティスを効率化しようとしている組織向けに、再利用可能なアセットをより有効に使用できます。

次の表は、各概念の比較を示しています。

概念	SDK v1	SDK v2
呼び出し用のパイプラインの REST エンドポイント	パイプライン エンドポイント	バッチ エンドポイント
エンドポイント下のパイプラインの特定のバージョン	公開されたパイプライン	パイプライン コンポーネント デプロイ
呼び出し時のパイプラインの引数	パイプライン パラメーター	ジョブの入力
公開されたパイプラインから生成されるジョブ	パイプライン ジョブ	バッチ ジョブ

最初のパイプライン コンポーネント デプロイを作成する方法については、バッチ エンドポイントでパイプラインをデプロイする方法に関するページを参照してください。

バッチ エンドポイントへの移動

バッチ エンドポイントの概念を使用して SDK v1 から SDK v2 に移行する方法については、次のガイドラインを使用してください。

パイプラインを発行する

パイプラインの公開が v1 から v2 でどのように変わったかを比較します。

SDK v1

1. まず、公開するパイプラインを取得する必要があります。

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

2. パイプラインは次のようにして公開できます。

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

SDK v2

1. まず、発行するパイプラインを定義する必要があります。

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

2. バッチ エンドポイントはパイプラインではなくパイプライン コンポーネントをデプロイします。 コンポーネントは、エンドポイントの下にデプロイされている資産のソース管理をより信頼性の高い方法で提案します。 次のように、任意のパイプライン定義をパイプライン コンポーネントに変換できます。

   pipeline_component = pipeline().component
   

3. ベスト プラクティスとして、ワークスペース内、さらには共有レジストリ内で一元化された方法でバージョン管理を維持できるよう、パイプライン コンポーネントを登録することをお勧めします。

   ml_client.components.create(pipeline_component)
   

4. 次に、すべてのパイプライン デプロイをホストするエンドポイントを作成する必要があります。

   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. パイプライン コンポーネントのデプロイを作成します。

   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)
   

パイプライン エンドポイントにジョブを送信する

SDK v1

パイプラインの既定のバージョンを呼び出すには、次を使用できます。

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

SDK v2

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

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

SDK v2

バッチ エンドポイントでは、デプロイはバージョン管理されません。 ただし、同じエンドポイント下に複数のパイプライン コンポーネント バージョンをデプロイできます。 この意味で、v1 の各パイプライン バージョンは、異なるパイプライン コンポーネント のバージョンとそれに対応するエンドポイントの下のデプロイに対応します。

その後、エンドポイントの下で実行されている特定のデプロイをデプロイできます (そのデプロイで関心のあるバージョンが実行されている場合)。

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

すべてのパイプラインをデプロイする

SDK v1

all_pipelines = PublishedPipeline.get_all(ws)

SDK v2

次のコードは、ワークスペースに存在するすべてのエンドポイントを一覧表示します。

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

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

SDK v2

バッチ エンドポイントでは、複数の入力型がサポートされています。 次の例は、型 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
                }
            }
        }
    }
)

次のステップ

• バッチ エンドポイントでパイプラインをデプロイする方法
• バッチ エンドポイントを使用してトレーニング パイプラインを運用化する方法
• バッチ エンドポイントでスコアリング ルーチンを運用化する方法