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

1. نظرة عامة

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

ما ستتعلمه

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

  • استخدام Kubeflow Pipelines SDK لإنشاء خطوط أنابيب قابلة للتوسيع خاصة بتعلُّم الآلة
  • إنشاء مسار تمهيدي من 3 خطوات وتنفيذه، مع إدخال نص
  • إنشاء وتنفيذ مسار يعلّم نموذج تصنيف AutoML ويقيّمه وينشره
  • استخدام مكوّنات مُنشأة مسبقًا للتفاعل مع خدمات 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

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

لماذا تكون مسارات تعلُّم الآلة مفيدة؟

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

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

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 init

يتم تحميل هذا الجهاز الافتراضي بجميع أدوات التطوير التي تحتاج إليها. توفّر هذه الخدمة دليلًا رئيسيًا دائمًا بسعة 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 الذي يحتوي على اسم مشروع على السحابة الإلكترونية الحالي. سنستخدم هذا المتغير في مواضع مختلفة خلال هذا الدرس التطبيقي. يمكنك الاطّلاع عليه من خلال تنفيذ الأمر التالي:

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 Pipelines

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

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

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

أولاً، من قائمة "مشغّل التطبيقات" في مثيل "دفتر الملاحظات"، أنشئ دفتر ملاحظات من خلال اختيار 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: ضبط رقم تعريف المشروع والحزمة

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

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

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، سننشئ أولاً مسارًا قصيرًا باستخدام حزمة تطوير البرامج (SDK) الخاصة بمنصة KFP. لا تتضمّن هذه السلسلة أي خطوات متعلقة بالتعلم الآلي (لا تقلق، سنتناول ذلك لاحقًا)، بل نستخدمها لتعليمك ما يلي:

  • كيفية إنشاء مكوّنات مخصّصة في حزمة تطوير البرامج (SDK) لمنصة KFP
  • كيفية تنفيذ ومراقبة مسار في 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 المكوّن بأي مهام تابعة لمكتبة خارجية لهذه الحاوية. في هذه الحالة، نستخدم مكتبة باسم emoji.
  • يعرض هذا المكوّن 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 للاطّلاع على الناتج النهائي:

ناتج مسار المقدمة

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

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

حان الوقت لإنشاء أول مسار تعلّم آلي. في مسار التعلّم هذا، سنستخدم مجموعة بيانات الفاصوليا الجافة من مستودع UCI Machine Learning، من: KOKLU، وM. وOZKAN، وI.A. (2020), "Multiclass Classification of Dry Beans Using Computer Vision and Machine Learning Techniques."In 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
  • مقارنة المقاييس بحدّ معيّن لتحديد ما إذا كان يجب نشر النموذج

قبل تحديد المكوّن، دعنا نتعرّف على مَعلمات الإدخال والإخراج. تتلقّى مسار التعلّم هذا كمدخل بعض البيانات الوصفية حول مشروعنا على السحابة الإلكترونية، والنموذج المدرَّب الناتج (سنحدّد هذا المكوّن لاحقًا)، ومقاييس تقييم النموذج، وthresholds_dict_str. thresholds_dict_str هو شيء سنحدّده عند تشغيل خط الإنتاج. في حالة نموذج التصنيف هذا، ستكون هذه هي المساحة تحت قيمة منحنى ROC التي يجب أن ننشر النموذج من أجلها. على سبيل المثال، إذا أدخلنا القيمة 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: تجميع وتشغيل مسار تعلُّم الآلة الشامل

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

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 Pipelines من أجل تتبُّع مصدر البيانات. من خلال تتبُّع مصدر البيانات، نعني تتبُّع العناصر التي تم إنشاؤها في جميع مراحل عملية المعالجة. يمكن أن يساعدنا ذلك في معرفة مكان إنشاء العناصر وكيفية استخدامها في سير عمل تعلُّم الآلة. على سبيل المثال، للاطّلاع على تتبُّع مصدر مجموعة البيانات التي تم إنشاؤها في مسار النقل هذا، انقر على عنصر مجموعة البيانات ثم على عرض المصدر:

عرض تتبُّع سير البيانات

يعرض لنا هذا القسم جميع المواضع التي يتم فيها استخدام هذا العنصر:

تفاصيل مصدر البيانات

الخطوة 4: مقارنة المقاييس على مستوى عمليات تنفيذ خط الأنابيب

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

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: إيقاف مثيل Notebooks أو حذفه

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

إيقاف الجهاز الافتراضي

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

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

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

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

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

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

حذف النموذج

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

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

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