파이프라인에서 병렬 작업을 사용하는 방법(V2)

적용 대상:Azure CLI ml 확장 v2(현재)Python SDK azure-ai-ml v2(현재)

병렬 작업을 통해 사용자는 강력한 다중 노드 컴퓨팅 클러스터에 반복 작업을 배포하여 작업 실행을 가속화할 수 있습니다. 큰 이미지 집합에서 개체 감지 모델을 실행하는 시나리오를 예로 들어 보겠습니다. Azure Machine Learning 병렬 작업을 사용하면 이미지를 쉽게 배포하여 특정 컴퓨팅 클러스터에서 사용자 지정 코드를 병렬로 실행할 수 있습니다. 병렬화를 통해 시간 비용을 대폭 절감할 수 있습니다. 또한 Azure Machine Learning 병렬 작업을 사용하면 프로세스를 간소화하고 자동화하여 더 효율적으로 만들 수 있습니다.

전제 조건

Azure Machine Learning 병렬 작업은 파이프라인 작업의 단계 중 하나로만 사용할 수 있습니다. 따라서 파이프라인을 사용하는 데 익숙해지는 것이 중요합니다. Azure Machine Learning 파이프라인에 대해 자세히 알아보려면 다음 문서를 참조하세요.

• Azure Machine Learning 파이프라인을 이해합니다.
• CLI v2 및 SDK v2와 함께 Azure Machine Learning 파이프라인을 사용하는 방법을 이해합니다.

병렬 작업이 필요한 이유

실제 환경의 ML 엔지니어에게는 항상 학습 또는 추론 작업에 대한 스케일링 요구 사항이 있습니다. 예를 들어, 데이터 과학자가 판매 예측 모델을 학습시키는 단일 스크립트를 제공하는 경우 ML 엔지니어는 이 학습 작업을 각 개별 저장소에 적용해야 합니다. 이 스케일 아웃 프로세스 중에는 다음의 몇 가지 문제가 나타납니다.

• 긴 실행 시간으로 인해 발생하는 지연 압력
• 작업을 계속 진행하기 위해 예기치 않은 문제를 처리하는 수동 개입

Azure Machine Learning 병렬 작업의 코어 가치는 단일 직렬 작업을 미니 일괄 처리로 분할하고 이러한 미니 일괄 처리를 여러 컴퓨팅에 디스패치하여 병렬로 실행하는 것입니다. 병렬 작업을 사용하면 다음을 얻을 수 있습니다.

• 엔드투엔드 실행 시간을 대폭 단축
• Azure Machine Learning 병렬 작업의 자동 오류 처리 설정을 사용합니다.

다음과 같은 경우 Azure Machine Learning 병렬 작업 사용을 고려해야 합니다.

• 분할된 데이터를 기반으로 많은 모델을 학습할 계획인 경우
• 대규모 스케일링 일괄 처리 추론 작업을 가속화하려는 경우

병렬 작업 준비

다른 유형의 작업과 달리 병렬 작업에는 준비가 필요합니다. 다음 섹션에 따라 병렬 작업 만들기를 준비합니다.

분배할 입력 선언 및 데이터 분할 설정

병렬 작업을 수행하려면 주요 입력 데이터 하나만 병렬로 분할하고 처리해야 합니다. 주요 입력 데이터는 표 형식 데이터 또는 파일 집합일 수 있습니다. 다른 입력 형식은 다른 데이터 분할 방법을 가질 수 있습니다.

다음 표는 입력 데이터와 데이터 분할 방법 간의 관계를 보여 줍니다.

데이터 형식	Azure Machine Learning 입력 형식	Azure Machine Learning 입력 모드	데이터 분할 방법
파일 목록	mltable 또는 uri_folder	ro_mount 또는 다운로드로 사용 가능한 제품 설명서에서 데이터 공급자 설치 섹션을 참조하세요	크기별(파일 수) 파티션별
표 형식 데이터	mltable	direct	크기별(예상되는 실제 크기) 파티션별

YAML 또는 Python SDK 병렬 작업에서 input_data 특성을 사용하여 주요 입력 데이터를 선언할 수 있습니다. 또한 ${{inputs.<input name>}}을 사용하여 정의된 inputs 병렬 작업 중 하나로 바인딩할 수 있습니다. 그런 다음 다른 특성을 채워 주요 입력에 대한 데이터 분할 방법을 정의해야 합니다.

