كيفية استخدام مهمة متوازية في البنية الأساسية لبرنامج ربط العمليات التجارية (V2)

ينطبق على:ملحق ML Azure CLI v2 (الحالي)Python SDK azure-ai-ml v2 (الحالي)

تتيح الوظيفة المتوازية للمستخدمين تسريع تنفيذ وظائفهم عن طريق توزيع المهام المتكررة على مجموعات حساب قوية متعددة العقد. على سبيل المثال، خذ السيناريو حيث تقوم بتشغيل نموذج الكشف عن الكائنات على مجموعة كبيرة من الصور. باستخدام Azure التعلم الآلي Parallel job، يمكنك بسهولة توزيع صورك لتشغيل التعليمات البرمجية المخصصة بالتوازي على مجموعة حوسبة معينة. ويمكن للتوازي أن يقلل بشكل كبير من تكلفة الوقت. أيضا باستخدام Azure التعلم الآلي مهمة متوازية يمكنك تبسيط العملية وأتمتتها لجعلها أكثر كفاءة.

المتطلب الأساسي

يمكن استخدام وظيفة متوازية التعلم الآلي Azure فقط كخطوة واحدة في مهمة البنية الأساسية لبرنامج ربط العمليات التجارية. وبالتالي، من المهم أن تكون على دراية باستخدام البنية الأساسية لبرنامج ربط العمليات التجارية. لمعرفة المزيد حول البنية الأساسية لبرنامج ربط العمليات التجارية التعلم الآلي Azure، راجع المقالات التالية.

• فهم ما هو مسار التعلم الآلي من Azure
• فهم كيفية استخدام البنية الأساسية لبرنامج ربط العمليات التجارية Azure التعلم الآلي مع CLI v2 وSDK v2.

لماذا هناك حاجة إلى وظائف متوازية؟

في العالم الحقيقي، يكون لدى مهندسي التعلم الآلي دائما متطلبات تغيير الحجم على مهام التدريب أو الاستدلال. على سبيل المثال، عندما يوفر عالم البيانات نصا نصيا واحدا لتدريب نموذج توقع المبيعات، يحتاج مهندسو التعلم الآلي إلى تطبيق مهمة التدريب هذه على كل متجر فردي. خلال عملية التوسع هذه، بعض التحديات هي:

• تأخير الضغط الناجم عن وقت التنفيذ الطويل.
• التدخل اليدوي لمعالجة المشكلات غير المتوقعة للحفاظ على متابعة المهمة.

القيمة الأساسية ل Azure التعلم الآلي مهمة متوازية هي تقسيم مهمة تسلسلية واحدة إلى دفعات صغيرة وإرسال هذه الدفعات الصغيرة إلى حسابات متعددة للتنفيذ بالتوازي. باستخدام الوظائف المتوازية، يمكننا:

• تقليل وقت التنفيذ من طرف إلى طرف بشكل كبير.
• استخدم إعدادات معالجة الأخطاء التلقائية للوظيفة المتوازية في Azure التعلم الآلي.

يجب أن تفكر في استخدام Azure التعلم الآلي Parallel job إذا:

• تخطط لتدريب العديد من النماذج أعلى البيانات المقسمة.
• تريد تسريع مهمة الاستدلال الدفعي على نطاق واسع.

الاستعداد لوظيفة موازية

على عكس الأنواع الأخرى من الوظائف، تتطلب الوظيفة المتوازية الإعداد. اتبع الأقسام التالية للتحضير لإنشاء وظيفتك المتوازية.

الإعلان عن المدخلات التي سيتم توزيعها وإعداد تقسيم البيانات

تتطلب الوظيفة المتوازية تقسيم بيانات إدخال رئيسية واحدة فقط ومعالجتها بالتوازي. يمكن أن تكون بيانات الإدخال الرئيسية إما بيانات جدولية أو مجموعة من الملفات. يمكن أن تحتوي أنواع الإدخال المختلفة على أسلوب تقسيم بيانات مختلف.

يوضح الجدول التالي العلاقة بين بيانات الإدخال وطريقة تقسيم البيانات:

تنسيق البيانات	نوع إدخال Azure التعلم الآلي	وضع إدخال azure التعلم الآلي	أسلوب تقسيم البيانات
قائمة الملفات	mltable أو uri_folder	ro_mount أو تنزيل	حسب الحجم (عدد الملفات) حسب الأقسام
البيانات الجدولية	mltable	مباشر	حسب الحجم (الحجم الفعلي المقدر) حسب الأقسام

