بناء خط أنابيب TFX محليًا

تعمل TFX على تسهيل تنسيق سير عمل التعلم الآلي (ML) الخاص بك كمسار، من أجل:

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

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

• قم بتخصيص قالب خط أنابيب TFX ليناسب احتياجات سير عمل ML لديك. قوالب خطوط أنابيب TFX عبارة عن مسارات عمل تم إنشاؤها مسبقًا توضح أفضل الممارسات باستخدام مكونات TFX القياسية.
• بناء خط أنابيب باستخدام TFX. في حالة الاستخدام هذه، يمكنك تحديد مسار دون البدء من القالب.

أثناء قيامك بتطوير خط الأنابيب الخاص بك، يمكنك تشغيله باستخدام LocalDagRunner . وبعد ذلك، بمجرد تحديد واختبار مكونات خط الأنابيب بشكل جيد، يمكنك استخدام منسق على مستوى الإنتاج مثل Kubeflow أو Airflow.

قبل أن تبدأ

TFX هي حزمة Python، لذا ستحتاج إلى إعداد بيئة تطوير Python، مثل بيئة افتراضية أو حاوية Docker. ثم:

pip install tfx

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

بناء خط أنابيب باستخدام القالب

تسهل قوالب خطوط أنابيب TFX البدء في تطوير خطوط الأنابيب من خلال توفير مجموعة محددة مسبقًا من تعريفات خطوط الأنابيب التي يمكنك تخصيصها لحالة الاستخدام الخاصة بك.

تصف الأقسام التالية كيفية إنشاء نسخة من القالب وتخصيصها لتلبية احتياجاتك.

قم بإنشاء نسخة من قالب خط الأنابيب

1. راجع قائمة قوالب خطوط أنابيب TFX المتاحة:

   tfx template list
   

2. حدد قالبًا من القائمة

   tfx template copy --model=template --pipeline_name=pipeline-name \
   --destination_path=destination-path
   

   استبدل ما يلي:

   • template : اسم القالب الذي تريد نسخه.
   • pipeline-name : اسم خط الأنابيب المراد إنشاؤه.
   • destination-path : المسار الذي سيتم نسخ القالب إليه.

   تعرف على المزيد حول أمر tfx template copy .

3. تم إنشاء نسخة من قالب خط الأنابيب على المسار الذي حددته.

استكشف قالب خط الأنابيب

يقدم هذا القسم نظرة عامة على السقالات التي تم إنشاؤها بواسطة القالب.

1. استكشف الدلائل والملفات التي تم نسخها إلى الدليل الجذر لخط الأنابيب الخاص بك

   • دليل خطوط الأنابيب مع
     • pipeline.py - يحدد المسار، ويسرد المكونات التي يتم استخدامها
     • configs.py - يحتفظ بتفاصيل التكوين مثل مصدر البيانات أو المُنسق الذي يتم استخدامه
   • دليل البيانات
     • يحتوي هذا عادةً على ملف data.csv ، وهو المصدر الافتراضي لـ ExampleGen . يمكنك تغيير مصدر البيانات في configs.py .

   • دليل النماذج الذي يحتوي على تعليمات برمجية للمعالجة المسبقة وتنفيذ النماذج

   • يقوم القالب بنسخ مشغلات DAG للبيئة المحلية وKubeflow.

   • تتضمن بعض القوالب أيضًا دفاتر ملاحظات Python حتى تتمكن من استكشاف بياناتك وعناصرك باستخدام البيانات الوصفية للتعلم الآلي.

2. قم بتشغيل الأوامر التالية في دليل خط الأنابيب الخاص بك:

   tfx pipeline create --pipeline_path local_runner.py
   

   tfx run create --pipeline_name pipeline_name
   

   يقوم الأمر بإنشاء خط أنابيب يتم تشغيله باستخدام LocalDagRunner ، والذي يضيف الدلائل التالية إلى خط الأنابيب الخاص بك:

   • دليل tfx_metadata الذي يحتوي على مخزن بيانات تعريف ML المستخدم محليًا.
   • دليل tfx_pipeline_output الذي يحتوي على مخرجات ملف خط الأنابيب.

3. افتح ملف pipeline/configs.py الأنابيب الخاص بك وراجع محتوياته. يحدد هذا البرنامج النصي خيارات التكوين التي يستخدمها المسار ووظائف المكونات. هذا هو المكان الذي يمكنك فيه تحديد أشياء مثل موقع مصدر البيانات أو عدد خطوات التدريب في التشغيل.