데이터 분할 방법	Attribute name	특성 유형	작업 예
크기별	mini_batch_size	string	홍채 일괄 처리 예측
파티션별	partition_keys	문자열 목록	오렌지 주스 판매 예측

Azure CLI

적용 대상:Azure CLI ml 확장 v2(현재)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

Python

적용 대상: Python SDK azure-ai-ml v2(현재)

job_data_path를 입력 중 하나로 선언합니다. input_data 특성에 바인딩합니다.

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.0/labels/latest",
    ),
)

데이터 분할 설정을 정의한 후에는 아래 두 가지 특성을 입력하여 병렬화를 위한 리소스 수를 구성할 수 있습니다.

Attribute name	Type	설명	기본값
instance_count	정수	작업에 사용할 노드 수입니다.	1
max_concurrency_per_instance	정수	각 노드의 프로세서 수입니다.	GPU 컴퓨팅의 경우 기본값은 1입니다. CPU 컴퓨팅의 경우 기본값은 코어 수입니다.

이러한 두 특성은 지정된 컴퓨팅 클러스터와 함께 작동합니다.

두 가지 특성을 설정하는 샘플 코드:

Azure CLI

적용 대상:Azure CLI ml 확장 v2(현재)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

Python

적용 대상: Python SDK azure-ai-ml v2(현재)

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.0/labels/latest",
    ),
)

참고 항목

mltable 표 형식을 주요 입력 데이터로 사용하는 경우, 특정 경로 아래에 transformations - read_delimited 섹션이 채워진 MLTABLE 사양 파일이 있어야 합니다. 자세한 예제는 mltable 데이터 자산 만들기를 참조하세요.

항목 스크립트에서 미리 정의된 함수 구현

항목 스크립트는 사용자가 사용자 지정 코드로 미리 정의된 세 가지 함수를 구현해야 하는 단일 Python 파일입니다. Azure Machine Learning 병렬 작업은 아래 다이어그램에 따라 각 프로세서에서 실행합니다.

함수 이름	Required	Description	Input	반환 값
Init()	Y	미니 일괄 처리 실행을 시작하기 전에 일반적으로 준비하는 데 이 함수를 사용합니다. 예를 들어 모델을 글로벌 개체에 로드하는 데 사용합니다.	--	--
Run(mini_batch)	Y	mini_batches 기본 실행 논리를 구현합니다.	mini_batch: 입력 데이터가 표 형식 데이터인 경우 Pandas 데이터 프레임입니다. 입력 데이터가 디렉터리인 경우 파일 경로의 목록입니다.	데이터 프레임, 목록 또는 튜플
Shutdown()	N	컴퓨팅을 풀로 다시 반환하기 전에 사용자 지정 정리 작업을 수행하는 선택적 함수입니다.	--	--

자세한 내용을 보려면 다음 항목 스크립트 예제를 확인하세요.

• 이미지 파일 목록에 대한 이미지 식별
• 표 형식 아이리스 데이터에 대한 아이리스 분류

항목 스크립트가 준비되었으면 다음의 두 특성을 설정하여 병렬 작업에서 사용할 수 있습니다.

Attribute name	Type	설명	기본값
code	string	업로드하여 작업에 사용할 소스 코드 디렉터리의 로컬 경로입니다.
entry_script	string	미리 정의된 병렬 함수의 구현을 포함하는 Python 파일입니다.

두 가지 특성을 설정하는 샘플 코드:

Azure CLI

적용 대상:Azure CLI ml 확장 v2(현재)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

  task:
    type: run_function
    code: "./script"
    entry_script: iris_prediction.py
    environment:
      name: "prs-env"
      version: 1
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
      conda_file: ./environment/environment_parallel.yml
    program_arguments: >-
      --model ${{inputs.score_model}}
      --error_threshold 5
      --allowed_failed_percent 30
      --task_overhead_timeout 1200
      --progress_update_timeout 600
      --first_task_creation_timeout 600
      --copy_logs_to_parent True
      --resource_monitor_interva 20
    append_row_to: ${{outputs.job_output_file}}

Python

적용 대상: Python SDK azure-ai-ml v2(현재)