يمكنك الإعلان عن بيانات الإدخال الرئيسية الخاصة بك مع input_data سمة في وظيفة متوازية YAML أو Python SDK. ويمكنك ربطه بأحد معرفاتك inputs لمهمتك المتوازية باستخدام ${{inputs.<input name>}}. ثم تحتاج إلى تحديد أسلوب تقسيم البيانات لإدخالك الرئيسي عن طريق ملء سمة مختلفة:

أسلوب تقسيم البيانات	اسم السمة	نوع السمة	مثال المهمة
حسب الحجم	mini_batch_size	سلسلة	توقع دفعة قزحية القزحية
حسب الأقسام	partition_keys	قائمة السلسلة	توقع مبيعات عصير البرتقال

Azure CLI

ينطبق على: ملحق ML Azure CLI 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.5/labels/latest",
    ),
)

بمجرد تحديد إعداد تقسيم البيانات، يمكنك تكوين عدد الموارد للتوازي الخاص بك عن طريق ملء سمتين أدناه:

اسم السمة	نوع	الوصف	القيمة الافتراضية
instance_count	integer	عدد العقد التي يجب استخدامها للمهمة.	1
max_concurrency_per_instance	integer	عدد المعالجات على كل عقدة.	بالنسبة لحساب GPU، القيمة الافتراضية هي 1. بالنسبة لحساب وحدة المعالجة المركزية، القيمة الافتراضية هي عدد الذاكرات الأساسية.

تعمل هاتين السمتين معا مع مجموعة الحوسبة المحددة.

نموذج التعليمات البرمجية لتعيين سمتين:

Azure CLI

ينطبق على: ملحق ML Azure CLI 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.5/labels/latest",
    ),
)

إشعار

إذا كنت تستخدم جدولي mltable كبيانات إدخال رئيسية، فستحتاج إلى تعبئة ملف مواصفات MLTABLE مع transformations - read_delimited قسم تحت المسار المحدد. للحصول على مزيد من الأمثلة، راجع إنشاء أصل بيانات قابل للنطاق القابل للنطاق

تنفيذ دالات معرفة مسبقا في البرنامج النصي للإدخال

البرنامج النصي للإدخال هو ملف Python واحد حيث يحتاج المستخدم إلى تنفيذ ثلاث وظائف معرفة مسبقا مع تعليمات برمجية مخصصة. يتبع Azure التعلم الآلي مهمة متوازية الرسم التخطيطي أدناه لتنفيذها في كل معالج.

اسم الوظيفة	المطلوب	الوصف	إدخال	الإرجاع
Init()	Y	استخدم هذه الدالة للتحضير الشائع قبل البدء في تشغيل دفعات صغيرة. على سبيل المثال، قم باستخدامه لتحميل النموذج في كائن عام.	--	--
تشغيل (mini_batch)	Y	تنفيذ منطق التنفيذ الرئيسي mini_batches.	mini_batch: إطار بيانات Pandas إذا كانت بيانات الإدخال بيانات جدولية. قائمة مسار الملف إذا كانت بيانات الإدخال دليلا.	إطار البيانات أو القائمة أو المجموعة.
إيقاف التشغيل()	N	دالة اختيارية للقيام بتنظيف مخصص قبل إرجاع الحساب مرة أخرى إلى التجمع.	--	--

تحقق من أمثلة البرنامج النصي للإدخل التالية للحصول على مزيد من التفاصيل:

• تعريف الصورة لقائمة ملفات الصور
• تصنيف قزحية العين لبيانات قزحية العين الجدولية

بمجرد أن يكون لديك برنامج نصي للإدخل جاهز، يمكنك تعيين السمتين التاليتين لاستخدامه في وظيفتك المتوازية:

اسم السمة	نوع	الوصف	القيمة الافتراضية
code	سلسلة	المسار المحلي إلى دليل التعليمة البرمجية المصدر ليتم تحميله واستخدامه للوظيفة.
entry_script	سلسلة	ملف Python الذي يحتوي على تنفيذ وظائف متوازية محددة مسبقا.

نموذج التعليمات البرمجية لتعيين سمتين:

Azure CLI

ينطبق على: ملحق ML Azure CLI 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.5/labels/latest",
    ),
)

هام

