ساخت خط لوله TFX به صورت محلی

TFX تنظیم گردش کار یادگیری ماشین (ML) خود را به عنوان خط لوله آسان‌تر می‌کند تا:

  • فرآیند ML خود را خودکار کنید، که به شما امکان می دهد به طور منظم مدل خود را بازآموزی، ارزیابی و استقرار دهید.
  • خطوط لوله ML را ایجاد کنید که شامل تجزیه و تحلیل عمیق عملکرد مدل و اعتبارسنجی مدل های تازه آموزش دیده برای اطمینان از عملکرد و قابلیت اطمینان است.
  • نظارت بر داده های تمرین برای ناهنجاری ها و حذف انحراف در زمان ارائه آموزش
  • سرعت آزمایش را با اجرای یک خط لوله با مجموعه های مختلف فراپارامترها افزایش دهید.

یک فرآیند توسعه خط لوله معمولی در یک ماشین محلی، با تجزیه و تحلیل داده‌ها و راه‌اندازی اجزا، قبل از استقرار در تولید آغاز می‌شود. این راهنما دو روش برای ساخت یک خط لوله به صورت محلی را شرح می دهد.

  • یک الگوی خط لوله TFX را متناسب با نیازهای گردش کار ML خود سفارشی کنید. الگوهای خط لوله TFX گردش کار از پیش ساخته شده ای هستند که بهترین شیوه ها را با استفاده از مؤلفه های استاندارد TFX نشان می دهند.
  • با استفاده از TFX یک خط لوله بسازید. در این مورد، شما یک خط لوله را بدون شروع از یک الگو تعریف می کنید.

همانطور که در حال توسعه خط لوله خود هستید، می توانید آن را با LocalDagRunner اجرا کنید. سپس، هنگامی که اجزای خط لوله به خوبی تعریف و آزمایش شدند، می توانید از یک ارکستراتور درجه تولید مانند Kubeflow یا Airflow استفاده کنید.

قبل از شروع

TFX یک بسته پایتون است، بنابراین شما باید یک محیط توسعه پایتون، مانند یک محیط مجازی یا یک کانتینر داکر را راه اندازی کنید. سپس:

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 کپی می کند.

    • برخی از قالب‌ها شامل نوت‌بوک‌های پایتون هستند تا بتوانید داده‌ها و مصنوعات خود را با فراداده‌های یادگیری ماشینی کاوش کنید.

  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 pipeline/configs.py لوله خود را باز کنید و محتویات آن را بررسی کنید. این اسکریپت گزینه های پیکربندی مورد استفاده توسط خط لوله و توابع جزء را تعریف می کند. اینجاست که می‌توانید مواردی مانند مکان منبع داده یا تعداد مراحل آموزش در یک اجرا را مشخص کنید.

  4. فایل pipeline 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 ارائه می دهند. اگر یک جریان کاری ML موجود را به یک خط لوله منتقل می کنید، ممکن است لازم باشد کد خود را برای استفاده کامل از مؤلفه های استاندارد 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, 
    )

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 تعریف می کنید و با استفاده از runner محلی، خط لوله خود را به صورت محلی اجرا می کنید.

    به طور مکرر خط لوله خود را با استفاده از فرآیند زیر بسازید.

    1. خط لوله یا یک جزء را متناسب با مورد استفاده خود سفارشی کنید. این سفارشی‌سازی‌ها ممکن است شامل تغییراتی مانند موارد زیر باشد:

      • تغییر پارامترهای خط لوله
      • افزودن قطعات به خط لوله یا حذف آنها.
      • جایگزینی فایل ورودی داده
      • تغییر پیکربندی یک جزء در خط لوله.
      • تغییر تابع سفارشی سازی یک جزء

    2. کامپوننت را به صورت محلی با استفاده از runner محلی یا با اجرای مستقیم اسکریپت اجرا کنید. اگر اسکریپت ناموفق بود، خطا را رفع اشکال کنید و دوباره اسکریپت را اجرا کنید.

    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, 
    )

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. به افزودن اجزای مکرر به خط لوله خود ادامه دهید.