مقدمة عن مسارات Vertex

1. نظرة عامة

في هذا الدرس التطبيقي، ستتعرّف على كيفية إنشاء مسارات تعلُّم الآلة وتشغيلها باستخدام مسارات Vertex.

المعلومات التي تطّلع عليها

ستتعرّف على كيفية:

  • استخدام حِزم تطوير البرامج (SDK) لمسار Kubeflow Pipelines لإنشاء مسارات تعلُّم آلي قابلة للتطوير
  • إنشاء مسار تقديمي من 3 خطوات يتطلّب إدخال النص وتشغيله
  • إنشاء وتشغيل عملية معالجة تُدرِّب نموذج تصنيف "التعلم الآلي التلقائي" وتُقيّمه وتنشره
  • استخدام مكوّنات مُنشأة مسبقًا للتفاعل مع خدمات Vertex AI، والتي يتم توفيرها من خلال مكتبة google_cloud_pipeline_components
  • جدولة مهمة مسار بيانات باستخدام Cloud Scheduler

تبلغ التكلفة الإجمالية لتنفيذ هذا البرنامج التدريبي على Google Cloud حوالي 25 دولار أمريكي.

2. مقدّمة عن Vertex AI

يستخدم هذا المختبر أحدث منتجات الذكاء الاصطناعي المتاحة على Google Cloud. تدمج Vertex AI حلول تعلُّم الآلة في Google Cloud ضمن تجربة تطوير سلسة. في السابق، كان بالإمكان الوصول إلى النماذج المدربة باستخدام AutoML والنماذج المخصّصة من خلال خدمات منفصلة. ويجمع العرض الجديد بين الاثنين في واجهة برمجة تطبيقات واحدة، إلى جانب منتجات جديدة أخرى. يمكنك أيضًا نقل المشاريع الحالية إلى Vertex AI.

بالإضافة إلى خدمات تدريب النماذج ونشرها، تتضمّن Vertex AI أيضًا مجموعة متنوعة من منتجات MLOps، بما في ذلك Vertex Pipelines (موضوع هذا المختبر) وModel Monitoring وFeature Store والمزيد. يمكنك الاطّلاع على جميع عروض منتجات Vertex AI في المخطّط البياني أدناه.

نظرة عامة على منتجات Vertex

إذا كان لديك أي ملاحظات، يُرجى الاطّلاع على صفحة الدعم.

ما هي فائدة قنوات تعلُّم الآلة؟

قبل البدء، لنلقِ نظرة على أسباب استخدام مسار الإحالة الناجحة. لنفترض أنّك بصدد إنشاء سير عمل تعلُّم آلي يتضمّن معالجة البيانات وتدريب نموذج وضبط المَعلمات الفائقة وتقييم النموذج ونشره. قد يكون لكل خطوة من هذه الخطوات تبعيات مختلفة، والتي قد تصبح غير عملية إذا تعاملت مع سير العمل بأكمله على أنه كتلة واحدة. عند بدء توسيع نطاق عملية تعلُّم الآلة، قد تحتاج إلى مشاركة سير عمل تعلُّم الآلة مع الآخرين في فريقك حتى يتمكّنوا من تشغيله وتقديم مساهمات في الرموز البرمجية. ويمكن أن يصبح ذلك صعبًا بدون عملية موثوقة يمكن تكرارها. باستخدام المسارات، تكون كل خطوة في عملية تعلُّم الآلة هي حاويتها الخاصة. يتيح لك ذلك تطوير الخطوات بشكل مستقل وتتبُّع الإدخالات والمخرجات من كل خطوة بطريقة يمكن إعادة إنتاجها. يمكنك أيضًا جدولة عمليات تنفيذ مسار الإحالة الناجحة أو تشغيلها استنادًا إلى أحداث أخرى في بيئة Cloud، مثل بدء عملية تنفيذ مسار الإحالة الناجحة عند توفّر بيانات تدريب جديدة.

النص المختصَر: تساعدك المسارات في تنفيذ المهام آليًا وإعادة إنتاج سير عمل تعلُّم الآلة.

3- إعداد بيئة السحابة الإلكترونية

ستحتاج إلى مشروع Google Cloud Platform مع تفعيل الفوترة لتشغيل هذا الدرس التطبيقي حول الترميز. لإنشاء مشروع، يُرجى اتّباع التعليمات هنا.

الخطوة 1: بدء Cloud Shell

ستعمل في هذا التمرين المعملي في جلسة Cloud Shell، وهي عبارة عن مترجم أوامر يستضيفه جهاز افتراضي يعمل على السحابة الإلكترونية من Google. يمكنك تنفيذ هذا القسم على جهاز الكمبيوتر الخاص بك بسهولة، ولكن باستخدام Cloud Shell، يمكن للجميع الوصول إلى تجربة قابلة للتكرار في بيئة متّسقة. بعد انتهاء التمرين المعملي، يمكنك إعادة محاولة استخدام هذا القسم على جهاز الكمبيوتر.

منح الإذن في Cloud Shell

تفعيل Cloud Shell

في أعلى يسار Cloud Console، انقر على الزر أدناه لتفعيل Cloud Shell:

تفعيل Cloud Shell

إذا لم يسبق لك تشغيل Cloud Shell، ستظهر لك شاشة وسيطة (أسفل الصفحة) توضّح ماهية هذه الخدمة. في هذه الحالة، انقر على متابعة (ولن تظهر لك هذه الشاشة مرة أخرى). في ما يلي شكل الشاشة التي تظهر لمرة واحدة:

إعداد Cloud Shell

من المفترَض أن تستغرق عملية إدارة الحسابات والاتصال بخدمة Cloud Shell بضع دقائق فقط.

إعداد Cloud Shell