4. افتح ملف pipeline/pipeline.py الخاص بخط الأنابيب الخاص بك وراجع محتوياته. يقوم هذا البرنامج النصي بإنشاء خط أنابيب TFX. في البداية، يحتوي المسار على مكون ExampleGen فقط.

   • اتبع الإرشادات الواردة في تعليقات TODO في pipeline.py لإضافة المزيد من الخطوات إلى المسار.

5. افتح ملف local_runner.py وراجع محتوياته. يقوم هذا البرنامج النصي بإنشاء مسار تشغيل ويحدد معلمات التشغيل، مثل data_path و preprocessing_fn .

6. لقد قمت بمراجعة السقالات التي أنشأها القالب وقمت بإنشاء مسار تشغيل باستخدام LocalDagRunner . بعد ذلك، قم بتخصيص القالب ليناسب متطلباتك.

تخصيص خط الأنابيب الخاص بك

يقدم هذا القسم نظرة عامة حول كيفية البدء في تخصيص القالب الخاص بك.

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

2. بمجرد تصميم المسار الخاص بك، قم بتخصيص المسار بشكل متكرر باستخدام العملية التالية. ابدأ من المكون الذي يستوعب البيانات في المسار الخاص بك، والذي عادةً ما يكون مكون ExampleGen .

   1. قم بتخصيص المسار أو المكون ليناسب حالة الاستخدام الخاصة بك. قد تتضمن هذه التخصيصات تغييرات مثل:

      • تغيير معلمات خط الأنابيب.
      • إضافة مكونات إلى خط الأنابيب أو إزالتها.
      • استبدال مصدر إدخال البيانات. يمكن أن يكون مصدر البيانات هذا ملفًا أو استعلاماتًا في خدمات مثل BigQuery.
      • تغيير تكوين المكون في خط الأنابيب.
      • تغيير وظيفة تخصيص المكون.

   2. قم بتشغيل المكون محليًا باستخدام البرنامج النصي local_runner.py ، أو مشغل DAG مناسب آخر إذا كنت تستخدم منسقًا مختلفًا. إذا فشل البرنامج النصي، فقم بتصحيح الفشل وأعد محاولة تشغيل البرنامج النصي.

   3. بمجرد أن يعمل هذا التخصيص، انتقل إلى التخصيص التالي.

3. من خلال العمل بشكل متكرر، يمكنك تخصيص كل خطوة في سير عمل القالب لتلبية احتياجاتك.

بناء خط أنابيب مخصص

استخدم الإرشادات التالية لمعرفة المزيد حول إنشاء مسار مخصص دون استخدام قالب.

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

   • تعرف على المزيد حول مكونات TFX القياسية .
   • تعرف على المزيد حول المكونات المخصصة .

2. قم بإنشاء ملف نصي لتحديد خط الأنابيب الخاص بك باستخدام المثال التالي. يشير هذا الدليل إلى هذا الملف باسم my_pipeline.py .

   import os
   from typing import Optional, Text, List
   from absl import logging
   from ml_metadata.proto import metadata_store_pb2
   import tfx.v1 as tfx
   
   PIPELINE_NAME = 'my_pipeline'
   PIPELINE_ROOT = os.path.join('.', 'my_pipeline_output')
   METADATA_PATH = os.path.join('.', 'tfx_metadata', PIPELINE_NAME, 'metadata.db')
   ENABLE_CACHE = True
   
   def create_pipeline(
     pipeline_name: Text,
     pipeline_root:Text,
     enable_cache: bool,
     metadata_connection_config: Optional[
       metadata_store_pb2.ConnectionConfig] = None,
     beam_pipeline_args: Optional[List[Text]] = None
   ):
     components = []
   
     return tfx.dsl.Pipeline(
           pipeline_name=pipeline_name,
           pipeline_root=pipeline_root,
           components=components,
           enable_cache=enable_cache,
           metadata_connection_config=metadata_connection_config,
           beam_pipeline_args=beam_pipeline_args, <!-- needed? -->
       )
   
   def run_pipeline():
     my_pipeline = create_pipeline(
         pipeline_name=PIPELINE_NAME,
         pipeline_root=PIPELINE_ROOT,
         enable_cache=ENABLE_CACHE,
         metadata_connection_config=tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)
         )
   
     tfx.orchestration.LocalDagRunner().run(my_pipeline)
   
   if __name__ == '__main__':
     logging.set_verbosity(logging.INFO)
     run_pipeline()

   في الخطوات التالية، يمكنك تحديد خط الأنابيب الخاص بك في create_pipeline وتشغيل خط الأنابيب الخاص بك محليًا باستخدام المشغل المحلي.

   قم ببناء خط الأنابيب الخاص بك بشكل متكرر باستخدام العملية التالية.

   1. قم بتخصيص المسار أو المكون ليناسب حالة الاستخدام الخاصة بك. قد تتضمن هذه التخصيصات تغييرات مثل:

      • تغيير معلمات خط الأنابيب.
      • إضافة مكونات إلى خط الأنابيب أو إزالتها.
      • استبدال ملف إدخال البيانات.
      • تغيير تكوين المكون في خط الأنابيب.
      • تغيير وظيفة تخصيص المكون.

   2. قم بتشغيل المكون محليًا باستخدام برنامج التشغيل المحلي أو عن طريق تشغيل البرنامج النصي مباشرةً. إذا فشل البرنامج النصي، فقم بتصحيح الفشل وأعد محاولة تشغيل البرنامج النصي.

   3. بمجرد أن يعمل هذا التخصيص، انتقل إلى التخصيص التالي.

   ابدأ من العقدة الأولى في سير عمل خط الأنابيب الخاص بك، وعادةً ما تقوم العقدة الأولى بإدخال البيانات في خط الأنابيب الخاص بك.