# parallel task to process file data
file_batch_inference = parallel_run_function(
    name="file_batch_score",
    display_name="Batch Score with File Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        )
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=1,
    mini_batch_size="1",
    mini_batch_error_threshold=1,
    retry_settings=dict(max_retries=2, timeout=60),
    logging_level="DEBUG",
    task=RunFunction(
        code="./src",
        entry_script="file_batch_inference.py",
        program_arguments="--job_output_path ${{outputs.job_output_path}}",
        environment="azureml://registries/azureml/environments/sklearn-1.0/labels/latest",
    ),
)

Important

Run(mini_batch) 함수에는 데이터 프레임, 목록 또는 튜플 항목의 반환이 필요합니다. 병렬 작업은 이 반환 수를 사용하여 해당 미니 일괄 처리에서 성공 항목을 측정합니다. 이 미니 일괄 처리에서 모든 항목이 잘 처리된 경우 원칙적으로 미니 일괄 처리 수는 반환 목록의 수와 같아야 합니다.

Important

Init() 또는 Run(mini_batch) 함수에서 인수를 구문 분석하려는 경우 ‘parse_args’ 대신 ‘parse_known_args’를 사용하여 예외를 방지합니다. 인수 파서가 있는 항목 스크립트는 iris_score 예제를 참조하세요.

Important

mltable을 주요 입력 데이터로 사용하는 경우 환경에 'mltable' 라이브러리를 설치해야 합니다. 이 conda 파일 예의 9행을 참조하세요.

자동화 설정 고려

Azure Machine Learning 병렬 작업은 수동 개입 없이 작업을 자동으로 제어하기 위해 다양한 설정을 노출합니다. 자세한 내용은 다음 표를 참조하세요.

키	형식	설명	허용된 값	기본값	특성에서 설정	프로그램 인수에서 설정
미니 일괄 처리 오류 임계값	정수	이 병렬 작업에서 무시할 수 있는, 실패한 미니 일괄 처리 수를 정의합니다. 실패한 미니 일괄 처리 수가 이 임계값보다 높으면 병렬 작업이 실패한 것으로 표시됩니다. 다음과 같은 경우 미니 일괄 처리가 실패한 것으로 표시됩니다. - run()의 반환 수가 미니 일괄 처리 입력 수보다 적은 경우 - 사용자 지정 run() 코드에서 예외를 catch한 경우 “-1”은 기본 수로, 병렬 작업 중에 실패한 모든 미니 일괄 처리를 무시한다는 의미입니다.	[-1, int.max]	-1	mini_batch_error_threshold	해당 없음
미니 일괄 처리 최대 다시 시도	정수	미니 일괄 처리가 실패하거나 시간이 초과되는 경우의 다시 시도 횟수를 정의합니다. 다시 시도에 모두 실패하면 미니 일괄 처리가 mini_batch_error_threshold 계산에 따라 계산하는 데 실패한 것으로 표시됩니다.	[0, int.max]	2	retry_settings.max_retries	해당 없음
미니 일괄 처리 시간 제한	정수	사용자 지정 run() 함수 실행 시 시간 제한(초)을 정의합니다. 실행 시간이 이 임계값보다 길면 미니 일괄 처리가 중단되고, 다시 시도를 트리거하는 데 실패한 미니 일괄 처리로 표시됩니다.	(0, 259200]	60	retry_settings.timeout	해당 없음
항목 오류 임계값	정수	실패한 항목의 임계값입니다. 실패한 항목은 각 미니 일괄 처리의 입력과 반환 간의 숫자 차이로 계산됩니다. 실패한 항목의 합계가 이 임계값보다 크면 병렬 작업이 실패한 것으로 표시됩니다. 참고: ‘-1’은 기본 수로, 병렬 작업 중에 모든 오류를 무시한다는 의미입니다.	[-1, int.max]	-1	해당 없음	--error_threshold
허용되는 실패율	정수	mini_batch_error_threshold와 비슷하지만, 개수 대신 실패한 미니 일괄 처리의 백분율을 사용합니다.	[0, 100]	100	해당 없음	--allowed_failed_percent
오버헤드 시간 제한	정수	각 미니 일괄 처리의 초기화에 대한 시간 제한(초)입니다. 예를 들면, 미니 일괄 처리 데이터를 로드하여 run() 함수에 전달합니다.	(0, 259200]	600	해당 없음	--task_overhead_timeout
진행률 업데이트 시간 제한	정수	미니 일괄 처리 실행의 진행률 모니터링에 대한 시간 제한(초)입니다. 진행률 업데이트를 이 시간 제한 설정 내에 받지 못하면 병렬 작업이 실패한 것으로 표시됩니다.	(0, 259200]	다른 설정에 따라 동적으로 계산됩니다.	해당 없음	--progress_update_timeout
첫 번째 작업 만들기 시간 제한	정수	작업 시작과 첫 번째 미니 일괄 처리 실행 사이의 시간을 모니터링하기 위한 시간 제한(초)입니다.	(0, 259200]	600	해당 없음	--first_task_creation_timeout
로깅 수준	string	사용자 로그 파일에 덤프할 로그 수준을 정의합니다.	정보, 경고 또는 디버그	INFO	logging_level	해당 없음
다음에 행 추가	string	각 미니 일괄 처리 실행의 모든 반환을 집계하고 이 파일에 출력합니다. ${{outputs.<output_name>}} 식을 사용하여 병렬 작업의 출력 중 하나를 참조할 수 있습니다.			task.append_row_to	해당 없음
부모에 로그 복사	string	작업 진행률, 개요 및 로그를 부모 파이프라인 작업에 복사할지 여부에 대한 부울 옵션입니다.	True 또는 False	False	해당 없음	--copy_logs_to_parent
리소스 모니터 간격	정수	노드 리소스 사용량(예: cpu, 메모리)을 "logs/sys/perf" 경로 아래의 로그 폴더에 덤프하는 시간 간격(초)입니다. 참고: 덤프 리소스 로그를 자주 사용하면 미니 일괄 처리의 실행 속도가 약간 느려집니다. 리소스 사용량 덤프를 중지하려면 이 값을 ‘0’으로 설정합니다.	[0, int.max]	600	해당 없음	--resource_monitor_interval