تم تحميل هذا الجهاز الظاهري بجميع أدوات التطوير التي تحتاج إليها. ويقدّم هذا الدليل دليلاً منزليًا دائمًا بسعة 5 غيغابايت ويتم تشغيله في Google Cloud، ما يُحسِّن بشكل كبير أداء الشبكة والمصادقة. يمكن تنفيذ الكثير من عملك في هذا الدليل التعليمي حول رموز البرامج، إن لم يكن كلّه، باستخدام متصفّح أو جهاز Chromebook فقط.

بعد الربط بخدمة Cloud Shell، من المفترَض أن يظهر لك أنّه سبق أن تمت مصادقتك وأنّ المشروع قد تم ضبطه على رقم تعريف مشروعك.

شغِّل الأمر التالي في Cloud Shell لتأكيد مصادقتك:

gcloud auth list

من المفترض أن يظهر لك شيء مثل هذا في إخراج الأمر:

ناتج Cloud Shell

شغِّل الأمر التالي في Cloud Shell للتأكّد من أنّ الأمر gcloud يعرف مشروعك:

gcloud config list project

ناتج الأمر

[core]
project = <PROJECT_ID>

إذا لم يكن كذلك، يمكنك تعيينه من خلال هذا الأمر:

gcloud config set project <PROJECT_ID>

ناتج الأمر

Updated property [core/project].

تحتوي Cloud Shell على بعض متغيّرات البيئة، بما في ذلك GOOGLE_CLOUD_PROJECT التي تحتوي على اسم مشروعنا الحالي على Cloud. سنستخدم هذا في أماكن مختلفة خلال هذا التمرين المعملي. يمكنك الاطّلاع عليها من خلال تنفيذ:

echo $GOOGLE_CLOUD_PROJECT

الخطوة 2: تفعيل واجهات برمجة التطبيقات

في الخطوات اللاحقة، ستتعرّف على الأماكن التي تكون فيها هذه الخدمات مطلوبة (والأسباب التي تدفعك إلى ذلك)، ولكن في الوقت الحالي، يمكنك تنفيذ هذا الأمر لمنح مشروعك إذن الوصول إلى خدمات Compute Engine وContainer Registry وVertex AI:

gcloud services enable compute.googleapis.com         \
                       containerregistry.googleapis.com  \
                       aiplatform.googleapis.com  \
                       cloudbuild.googleapis.com \
                       cloudfunctions.googleapis.com

من المفترض أن تؤدي هذه الخطوة إلى ظهور رسالة تفيد بنجاح العملية على النحو التالي:

Operation "operations/acf.cc11852d-40af-47ad-9d59-477a12847c9e" finished successfully.

الخطوة 3: إنشاء حزمة في Cloud Storage

لتشغيل وظيفة تدريب على Vertex AI، سنحتاج إلى حزمة تخزين لتخزين مواد عرض النماذج المحفوظة. يجب أن تكون الحزمة إقليمية. سنستخدم us-central هنا، ولكن يمكنك استخدام منطقة أخرى (ما عليك سوى استبدالها في جميع أنحاء هذا البرنامج التعليمي). إذا كان لديك حزمة حاليًا، يمكنك تخطّي هذه الخطوة.

شغِّل الأوامر التالية في الوحدة الطرفية في Cloud Shell لإنشاء حزمة:

BUCKET_NAME=gs://$GOOGLE_CLOUD_PROJECT-bucket
gsutil mb -l us-central1 $BUCKET_NAME

بعد ذلك، سنمنح حساب خدمة الحوسبة إذن الوصول إلى هذه الحزمة. سيضمن ذلك أنّ Vertex Pipelines لديها الأذونات اللازمة لكتابة الملفات في حزمة التخزين هذه. شغِّل الأمر التالي لإضافة هذا الإذن:

gcloud projects describe $GOOGLE_CLOUD_PROJECT > project-info.txt
PROJECT_NUM=$(cat project-info.txt | sed -nre 's:.*projectNumber\: (.*):\1:p')
SVC_ACCOUNT="${PROJECT_NUM//\'/}-compute@developer.gserviceaccount.com"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT --member serviceAccount:$SVC_ACCOUNT --role roles/storage.objectAdmin

الخطوة 4: إنشاء مثيل Vertex AI Workbench

من قسم Vertex AI في Cloud Console، انقر على Workbench:

قائمة Vertex AI

من هناك، ضمن دفاتر الملاحظات التي يديرها المستخدم، انقر على دفتر ملاحظات جديد:

إنشاء دفتر ملاحظات جديد

بعد ذلك، اختَر نوع الإصدار TensorFlow Enterprise 2.3 (مع LTS) بدون وحدات معالجة الرسومات:

مثيل TFE

استخدِم الخيارات التلقائية، ثم انقر على إنشاء.

الخطوة 5: فتح دفتر ملاحظاتك

بعد إنشاء المثيل، اختَر فتح JupyterLab:

فتح دفتر ملاحظات

4. إعداد خطوط أنابيب Vertex

هناك بعض المكتبات الإضافية التي سنحتاج إلى تثبيتها لاستخدام Vertex Pipelines:

  • Kubeflow Pipelines: هذه هي حزمة SDK التي سنستخدمها لإنشاء مسار الإحالة الناجحة. تتيح Vertex Pipelines تشغيل مسارات بيانات تم إنشاؤها باستخدام كلّ من Kubeflow Pipelines أو TFX.
  • مكوّنات Google Cloud Pipeline: توفّر هذه المكتبة مكوّنات مُعدّة مسبقًا تسهّل التفاعل مع خدمات Vertex AI من خطوات مسار الإحالة الناجحة.

الخطوة 1: إنشاء دفتر ملاحظات Python وتثبيت المكتبات

أولاً، من قائمة "مشغّل التطبيقات" في مثيل "دفتر ملاحظات Google"، أنشئ دفتر ملاحظات من خلال اختيار Python 3:

إنشاء دفتر ملاحظات Python3

يمكنك الوصول إلى قائمة "مشغّل التطبيقات" بالنقر على علامة + في أعلى يمين مثيل دفتر البيانات.

لتثبيت كلتا الخدمتَين اللتين سنستخدمهما في هذا الدرس التطبيقي، عليك أولاً ضبط علامة المستخدم في خلية دفتر ملاحظات:

USER_FLAG = "--user"

بعد ذلك، نفِّذ ما يلي من دفتر ملاحظاتك:

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.7.0 --upgrade
!pip3 install {USER_FLAG} kfp==1.8.9 google-cloud-pipeline-components==0.2.0

بعد تثبيت هذه الحِزم، عليك إعادة تشغيل النواة:

import os

if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython

    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)

أخيرًا، تأكَّد من تثبيت الحِزم بشكلٍ صحيح. يجب أن يكون إصدار حزمة تطوير البرامج (SDK) لـ KFP أكبر من أو يساوي 1.8:

!python3 -c "import kfp; print('KFP SDK version: {}'.format(kfp.__version__))"
!python3 -c "import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))"

الخطوة 2: ضبط رقم تعريف المشروع والحزمة

خلال هذا التمرين، ستشير إلى رقم تعريف مشروعك على Cloud والحزمة التي أنشأتها سابقًا. بعد ذلك، سننشئ متغيّرات لكلّ منهما.

إذا كنت لا تعرف رقم تعريف مشروعك، قد تتمكّن من الحصول عليه من خلال تنفيذ ما يلي:

import os
PROJECT_ID = ""

# Get your Google Cloud project ID from gcloud
if not os.getenv("IS_TESTING"):
    shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID: ", PROJECT_ID)

بخلاف ذلك، يمكنك ضبطها هنا:

if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "your-project-id"  # @param {type:"string"}

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

BUCKET_NAME="gs://" + PROJECT_ID + "-bucket"

الخطوة 3: استيراد المكتبات

أضِف ما يلي لاستيراد المكتبات التي سنستخدمها في هذا الدليل التعليمي للترميز:

import kfp

from kfp.v2 import compiler, dsl
from kfp.v2.dsl import component, pipeline, Artifact, ClassificationMetrics, Input, Output, Model, Metrics

from google.cloud import aiplatform
from google_cloud_pipeline_components import aiplatform as gcc_aip
from typing import NamedTuple

الخطوة 4: تعريف الثوابت

آخر إجراء علينا اتّخاذه قبل إنشاء مسار الإحالة الناجحة هو تحديد بعض المتغيّرات الثابتة. PIPELINE_ROOT هو مسار Cloud Storage حيث تتم كتابة العناصر التي أنشأها المسار. سنستخدم us-central1 كمنطقة هنا، ولكن إذا استخدمت منطقة مختلفة عند إنشاء الحزمة، عدِّل المتغيّر REGION في الرمز أدناه:

PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin
REGION="us-central1"

PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/"
PIPELINE_ROOT

بعد تشغيل التعليمة البرمجية أعلاه، ستتم طباعة الدليل الجذر لمسار العملية. هذا هو الموقع على Cloud Storage الذي سيتم فيه كتابة العناصر من مسار الإحالة الناجحة. وسيكون بالتنسيق gs://YOUR-BUCKET-NAME/pipeline_root/.

5- إنشاء مسار الإحالة الناجحة الأول

للتعرّف على آلية عمل Vertex Pipelines، سننشئ أولاً مسارًا قصيرًا باستخدام حزمة KFP SDK. لا تُجري هذه السلسلة أيّ عمليات ذات صلة بالذكاء الاصطناعي (لا داعي للقلق، سنصل إلى ذلك)، بل نستخدمها لتعليمك ما يلي:

  • كيفية إنشاء مكوّنات مخصّصة في حزمة تطوير البرامج (SDK) لبرنامج "حملات الفيديو الإجرائية"
  • كيفية تشغيل مسار بيانات ومراقبته في Vertex Pipelines

سننشئ سلسلة معالجة تطبع جملة باستخدام مخرجَين: اسم المنتج ووصف الرموز التعبيرية. سيتألف مسار الإحالة الناجحة هذا من ثلاثة مكوّنات:

  • product_name: سيأخذ هذا المكوّن اسم منتج (أو أيّ اسم تريده) كمدخل، وسيعرض هذه السلسلة كمخرج.
  • emoji: سيأخذ هذا المكوّن الوصف النصي لرمز تعبيري ويحوّله إلى رمز تعبيري. على سبيل المثال، الرمز النصي لرمز ✨ هو "sparkles". يستخدم هذا المكوّن مكتبة رموز تعبيرية لعرض كيفية إدارة العناصر الخارجية التي تعتمد عليها في مسار الإحالة الناجحة.
  • build_sentence: سيستهلك هذا المكوِّن الأخير ناتج العنصرين السابقين لإنشاء جملة تستخدم الرمز التعبيري. على سبيل المثال، قد يكون الناتج الناتج هو "مسارات Vertex Pipelines ✨".

لنبدأ بكتابة الرموز البرمجية.

الخطوة 1: إنشاء مكوّن يستند إلى دالة Python

باستخدام حزمة تطوير البرامج (SDK) لـ KFP، يمكننا إنشاء مكوّنات استنادًا إلى دوالّ Python. سنستخدم ذلك للمكونات الثلاثة في مسار العملية الأول. سننشئ أولاً المكوّن product_name الذي يأخذ سلسلة كإدخال ويعرض هذه السلسلة. أضف ما يلي إلى دفتر ملاحظاتك:

@component(base_image="python:3.9", output_component_file="first-component.yaml")
def product_name(text: str) -> str:
    return text