3. أضف العقدة الأولى في سير العمل الخاص بك إلى خط الأنابيب الخاص بك. في هذا المثال، يستخدم المسار المكون القياسي ExampleGen لتحميل ملف CSV من دليل في ./data .

   from tfx.components import CsvExampleGen
   
   DATA_PATH = os.path.join('.', 'data')
   
   def create_pipeline(
     pipeline_name: Text,
     pipeline_root:Text,
     data_path: Text,
     enable_cache: bool,
     metadata_connection_config: Optional[
       metadata_store_pb2.ConnectionConfig] = None,
     beam_pipeline_args: Optional[List[Text]] = None
   ):
     components = []
   
     example_gen = tfx.components.CsvExampleGen(input_base=data_path)
     components.append(example_gen)
   
     return tfx.dsl.Pipeline(
           pipeline_name=pipeline_name,
           pipeline_root=pipeline_root,
           components=components,
           enable_cache=enable_cache,
           metadata_connection_config=metadata_connection_config,
           beam_pipeline_args=beam_pipeline_args, <!-- needed? -->
       )
   
   def run_pipeline():
     my_pipeline = create_pipeline(
       pipeline_name=PIPELINE_NAME,
       pipeline_root=PIPELINE_ROOT,
       data_path=DATA_PATH,
       enable_cache=ENABLE_CACHE,
       metadata_connection_config=tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)
       )
   
     tfx.orchestration.LocalDagRunner().run(my_pipeline)

   يقوم CsvExampleGen بإنشاء سجلات أمثلة متسلسلة باستخدام البيانات الموجودة في ملف CSV في مسار البيانات المحدد. عن طريق تعيين معلمة input_base الخاصة بمكون CsvExampleGen باستخدام جذر البيانات.

4. قم بإنشاء دليل data في نفس الدليل مثل my_pipeline.py . أضف ملف CSV صغيرًا إلى دليل data .

5. استخدم الأمر التالي لتشغيل البرنامج النصي my_pipeline.py .

   python my_pipeline.py
   

   يجب أن تكون النتيجة شيء مثل ما يلي:

   INFO:absl:Component CsvExampleGen depends on [].
   INFO:absl:Component CsvExampleGen is scheduled.
   INFO:absl:Component CsvExampleGen is running.
   INFO:absl:Running driver for CsvExampleGen
   INFO:absl:MetadataStore with DB connection initialized
   INFO:absl:Running executor for CsvExampleGen
   INFO:absl:Generating examples.
   INFO:absl:Using 1 process(es) for Local pipeline execution.
   INFO:absl:Processing input csv data ./data/* to TFExample.
   WARNING:root:Couldn't find python-snappy so the implementation of _TFRecordUtil._masked_crc32c is not as fast as it could be.
   INFO:absl:Examples generated.
   INFO:absl:Running publisher for CsvExampleGen
   INFO:absl:MetadataStore with DB connection initialized
   INFO:absl:Component CsvExampleGen is finished.
   

6. استمر في إضافة المكونات بشكل متكرر إلى خط الأنابيب الخاص بك.