이러한 설정을 업데이트하는 샘플 코드:

Azure CLI

적용 대상:Azure CLI ml 확장 v2(현재)

batch_prediction:
  type: parallel
  compute: azureml:cpu-cluster
  inputs:
    input_data: 
      type: mltable
      path: ./neural-iris-mltable
      mode: direct
    score_model: 
      type: uri_folder
      path: ./iris-model
      mode: download
  outputs:
    job_output_file:
      type: uri_file
      mode: rw_mount

  input_data: ${{inputs.input_data}}
  mini_batch_size: "10kb"
  resources:
      instance_count: 2
  max_concurrency_per_instance: 2

  logging_level: "DEBUG"
  mini_batch_error_threshold: 5
  retry_settings:
    max_retries: 2
    timeout: 60

  task:
    type: run_function
    code: "./script"
    entry_script: iris_prediction.py
    environment:
      name: "prs-env"
      version: 1
      image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
      conda_file: ./environment/environment_parallel.yml
    program_arguments: >-
      --model ${{inputs.score_model}}
      --error_threshold 5
      --allowed_failed_percent 30
      --task_overhead_timeout 1200
      --progress_update_timeout 600
      --first_task_creation_timeout 600
      --copy_logs_to_parent True
      --resource_monitor_interva 20
    append_row_to: ${{outputs.job_output_file}}

Python

적용 대상: Python SDK azure-ai-ml v2(현재)

# parallel task to process tabular data
tabular_batch_inference = parallel_run_function(
    name="batch_score_with_tabular_input",
    display_name="Batch Score with Tabular Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        ),
        score_model=Input(
            type=AssetTypes.URI_FOLDER, description="The model for batch score."
        ),
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=2,
    mini_batch_size="100",
    mini_batch_error_threshold=5,
    logging_level="DEBUG",
    retry_settings=dict(max_retries=2, timeout=60),
    task=RunFunction(
        code="./src",
        entry_script="tabular_batch_inference.py",
        environment=Environment(
            image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
            conda_file="./src/environment_parallel.yml",
        ),
        program_arguments="--model ${{inputs.score_model}} "
        "--job_output_path ${{outputs.job_output_path}} "
        "--error_threshold 5 "
        "--allowed_failed_percent 30 "
        "--task_overhead_timeout 1200 "
        "--progress_update_timeout 600 "
        "--first_task_creation_timeout 600 "
        "--copy_logs_to_parent True "
        "--resource_monitor_interva 20 ",
        append_row_to="${{outputs.job_output_path}}",
    ),
)

파이프라인에서 병렬 작업 만들기

Azure CLI

적용 대상:Azure CLI ml 확장 v2(현재)