لنلقِ نظرة عن كثب على بنية الجملة هنا:

  • ويقوم المصمم الزخرفي @component بتجميع هذه الدالة إلى أحد المكونات عند تشغيل المسار. ستستخدم هذا في أي وقت تكتب مكونًا مخصصًا.
  • تحدِّد المَعلمة base_image صورة الحاوية التي سيستخدمها هذا المكوّن.
  • المعلمة output_component_file اختيارية، وتحدد ملف yaml لكتابة المكون الذي تم تجميعه. بعد تشغيل الخلية، من المفترض أن يظهر لك هذا الملف مكتوبًا في مثيل دفتر البيانات. إذا كنت ترغب في مشاركة هذا المكون مع شخص ما، فيمكنك إرسال ملف yaml الذي تم إنشاؤه إليه وتحميله بما يلي:
product_name_component = kfp.components.load_component_from_file('./first-component.yaml')
  • يحدّد -> str بعد تعريف الدالة نوع الإخراج لهذا المكوِّن.

الخطوة 2: إنشاء مكوّنَين إضافيَّين

لإكمال مسار الإحالة الناجحة، سننشئ مكوّنَين إضافيَين. يأخذ الإجراء الأول الذي سنحدّده سلسلة كإدخال، ويحوّل هذه السلسلة إلى الرموز التعبيرية المقابلة لها إذا توفّرت. يعرض صفًا يحتوي على نص الإدخال الذي تم إدخاله، والرموز التعبيرية الناتجة:

@component(packages_to_install=["emoji"])
def emoji(
    text: str,
) -> NamedTuple(
    "Outputs",
    [
        ("emoji_text", str),  # Return parameters
        ("emoji", str),
    ],
):
    import emoji

    emoji_text = text
    emoji_str = emoji.emojize(':' + emoji_text + ':', language='alias')
    print("output one: {}; output_two: {}".format(emoji_text, emoji_str))
    return (emoji_text, emoji_str)

هذا المكوّن أكثر تعقيدًا بعض الشيء من المكوّن السابق. لنلقِ نظرة على الميزات الجديدة:

  • تُعلم المَعلمة packages_to_install المكوّن بأي تبعيات مكتبة خارجية لهذه الحاوية. ونحن في هذه الحالة نستخدم مكتبة اسمها الرموز التعبيرية.
  • يعرض هذا المكوّن NamedTuple يُسمى Outputs. لاحظ أنّ كل سلسلة في هذه المجموعة تحتوي على مفتاحَين: emoji_text وemoji. سنستخدم هذه في المكون التالي للوصول إلى المخرجات.

سيستهلك المكوّن الأخير في هذه السلسلة من الإجراءات ناتج المكوّنين الأولَين ويدمجه لعرض سلسلة:

@component
def build_sentence(
    product: str,
    emoji: str,
    emojitext: str
) -> str:
    print("We completed the pipeline, hooray!")
    end_str = product + " is "
    if len(emoji) > 0:
        end_str += emoji
    else:
        end_str += emojitext
    return(end_str)

قد تتساءل: كيف يعرف هذا المكوّن استخدام الإخراج من الخطوات السابقة التي حدّدتها؟ سؤال جيد. سنربط كل هذه العناصر ببعضها في الخطوة التالية.

الخطوة 3: تجميع المكوّنات معًا في مسار عمل

أدّت تعريفات المكوّنات التي حدّدناها أعلاه إلى إنشاء دوال مصنع يمكن استخدامها في تعريف مسار لإنشاء الخطوات. لإعداد مسار عمل، استخدِم زخرفة @pipeline، وامنح مسار العمل اسمًا ووصفًا، وقدِّم المسار الجذر الذي يجب كتابة عناصر مسار العمل فيه. نقصد بالعناصر أي ملفات ناتجة أنشأتها مسارك. لا تُنشئ مسار الإحالة الناجحة هذا أيّ تقارير، ولكن سيُنشئ مسار الإحالة الناجحة التالي تقارير.

في مجموعة التعليمات البرمجية التالية، نحدِّد دالة intro_pipeline. في هذا القسم، نحدّد المدخلات لخطوات مسار الإحالة الناجحة الأوليّة، وكيفية ربط الخطوات ببعضها:

  • يأخذ product_task اسم منتج كمدخل. نحن هنا نمرر "Vertex Pipelines" ولكن يمكنك تغيير ذلك إلى أي شيء تريده.
  • تأخذ emoji_task الرمز النصي للرمز التعبيري كإدخال. يمكنك أيضًا تغيير هذا إلى أي شيء تريده. على سبيل المثال، يشير "party_face" إلى الرمز التعبيري 🥳. يُرجى العِلم أنّه بما أنّ هذا المكوّن ومكوّن product_task لا يتضمّنان أي خطوات تُغذّيهما بالمدخلات، فإنّنا نحدّد يدويًا المدخلات لكلاهما عند تحديد مسار الإحالة الناجحة.
  • تحتوي الخطوة الأخيرة في مسار الإحالة الناجحة، وهي consumer_task، على ثلاث مَعلمات إدخال:
    • نتيجة product_task. بما أنّ هذه الخطوة تنتج مخرجًا واحدًا فقط، يمكننا الإشارة إليه من خلال product_task.output.
    • نتيجة emoji لخطوتنا emoji_task. راجِع المكوّن emoji المحدّد أعلاه حيث حدّدنا مَعلمات الإخراج.
    • وبالمثل، فإنّ emoji_text هو اسم الإخراج من المكوّن emoji. وفي حال تمرير المسار بنص لا يتوافق مع رمز تعبيري، سيتم استخدام هذا النص لإنشاء جملة.
@pipeline(
    name="hello-world",
    description="An intro pipeline",
    pipeline_root=PIPELINE_ROOT,
)