تتطلب وظيفة التشغيل (mini_batch) إرجاع إما إطار بيانات أو قائمة أو عنصر مجموعة. ستستخدم الوظيفة المتوازية عدد تلك العودة لقياس عناصر النجاح ضمن تلك الدفعة المصغرة. من الناحية المثالية، يجب أن يكون عدد الدفعات المصغرة مساويا لعدد قائمة الإرجاع إذا تمت معالجة جميع العناصر بشكل جيد في هذه الدفعة المصغرة.

هام

إذا كنت تريد تحليل الوسيطات في الدالة Init() أو Run(mini_batch)، فاستخدم "parse_known_args" بدلا من "parse_args" لتجنب الاستثناءات. راجع المثال iris_score للبرنامج النصي للإدخل مع محلل الوسيطات.

هام

إذا كنت تستخدم mltable كبيانات إدخال رئيسية، فأنت بحاجة إلى تثبيت مكتبة "mltable" في بيئتك. راجع السطر 9 من مثال ملف conda هذا.

ضع في اعتبارك إعدادات الأتمتة

تعرض وظيفة Azure التعلم الآلي المتوازية العديد من الإعدادات للتحكم تلقائيا في الوظيفة دون تدخل يدوي. راجع الجدول التالي للحصول على التفاصيل.

المفتاح	النوع	الوصف	القيم المسموح بها	القيمة الافتراضية	تعيين في السمة	تعيين في وسيطات البرنامج
حد خطأ دفعي صغير	integer	حدد عدد الدفعات المصغرة الفاشلة التي يمكن تجاهلها في هذه المهمة المتوازية. إذا كان عدد الدفعات المصغرة الفاشلة أعلى من هذا الحد، وضع علامة على المهمة المتوازية على أنها فاشلة. يتم وضع علامة على الدفعة المصغرة على أنها فاشلة إذا: - عدد العائدات من run() أقل من عدد الإدخالات الدفعية المصغرة. - التقاط الاستثناءات في التعليمات البرمجية المخصصة run() . "-1" هو الرقم الافتراضي، ما يعني تجاهل جميع الدفعات المصغرة الفاشلة أثناء المهمة المتوازية.	[-1، int.max]	-1	mini_batch_error_threshold	‏‫غير متوفر‬
الحد الأقصى لإعادة المحاولة للدفعة المصغرة	integer	حدد عدد مرات إعادة المحاولة عند فشل الدفعة المصغرة أو انتهاء المهلة. إذا فشلت جميع عمليات إعادة المحاولة، وضع علامة على الدفعة المصغرة على أنها فاشلة في حسابها mini_batch_error_threshold .	[0، int.max]	2	retry_settings.max_retries	‏‫غير متوفر‬
مهلة دفعة صغيرة	integer	حدد المهلة بالثوان لتنفيذ دالة run() المخصصة. إذا كان وقت التنفيذ أعلى من هذا الحد، إجهاض الدفعة المصغرة، ووضع علامة عليها كدفعة مصغرة فاشلة لتشغيل إعادة المحاولة.	(0, 259200]	60	مهلة retry_settings	‏‫غير متوفر‬
حد خطأ العنصر	integer	حد العناصر الفاشلة. يتم حساب العناصر الفاشلة حسب فجوة الرقم بين المدخلات والإرجاعات من كل دفعة صغيرة. إذا كان مجموع العناصر الفاشلة أعلى من هذا الحد، وضع علامة على المهمة المتوازية على أنها فاشلة. ملاحظة: "-1" هو الرقم الافتراضي، مما يعني تجاهل جميع حالات الفشل أثناء المهمة المتوازية.	[-1، int.max]	-1	‏‫غير متوفر‬	--error_threshold
النسبة المئوية الفاشلة المسموح بها	integer	mini_batch_error_threshold على غرار ولكن يستخدم النسبة المئوية للدفعات المصغرة الفاشلة بدلا من العدد.	[0, 100]	100	‏‫غير متوفر‬	--allowed_failed_percent
مهلة النفقات العامة	integer	المهلة بالثانية لتهيئة كل دفعة صغيرة. على سبيل المثال، قم بتحميل بيانات الدفعة المصغرة وتمريرها إلى وظيفة run().	(0, 259200]	600	‏‫غير متوفر‬	--task_overhead_timeout
مهلة تحديث التقدم	integer	المهلة في الثانية لمراقبة تقدم تنفيذ الدفعة المصغرة. إذا لم يتم تلقي أي تحديثات تقدم ضمن إعداد المهلة هذا، وضع علامة على المهمة المتوازية على أنها فاشلة.	(0, 259200]	يتم حسابها ديناميكيا بواسطة إعدادات أخرى.	‏‫غير متوفر‬	--progress_update_timeout
مهلة إنشاء المهمة الأولى	integer	المهلة في الثانية لمراقبة الوقت بين بدء المهمة لتشغيل الدفعة المصغرة الأولى.	(0, 259200]	600	‏‫غير متوفر‬	--first_task_creation_timeout
مستوى التسجيل	سلسلة	حدد مستوى السجلات التي سيتم تفريغها إلى ملفات سجل المستخدم.	معلومات أو تحذير أو تصحيح الأخطاء	معلومات	logging_level	‏‫غير متوفر‬
إلحاق صف ب	سلسلة	تجميع كافة الإرجاعات من كل تشغيل للدفعة المصغرة وإخراجها في هذا الملف. قد يشير إلى أحد مخرجات المهمة المتوازية باستخدام التعبير ${{outputs.<>output_name}}			task.append_row_to	‏‫غير متوفر‬
نسخ السجلات إلى الأصل	سلسلة	خيار منطقي لمعرفة ما إذا كان يتم نسخ تقدم المهمة، ونظرة عامة، والسجلات إلى مهمة البنية الأساسية لبرنامج ربط العمليات التجارية الأصل.	صواب أو خطأ	خطأ	‏‫غير متوفر‬	--copy_logs_to_parent
الفاصل الزمني لمراقبة الموارد	integer	الفاصل الزمني بالثوان لتفريغ استخدام موارد العقدة (على سبيل المثال، وحدة المعالجة المركزية، الذاكرة) لتسجيل المجلد ضمن مسار "logs/sys/perf". ملاحظة: ستؤدي سجلات موارد التفريغ المتكررة إلى إبطاء سرعة تنفيذ الدفعة الصغيرة قليلا. قم بتعيين هذه القيمة إلى "0" لإيقاف تفريغ استخدام الموارد.	[0، int.max]	600	‏‫غير متوفر‬	--resource_monitor_interval