파이프라인 작업을 사용하여 병렬 작업을 인라인으로 만들 수 있습니다.

$schema: https://azuremlschemas.azureedge.net/latest/pipelineJob.schema.json
type: pipeline

display_name: iris-batch-prediction-using-parallel
description: The hello world pipeline job with inline parallel job
tags:
  tag: tagvalue
  owner: sdkteam

settings:
  default_compute: azureml:cpu-cluster

jobs:
  batch_prediction:
    type: parallel
    compute: azureml:cpu-cluster
    inputs:
      input_data: 
        type: mltable
        path: ./neural-iris-mltable
        mode: direct
      score_model: 
        type: uri_folder
        path: ./iris-model
        mode: download
    outputs:
      job_output_file:
        type: uri_file
        mode: rw_mount

    input_data: ${{inputs.input_data}}
    mini_batch_size: "10kb"
    resources:
        instance_count: 2
    max_concurrency_per_instance: 2

    logging_level: "DEBUG"
    mini_batch_error_threshold: 5
    retry_settings:
      max_retries: 2
      timeout: 60

    task:
      type: run_function
      code: "./script"
      entry_script: iris_prediction.py
      environment:
        name: "prs-env"
        version: 1
        image: mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04
        conda_file: ./environment/environment_parallel.yml
      program_arguments: >-
        --model ${{inputs.score_model}}
        --error_threshold 5
        --allowed_failed_percent 30
        --task_overhead_timeout 1200
        --progress_update_timeout 600
        --first_task_creation_timeout 600
        --copy_logs_to_parent True
        --resource_monitor_interva 20
      append_row_to: ${{outputs.job_output_file}}

Python

적용 대상: Python SDK azure-ai-ml v2(현재)

먼저 필요한 라이브러리를 가져와서 적절한 자격 증명으로 ml_client를 시작한 후 컴퓨팅을 만들거나 검색해야 합니다.

# import required libraries
from azure.identity import DefaultAzureCredential, InteractiveBrowserCredential
from azure.ai.ml import MLClient, Input, Output, load_component
from azure.ai.ml.dsl import pipeline
from azure.ai.ml.entities import Environment
from azure.ai.ml.constants import AssetTypes, InputOutputModes
from azure.ai.ml.parallel import parallel_run_function, RunFunction

try:
    credential = DefaultAzureCredential()
    # Check if given credential can get token successfully.
    credential.get_token("https://management.azure.com/.default")
except Exception as ex:
    # Fall back to InteractiveBrowserCredential in case DefaultAzureCredential not work
    credential = InteractiveBrowserCredential()

# Get a handle to workspace
ml_client = MLClient.from_config(credential=credential)

# Retrieve an already attached Azure Machine Learning Compute.
cpu_compute_target = "cpu-cluster"
print(ml_client.compute.get(cpu_compute_target))
gpu_compute_target = "gpu-cluster"
print(ml_client.compute.get(gpu_compute_target))

그런 다음, parallel_run_function을 입력하여 병렬 작업을 구현합니다.

# parallel task to process tabular data
tabular_batch_inference = parallel_run_function(
    name="batch_score_with_tabular_input",
    display_name="Batch Score with Tabular Dataset",
    description="parallel component for batch score",
    inputs=dict(
        job_data_path=Input(
            type=AssetTypes.MLTABLE,
            description="The data to be split and scored in parallel",
        ),
        score_model=Input(
            type=AssetTypes.URI_FOLDER, description="The model for batch score."
        ),
    ),
    outputs=dict(job_output_path=Output(type=AssetTypes.MLTABLE)),
    input_data="${{inputs.job_data_path}}",
    instance_count=2,
    max_concurrency_per_instance=2,
    mini_batch_size="100",
    mini_batch_error_threshold=5,
    logging_level="DEBUG",
    retry_settings=dict(max_retries=2, timeout=60),
    task=RunFunction(
        code="./src",
        entry_script="tabular_batch_inference.py",
        environment=Environment(
            image="mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04",
            conda_file="./src/environment_parallel.yml",
        ),
        program_arguments="--model ${{inputs.score_model}} "
        "--job_output_path ${{outputs.job_output_path}} "
        "--error_threshold 5 "
        "--allowed_failed_percent 30 "
        "--task_overhead_timeout 1200 "
        "--progress_update_timeout 600 "
        "--first_task_creation_timeout 600 "
        "--copy_logs_to_parent True "
        "--resource_monitor_interva 20 ",
        append_row_to="${{outputs.job_output_path}}",
    ),
)