# You can change the `text` and `emoji_str` parameters here to update the pipeline output
def intro_pipeline(text: str = "Vertex Pipelines", emoji_str: str = "sparkles"):
    product_task = product_name(text)
    emoji_task = emoji(emoji_str)
    consumer_task = build_sentence(
        product_task.output,
        emoji_task.outputs["emoji"],
        emoji_task.outputs["emoji_text"],
    )

الخطوة 4: تجميع مسار الإحالة الناجحة وتشغيله

بعد تحديد مسار الإحالة الناجحة، تكون مستعدًا لتجميعه. سينشئ ما يلي ملف JSON ستستخدمه لتشغيل المسار:

compiler.Compiler().compile(
    pipeline_func=intro_pipeline, package_path="intro_pipeline_job.json"
)

بعد ذلك، أنشئ متغيّر TIMESTAMP. سنستخدم هذا المعرّف في رقم تعريف الوظيفة:

from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")

بعد ذلك، حدِّد وظيفة مسار الإحالة الناجحة:

job = aiplatform.PipelineJob(
    display_name="hello-world-pipeline",
    template_path="intro_pipeline_job.json",
    job_id="hello-world-pipeline-{0}".format(TIMESTAMP),
    enable_caching=True
)

أخيرًا، يمكنك تنفيذ المهمة لإنشاء عملية تنفيذ جديدة للمسار:

job.submit()

بعد تنفيذ هذه الخلية، من المفترض أن تظهر لك السجلات مع رابط لعرض عملية تنفيذ مسار الإحالة الناجحة في وحدة التحكّم:

سجلّات مهام مسار الإجراءات

انتقِل إلى هذا الرابط. من المفترض أن يظهر مسار الإحالة الناجحة على النحو التالي عند اكتماله:

اكتمال مسار التقديم

سيستغرق تنفيذ مسار الإحالة الناجحة هذا من 5 إلى 6 دقائق. عند اكتمال العملية، يمكنك النقر على المكوّن build-sentence للاطّلاع على النتيجة النهائية:

مخرجات مسار المقدّمة

بعد أن أصبحت على دراية بطريقة عمل حزمة KFP SDK وVertex Pipelines، أصبحت جاهزًا لإنشاء نموذج تعلُّم الآلة ونشره باستخدام خدمات Vertex AI الأخرى. لنطّلِع على التفاصيل.

6- إنشاء مسار تعلُّم آلة شامل

حان الوقت لإنشاء مسار عملك الأول للتعلم الآلي. في هذه العملية، سنستخدم مجموعة بيانات الفول الجاف من UCI Machine Learning، من: KOKLU، M. وOZKAN، I.A.، (2020)، "تصنيف متعدد الفئات للفاصوليا الجافة باستخدام تقنيات الرؤية الحاسوبية والتعلم الآلي"، في Computers and Electronics in Agriculture، 174، 105507. DOI:

هذه مجموعة بيانات جدولية، وسنستخدمها في مسار المعالجة لتدريب نموذج AutoML وتقييمه ونشره، وهو نموذج يصنف الفول إلى أحد الأنواع السبعة استنادًا إلى خصائصه.

ستعمل هذه السلسلة على ما يلي:

  • أنشئ مجموعة بيانات في .
  • تدريب نموذج تصنيف جدولي باستخدام AutoML
  • الحصول على مقاييس التقييم لهذا النموذج
  • استنادًا إلى مقاييس التقييم، عليك تحديد ما إذا كنت تريد نشر النموذج باستخدام المنطق الشَرطي في Vertex Pipelines.
  • نشر النموذج على نقطة نهاية باستخدام Vertex Prediction

ستكون كل خطوة من الخطوات الموضحة مكونًا. ستستخدم معظم خطوات مسار الإحالة الناجحة مكوّنات مُنشأة مسبقًا لخدمات Vertex AI من خلال مكتبة google_cloud_pipeline_components التي استوردناها سابقًا في هذا الدليل التعليمي. في هذا القسم، سنحدّد مكوّنًا مخصّصًا واحدًا أولاً. بعد ذلك، سنحدّد بقية خطوات مسار الإحالة الناجحة باستخدام مكوّنات تم إنشاؤها مسبقًا. تسهّل المكوّنات المُنشأة مسبقًا الوصول إلى خدمات Vertex AI، مثل تدريب النماذج ونشرها.

الخطوة 1: مكوّن مخصّص لتقييم النموذج

سيتم استخدام المكوّن المخصّص الذي سنحدّده في نهاية مسار الإحالة الناجحة بعد اكتمال تدريب النموذج. سينفّذ هذا المكوّن بعض الإجراءات:

  • الحصول على مقاييس التقييم من نموذج التصنيف المدَّرب في AutoML
  • تحليل المقاييس وعرضها في واجهة مستخدم Vertex Pipelines
  • مقارنة المقاييس بحدّ أدنى لتحديد ما إذا كان يجب نشر النموذج

قبل تحديد المكوّن، لنفهم مَعلمات الإدخال والإخراج. تتلقّى هذه السلسلة من الإجراءات بعض البيانات الوصفية عن مشروعنا على Cloud، والنموذج المدّرب الناتج (سنحدّد هذا المكوّن لاحقًا)، ومقاييس تقييم النموذج، وthresholds_dict_str. سنحدد السمة thresholds_dict_str عند تشغيل مسار العملية. وفي حالة نموذج التصنيف هذا، ستكون هذه هي المنطقة الواقعة تحت قيمة منحنى خاصية تشغيل جهاز الاستقبال الذي ينبغي لنا نشر النموذج لها. على سبيل المثال، إذا أدخلنا القيمة 0.95، يعني ذلك أنّنا نريد أن تنشئ مسار الإحالة الناجحة النموذج فقط إذا كان هذا المقياس أعلى من %95.

يعرض مكوّن التقييم سلسلة تشير إلى ما إذا كان يجب نشر النموذج أم لا. أضِف ما يلي في خلية في ورقة الملاحظات لإنشاء هذا المكوّن المخصّص:

@component(
    base_image="gcr.io/deeplearning-platform-release/tf2-cpu.2-3:latest",
    output_component_file="tabular_eval_component.yaml",
    packages_to_install=["google-cloud-aiplatform"],
)
def classification_model_eval_metrics(
    project: str,
    location: str,  # "us-central1",
    api_endpoint: str,  # "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str,
    model: Input[Artifact],
    metrics: Output[Metrics],
    metricsc: Output[ClassificationMetrics],
) -> NamedTuple("Outputs", [("dep_decision", str)]):  # Return parameter.

    import json
    import logging

    from google.cloud import aiplatform as aip

    # Fetch model eval info
    def get_eval_info(client, model_name):
        from google.protobuf.json_format import MessageToDict

        response = client.list_model_evaluations(parent=model_name)
        metrics_list = []
        metrics_string_list = []
        for evaluation in response:
            print("model_evaluation")
            print(" name:", evaluation.name)
            print(" metrics_schema_uri:", evaluation.metrics_schema_uri)
            metrics = MessageToDict(evaluation._pb.metrics)
            for metric in metrics.keys():
                logging.info("metric: %s, value: %s", metric, metrics[metric])
            metrics_str = json.dumps(metrics)
            metrics_list.append(metrics)
            metrics_string_list.append(metrics_str)

        return (
            evaluation.name,
            metrics_list,
            metrics_string_list,
        )

    # Use the given metrics threshold(s) to determine whether the model is
    # accurate enough to deploy.
    def classification_thresholds_check(metrics_dict, thresholds_dict):
        for k, v in thresholds_dict.items():
            logging.info("k {}, v {}".format(k, v))
            if k in ["auRoc", "auPrc"]:  # higher is better
                if metrics_dict[k] < v:  # if under threshold, don't deploy
                    logging.info("{} < {}; returning False".format(metrics_dict[k], v))
                    return False
        logging.info("threshold checks passed.")
        return True

    def log_metrics(metrics_list, metricsc):
        test_confusion_matrix = metrics_list[0]["confusionMatrix"]
        logging.info("rows: %s", test_confusion_matrix["rows"])

        # log the ROC curve
        fpr = []
        tpr = []
        thresholds = []
        for item in metrics_list[0]["confidenceMetrics"]:
            fpr.append(item.get("falsePositiveRate", 0.0))
            tpr.append(item.get("recall", 0.0))
            thresholds.append(item.get("confidenceThreshold", 0.0))
        print(f"fpr: {fpr}")
        print(f"tpr: {tpr}")
        print(f"thresholds: {thresholds}")
        metricsc.log_roc_curve(fpr, tpr, thresholds)

        # log the confusion matrix
        annotations = []
        for item in test_confusion_matrix["annotationSpecs"]:
            annotations.append(item["displayName"])
        logging.info("confusion matrix annotations: %s", annotations)
        metricsc.log_confusion_matrix(
            annotations,
            test_confusion_matrix["rows"],
        )

        # log textual metrics info as well
        for metric in metrics_list[0].keys():
            if metric != "confidenceMetrics":
                val_string = json.dumps(metrics_list[0][metric])
                metrics.log_metric(metric, val_string)
        # metrics.metadata["model_type"] = "AutoML Tabular classification"

    logging.getLogger().setLevel(logging.INFO)
    aip.init(project=project)
    # extract the model resource name from the input Model Artifact
    model_resource_path = model.metadata["resourceName"]
    logging.info("model path: %s", model_resource_path)

    client_options = {"api_endpoint": api_endpoint}
    # Initialize client that will be used to create and send requests.
    client = aip.gapic.ModelServiceClient(client_options=client_options)
    eval_name, metrics_list, metrics_str_list = get_eval_info(
        client, model_resource_path
    )
    logging.info("got evaluation name: %s", eval_name)
    logging.info("got metrics list: %s", metrics_list)
    log_metrics(metrics_list, metricsc)

    thresholds_dict = json.loads(thresholds_dict_str)
    deploy = classification_thresholds_check(metrics_list[0], thresholds_dict)
    if deploy:
        dep_decision = "true"
    else:
        dep_decision = "false"
    logging.info("deployment decision is %s", dep_decision)

    return (dep_decision,)

الخطوة 2: إضافة مكوّنات Google Cloud المُنشأة مسبقًا

في هذه الخطوة، سنحدّد بقية مكوّنات مسار الإحالة الناجحة ونرى كيف تتوافق مع بعضها. أولاً، حدِّد الاسم المعروض لتشغيل مسار الإحالة الناجحة باستخدام طابع زمني:

import time
DISPLAY_NAME = 'automl-beans{}'.format(str(int(time.time())))
print(DISPLAY_NAME)

بعد ذلك، انسخ ما يلي في خلية دفتر ملاحظات جديد:

@pipeline(name="automl-tab-beans-training-v2",
                  pipeline_root=PIPELINE_ROOT)