نموذج التعليمات البرمجية لتحديث هذه الإعدادات:

Azure CLI

ينطبق على: ملحق ML Azure CLI 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

ينطبق على: ملحق ML Azure CLI 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

Azure CLI

ينطبق على: ملحق ML Azure CLI v2 (الحالي)

يمكنك إرسال مهمة البنية الأساسية لبرنامج ربط العمليات التجارية الخاصة بك مع خطوة متوازية باستخدام الأمر CLI:

az ml job create --file pipeline.yml

Python

ينطبق على: Python SDK azure-ai-ml v2 (الحالي)

يمكنك إرسال مهمة البنية الأساسية لبرنامج ربط العمليات التجارية الخاصة بك مع خطوة متوازية باستخدام jobs.create_or_update وظيفة ml_client:

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

بمجرد إرسال مهمة البنية الأساسية لبرنامج ربط العمليات التجارية الخاصة بك، سيمنحك عنصر واجهة المستخدم SDK أو CLI رابط عنوان URL على الويب إلى واجهة مستخدم Studio. سيرشدك الرابط إلى عرض الرسم البياني لالتدفق بشكل افتراضي. حدد الخطوة المتوازية مرتين لفتح اللوحة اليمنى من وظيفتك المتوازية.

للتحقق من إعدادات المهمة المتوازية، انتقل إلى علامة التبويب Parameters ، وقم بتوسيع Run settings، وتحقق من قسم Parallel :

لتصحيح فشل المهمة المتوازية، انتقل إلى علامة التبويب Outputs + Logs ، وقم بتوسيع مجلد السجلات من دلائل الإخراج على اليسار، وتحقق من job_result.txt لفهم سبب فشل المهمة المتوازية. لمزيد من التفاصيل حول بنية تسجيل المهمة المتوازية ، راجع readme.txt ضمن نفس المجلد.

مهمة متوازية في أمثلة البنية الأساسية لبرنامج ربط العمليات التجارية

• مستودع مثال Azure CLI + YAML
• مستودع مثال SDK

الخطوات التالية

• للحصول على مخطط yaml التفصيلي للمهمة المتوازية ، راجع مرجع YAML للمهمة المتوازية.
• للحصول على كيفية إلحاق بياناتك في MLTABLE، راجع إنشاء أصل بيانات قابل للترجمة.
• لمعرفة كيفية تشغيل البنية الأساسية لبرنامج ربط العمليات التجارية بانتظام، راجع كيفية جدولة البنية الأساسية لبرنامج ربط العمليات التجارية.