마지막으로 병렬 작업을 파이프라인의 단계로 사용하고, 입력/출력을 다른 단계와 바인딩합니다.

@pipeline()
def parallel_in_pipeline(pipeline_job_data_path, pipeline_score_model):

    prepare_file_tabular_data = prepare_data(input_data=pipeline_job_data_path)
    # output of file & tabular data should be type MLTable
    prepare_file_tabular_data.outputs.file_output_data.type = AssetTypes.MLTABLE
    prepare_file_tabular_data.outputs.tabular_output_data.type = AssetTypes.MLTABLE

    batch_inference_with_file_data = file_batch_inference(
        job_data_path=prepare_file_tabular_data.outputs.file_output_data
    )
    # use eval_mount mode to handle file data
    batch_inference_with_file_data.inputs.job_data_path.mode = (
        InputOutputModes.EVAL_MOUNT
    )
    batch_inference_with_file_data.outputs.job_output_path.type = AssetTypes.MLTABLE

    batch_inference_with_tabular_data = tabular_batch_inference(
        job_data_path=prepare_file_tabular_data.outputs.tabular_output_data,
        score_model=pipeline_score_model,
    )
    # use direct mode to handle tabular data
    batch_inference_with_tabular_data.inputs.job_data_path.mode = (
        InputOutputModes.DIRECT
    )

    return {
        "pipeline_job_out_file": batch_inference_with_file_data.outputs.job_output_path,
        "pipeline_job_out_tabular": batch_inference_with_tabular_data.outputs.job_output_path,
    }


pipeline_job_data_path = Input(
    path="./dataset/", type=AssetTypes.MLTABLE, mode=InputOutputModes.RO_MOUNT
)
pipeline_score_model = Input(
    path="./model/", type=AssetTypes.URI_FOLDER, mode=InputOutputModes.DOWNLOAD
)
# create a pipeline
pipeline_job = parallel_in_pipeline(
    pipeline_job_data_path=pipeline_job_data_path,
    pipeline_score_model=pipeline_score_model,
)
pipeline_job.outputs.pipeline_job_out_tabular.type = AssetTypes.URI_FILE

# set pipeline level compute
pipeline_job.settings.default_compute = "cpu-cluster"

Studio UI에서 파이프라인 작업 제출 및 병렬 단계 확인

Azure CLI

적용 대상:Azure CLI ml 확장 v2(현재)

CLI 명령을 사용하여 병렬 단계로 파이프라인 작업을 제출할 수 있습니다.

az ml job create --file pipeline.yml

Python

적용 대상: Python SDK azure-ai-ml v2(현재)

ml_client의 jobs.create_or_update 함수를 사용하여 병렬 단계로 파이프라인 작업을 제출할 수 있습니다.

pipeline_job = ml_client.jobs.create_or_update(
    pipeline_job, experiment_name="pipeline_samples"
)
pipeline_job

파이프라인 작업을 제출하면 SDK 또는 CLI 위젯에서 Studio UI에 대한 웹 URL 링크를 제공합니다. 링크는 기본적으로 파이프라인 그래프 보기로 안내합니다. 병렬 단계를 두 번 선택하여 병렬 작업의 오른쪽 패널을 엽니다.

병렬 작업의 설정을 확인하려면 매개 변수 탭으로 이동하여 설정 실행을 확장하고 병렬 섹션을 선택합니다.

병렬 작업의 실패를 디버그하려면 출력 + 로그 탭으로 이동하여 왼쪽의 출력 디렉터리에서 logs 폴더를 확장하고, job_result.txt를 확인하여 병렬 작업에 실패한 이유를 이해합니다. 병렬 작업의 로깅 구조에 대한 자세한 내용은 동일한 폴더에서 readme.txt 참조하세요.

파이프라인 병렬 작업의 예제

• Azure CLI + YAML 리포지토리 예
• SDK 리포지토리 예

다음 단계

• 병렬 작업의 자세한 yaml 스키마는 병렬 작업용 YAML 참조를 참조하세요.
• MLTABLE에 데이터를 온보딩하는 방법은 mltable 데이터 자산 만들기를 참조하세요.
• 파이프라인을 정기적으로 트리거하는 방법은 파이프라인 예약 방법을 참조하세요.