def pipeline(
    bq_source: str = "bq://aju-dev-demos.beans.beans1",
    display_name: str = DISPLAY_NAME,
    project: str = PROJECT_ID,
    gcp_region: str = "us-central1",
    api_endpoint: str = "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str = '{"auRoc": 0.95}',
):
    dataset_create_op = gcc_aip.TabularDatasetCreateOp(
        project=project, display_name=display_name, bq_source=bq_source
    )

    training_op = gcc_aip.AutoMLTabularTrainingJobRunOp(
        project=project,
        display_name=display_name,
        optimization_prediction_type="classification",
        budget_milli_node_hours=1000,
        column_transformations=[
            {"numeric": {"column_name": "Area"}},
            {"numeric": {"column_name": "Perimeter"}},
            {"numeric": {"column_name": "MajorAxisLength"}},
            {"numeric": {"column_name": "MinorAxisLength"}},
            {"numeric": {"column_name": "AspectRation"}},
            {"numeric": {"column_name": "Eccentricity"}},
            {"numeric": {"column_name": "ConvexArea"}},
            {"numeric": {"column_name": "EquivDiameter"}},
            {"numeric": {"column_name": "Extent"}},
            {"numeric": {"column_name": "Solidity"}},
            {"numeric": {"column_name": "roundness"}},
            {"numeric": {"column_name": "Compactness"}},
            {"numeric": {"column_name": "ShapeFactor1"}},
            {"numeric": {"column_name": "ShapeFactor2"}},
            {"numeric": {"column_name": "ShapeFactor3"}},
            {"numeric": {"column_name": "ShapeFactor4"}},
            {"categorical": {"column_name": "Class"}},
        ],
        dataset=dataset_create_op.outputs["dataset"],
        target_column="Class",
    )
    model_eval_task = classification_model_eval_metrics(
        project,
        gcp_region,
        api_endpoint,
        thresholds_dict_str,
        training_op.outputs["model"],
    )

    with dsl.Condition(
        model_eval_task.outputs["dep_decision"] == "true",
        name="deploy_decision",
    ):

        endpoint_op = gcc_aip.EndpointCreateOp(
            project=project,
            location=gcp_region,
            display_name="train-automl-beans",
        )

        gcc_aip.ModelDeployOp(
            model=training_op.outputs["model"],
            endpoint=endpoint_op.outputs["endpoint"],
            dedicated_resources_min_replica_count=1,
            dedicated_resources_max_replica_count=1,
            dedicated_resources_machine_type="n1-standard-4",
        )

لنطّلِع على ما يحدث في هذا الرمز البرمجي:

  • أولاً، كما هو الحال في عملية النقل السابقة، نحدّد مَعلمات الإدخال التي تأخذها عملية النقل هذه. علينا ضبط هذه الإعدادات يدويًا لأنّها لا تعتمد على نتائج الخطوات الأخرى في مسار الإحالة الناجحة.
  • تستخدِم بقية مراحل مسار الإحالة الناجحة بعض المكوّنات المُنشأة مسبقًا للتفاعل مع خدمات Vertex AI:
    • TabularDatasetCreateOp لإنشاء مجموعة بيانات جدولية في Vertex AI استنادًا إلى مصدر مجموعة بيانات في Cloud Storage أو BigQuery في هذه السلسلة، يتم تمرير البيانات عبر عنوان URL لجدول BigQuery.
    • AutoMLTabularTrainingJobRunOp يبدأ مهمة تدريب AutoML لمجموعة بيانات جدولية. نمرر بعض معلمات التكوين إلى هذا المكون، بما في ذلك نوع النموذج (في هذه الحالة، التصنيف)، وبعض البيانات على الأعمدة، والمدة التي نرغب في تنفيذ التدريب عليها، ومؤشر إلى مجموعة البيانات. يُرجى العِلم أنّنا نوفّر ناتج المكوّن السابق عبر dataset_create_op.outputs["dataset"] لتمرير مجموعة البيانات إلى هذا المكوّن.
    • EndpointCreateOp لإنشاء نقطة نهاية في Vertex AI سيتمّ تمرير نقطة النهاية التي تمّ إنشاؤها من هذه الخطوة كمدخل إلى المكوّن التالي.
    • ينشر ModelDeployOp نموذجًا معيّنًا إلى نقطة نهاية في Vertex AI. في هذه الحالة، سنستخدم نقطة النهاية التي تم إنشاؤها في الخطوة السابقة. تتوفّر خيارات إعدادات إضافية، ولكننا سنقدّم هنا نوع جهاز نقطة النهاية وطرازه اللذين نريد نشرهما. يتم إدخال النموذج من خلال الوصول إلى نتائج خطوة التدريب في مسار الإحالة الناجحة.
  • تستخدِم هذه السلسلة أيضًا المنطق الشَرطي، وهي ميزة في Vertex Pipelines تتيح لك تحديد شرط، بالإضافة إلى فروع مختلفة استنادًا إلى نتيجة هذا الشرط. تذكَّر أنّنا عندما حدّدنا مسار الإحالة الناجحة، مررنا بمَعلمة thresholds_dict_str. هذا هو الحدّ الأدنى للدقة الذي نستخدمه لتحديد ما إذا كان سيتم نشر النموذج على نقطة نهاية. لتنفيذ ذلك، نستخدم الفئة Condition من حزمة تطوير البرامج (SDK) الخاصة بـ KFP. الشرط الذي نقْدمه هو ناتج مكوّن التقييم المخصّص الذي حدّدناه سابقًا في هذا الدليل التعليمي. إذا كان هذا الشرط صحيحًا، سيواصل مسار الإحالة الناجحة تنفيذ المكوّن deploy_op. وإذا لم تتوافق الدقة مع الحد الأدنى المحدد مسبقًا، سيتوقف المسار هنا ولن يتم نشر أي نموذج.

الخطوة 3: تجميع مسار ML من البداية إلى النهاية وتشغيله

بعد تحديد مسار الإحالة الناجحة بالكامل، حان وقت تجميعه:

compiler.Compiler().compile(
    pipeline_func=pipeline, package_path="tab_classif_pipeline.json"
)

بعد ذلك، حدد الوظيفة:

ml_pipeline_job = aiplatform.PipelineJob(
    display_name="automl-tab-beans-training",
    template_path="tab_classif_pipeline.json",
    pipeline_root=PIPELINE_ROOT,
    parameter_values={"project": PROJECT_ID, "display_name": DISPLAY_NAME},
    enable_caching=True
)

وأخيرًا، يمكنك تنفيذ المهمة:

ml_pipeline_job.submit()

انتقِل إلى الرابط المعروض في السجلّات بعد تنفيذ الخلية أعلاه للاطّلاع على مسار الإحالة الناجحة في وحدة التحكّم. ستستغرق عملية تنفيذ مسار الإحالة الناجحة هذه ما يزيد قليلاً عن ساعة. يتمّ إنفاق معظم الوقت في خطوة تدريب AutoML. سيبدو مسار الإحالة الناجحة المكتمل على النحو التالي:

مسار AutoML المكتمل

إذا فعّلت الزر "توسيع العناصر" في أعلى الصفحة، ستتمكّن من الاطّلاع على تفاصيل العناصر المختلفة التي تم إنشاؤها من مسار الإحالة الناجحة. على سبيل المثال، إذا نقرت على العنصر dataset، ستظهر لك تفاصيل حول مجموعة بيانات Vertex AI التي تم إنشاؤها. يمكنك النقر على الرابط هنا للانتقال إلى صفحة مجموعة البيانات هذه:

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

وبالمثل، للاطّلاع على الرسومات البيانية للمقاييس الناتجة من مكوّن التقييم المخصّص، انقر على العنصر المُسمى metricsc. على يسار لوحة البيانات، ستتمكّن من الاطّلاع على مصفوفة الالتباس لهذا النموذج:

تصور المقاييس

للاطّلاع على النموذج ونقطة النهاية اللذين تم إنشاؤهما من مسار المسار هذا، انتقِل إلى قسم النماذج وانقر على النموذج المُسمّى automl-beans. من المفترض أن يظهر لك هذا النموذج مُنشرًا في نقطة نهاية:

Model-endpoint

يمكنك أيضًا الوصول إلى هذه الصفحة من خلال النقر على عنصر نقطة النهاية في الرسم البياني لمسار الإحالة الناجحة.

بالإضافة إلى إلقاء نظرة على الرسم البياني لمخطط البيانات في وحدة التحكّم، يمكنك أيضًا استخدام مسارات Vertex لتتبُّع البيانات الخطية. نقصد بتتبُّع سلسلة الإحالة تتبع العناصر التي تم إنشاؤها في جميع مراحل عملية المعالجة. ويمكن أن يساعدنا ذلك في معرفة مكان إنشاء العناصر وكيفية استخدامها خلال سير عمل الذكاء الاصطناعي. على سبيل المثال، للاطّلاع على تتبُّع سلسلة الإحالة لمجموعة البيانات التي تم إنشاؤها في هذه السلسلة، انقر على عنصر مجموعة البيانات ثم على عرض سلسلة الإحالة:

عرض سلسلة الإحالة

يوضّح لنا ذلك جميع الأماكن التي يتم فيها استخدام هذا العنصر:

تفاصيل النسب

الخطوة 4: مقارنة المقاييس على مستوى عمليات تنفيذ مسار الإحالة الناجحة

في حال تنفيذ هذه السلسلة من الإجراءات عدّة مرات، قد تحتاج إلى مقارنة المقاييس على مستوى عمليات التنفيذ. يمكنك استخدام الطريقة aiplatform.get_pipeline_df() للوصول إلى البيانات الوصفية للتنفيذ. هنا، سنحصل على بيانات وصفية لجميع عمليات تنفيذ هذه السلسلة من الإجراءات وسنحمّلها إلى إطار بيانات Pandas:

pipeline_df = aiplatform.get_pipeline_df(pipeline="automl-tab-beans-training-v2")
small_pipeline_df = pipeline_df.head(2)
small_pipeline_df

بذلك تكون قد أكملت الدرس التطبيقي.

🎉 تهانينا. 🎉

لقد تعلمت كيفية استخدام Vertex AI لإجراء ما يلي:

  • استخدام حزمة تطوير البرامج (SDK) لـ Kubeflow Pipelines لإنشاء قنوات عمل متكاملة باستخدام مكوّنات مخصّصة
  • تشغيل المسارات على Vertex Pipelines وبدء عمليات التشغيل باستخدام حزمة تطوير البرامج (SDK)
  • عرض الرسم البياني لمسار الإحالة الناجحة في Vertex Pipelines وتحليله في وحدة التحكّم
  • استخدِم مكوّنات جاهزة مصمَّمة مسبقًا لإضافة خدمات Vertex AI إلى مسار التعلّم
  • جدولة المهام المتكررة لمسار التعلّم

لمعرفة المزيد عن الأجزاء المختلفة من Vertex، اطّلِع على المستندات.

7- تنظيف

لكي لا يتم تحصيل رسوم منك، ننصحك بحذف الموارد التي تم إنشاؤها خلال هذا الدرس العملي.

الخطوة 1: إيقاف أو حذف نسخة "دفاتر ملاحظات Google"

إذا أردت مواصلة استخدام دفتر البيانات الذي أنشأته في هذا الدرس العملي، ننصحك بإيقافه عندما لا يكون قيد الاستخدام. من واجهة مستخدم "دفاتر ملاحظات Google" في Cloud Console، اختَر دفتر الملاحظات ثم اختَر إيقاف. إذا كنت تريد حذف المثيل بالكامل، انقر على حذف:

إيقاف المثيل

الخطوة 2: حذف نقطة النهاية

لحذف نقطة النهاية التي تم نشرها، انتقِل إلى قسم نقاط النهاية في وحدة تحكّم Vertex AI وانقر على رمز الحذف:

حذف نقطة النهاية

بعد ذلك، انقر على إلغاء النشر من الطلب التالي:

إلغاء نشر النموذج

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

حذف النموذج

الخطوة 3: حذف حزمة Cloud Storage

لحذف "حزمة التخزين"، باستخدام قائمة التنقّل في Cloud Console، انتقِل إلى "مساحة التخزين" واختَر الحزمة وانقر على "حذف":

حذف مساحة التخزين