ساخت اجزای کاملا سفارشی

این راهنما نحوه استفاده از TFX API را برای ساخت یک کامپوننت کاملا سفارشی توضیح می دهد. کامپوننت های کاملا سفارشی به شما این امکان را می دهند که با تعریف مشخصات کامپوننت، مجری و کلاس های رابط کامپوننت، کامپوننت ها را بسازید. این رویکرد به شما این امکان را می‌دهد که از یک جزء استاندارد مطابق با نیازهای خود استفاده مجدد و گسترش دهید.

اگر با خطوط لوله TFX تازه کار هستید، درباره مفاهیم اصلی خطوط لوله TFX بیشتر بدانید .

مجری سفارشی یا جزء سفارشی

اگر فقط منطق پردازش سفارشی مورد نیاز باشد در حالی که ورودی ها، خروجی ها و ویژگی های اجرایی کامپوننت همانند یک جزء موجود است، یک مجری سفارشی کافی است. زمانی که هر یک از ورودی‌ها، خروجی‌ها یا ویژگی‌های اجرایی متفاوت از هر مؤلفه TFX موجود باشد، یک مؤلفه کاملاً سفارشی مورد نیاز است.

چگونه یک کامپوننت سفارشی ایجاد کنیم؟

توسعه یک جزء کاملاً سفارشی نیاز دارد:

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

ComponentSpec

کلاس ComponentSpec قرارداد کامپوننت را با تعریف مصنوعات ورودی و خروجی برای یک جزء و همچنین پارامترهایی که برای اجرای کامپوننت استفاده می شود، تعریف می کند. دارای سه بخش است:

• INPUTS : فرهنگ لغت پارامترهای تایپ شده برای مصنوعات ورودی که به مجری مؤلفه ارسال می شوند. معمولاً مصنوعات ورودی خروجی‌های اجزای بالادستی هستند و بنابراین نوع مشابهی دارند.
• خروجی ها : فرهنگ لغت پارامترهای تایپ شده برای مصنوعات خروجی که جزء تولید می کند.
• PARAMETERS : دیکشنری از آیتم های ExecutionParameter اضافی که به مجری مؤلفه ارسال می شود. اینها پارامترهای غیر مصنوع هستند که می‌خواهیم به صورت انعطاف‌پذیر در خط لوله DSL تعریف کنیم و اجرا کنیم.

در اینجا یک نمونه از ComponentSpec آورده شده است:

class HelloComponentSpec(types.ComponentSpec):
  """ComponentSpec for Custom TFX Hello World Component."""

  PARAMETERS = {
      # These are parameters that will be passed in the call to
      # create an instance of this component.
      'name': ExecutionParameter(type=Text),
  }
  INPUTS = {
      # This will be a dictionary with input artifacts, including URIs
      'input_data': ChannelParameter(type=standard_artifacts.Examples),
  }
  OUTPUTS = {
      # This will be a dictionary which this component will populate
      'output_data': ChannelParameter(type=standard_artifacts.Examples),
  }

مجری

سپس کد اجرایی کامپوننت جدید را بنویسید. اساساً، یک زیر کلاس جدید از base_executor.BaseExecutor باید با بازنویسی تابع Do آن ایجاد شود. در تابع Do ، آرگومان‌های input_dict ، output_dict و exec_properties که به ترتیب در نقشه به INPUTS ، OUTPUTS و PARAMETERS منتقل می‌شوند که در ComponentSpec تعریف شده‌اند. برای exec_properties ، مقدار را می توان مستقیماً از طریق جستجوی فرهنگ لغت واکشی کرد. برای مصنوعات در input_dict و output_dict ، توابع مناسبی در کلاس artifact_utils وجود دارد که می‌توان از آنها برای واکشی نمونه مصنوع یا آرتیفکت uri استفاده کرد.

class Executor(base_executor.BaseExecutor):
  """Executor for HelloComponent."""

  def Do(self, input_dict: Dict[Text, List[types.Artifact]],
         output_dict: Dict[Text, List[types.Artifact]],
         exec_properties: Dict[Text, Any]) -> None:
    ...

    split_to_instance = {}
    for artifact in input_dict['input_data']:
      for split in json.loads(artifact.split_names):
        uri = artifact_utils.get_split_uri([artifact], split)
        split_to_instance[split] = uri

    for split, instance in split_to_instance.items():
      input_dir = instance
      output_dir = artifact_utils.get_split_uri(
          output_dict['output_data'], split)
      for filename in tf.io.gfile.listdir(input_dir):
        input_uri = os.path.join(input_dir, filename)
        output_uri = os.path.join(output_dir, filename)
        io_utils.copy_file(src=input_uri, dst=output_uri, overwrite=True)

واحد آزمایش یک مجری سفارشی

تست های واحد برای مجری سفارشی می توانند مشابه این مورد ایجاد شوند.

رابط کامپوننت

اکنون که پیچیده ترین قسمت کامل شده است، مرحله بعدی مونتاژ این قطعات در یک رابط کامپوننت است تا بتوان از کامپوننت در خط لوله استفاده کرد. چندین مرحله وجود دارد:

• رابط کامپوننت را به زیر کلاس base_component.BaseComponent تبدیل کنید
• یک متغیر کلاس SPEC_CLASS را با کلاس ComponentSpec که قبلاً تعریف شده بود، اختصاص دهید
• یک متغیر کلاس EXECUTOR_SPEC را به کلاس Executor که قبلاً تعریف شده بود اختصاص دهید
• تابع سازنده __init__() را با استفاده از آرگومان های تابع برای ساختن نمونه ای از کلاس ComponentSpec و فراخوانی تابع super با آن مقدار به همراه یک نام اختیاری تعریف کنید.

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

from tfx.types import standard_artifacts
from hello_component import executor

class HelloComponent(base_component.BaseComponent):
  """Custom TFX Hello World Component."""

  SPEC_CLASS = HelloComponentSpec
  EXECUTOR_SPEC = executor_spec.ExecutorClassSpec(executor.Executor)

  def __init__(self,
               input_data: types.Channel = None,
               output_data: types.Channel = None,
               name: Optional[Text] = None):
    if not output_data:
      examples_artifact = standard_artifacts.Examples()
      examples_artifact.split_names = input_data.get()[0].split_names
      output_data = channel_utils.as_channel([examples_artifact])

    spec = HelloComponentSpec(input_data=input_data,
                              output_data=output_data, name=name)
    super(HelloComponent, self).__init__(spec=spec)

در یک خط لوله TFX جمع آوری کنید

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

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

مثال زیر تغییرات فوق را برجسته می کند. مثال کامل را می توان در مخزن TFX GitHub یافت.

def _create_pipeline():
  ...
  example_gen = CsvExampleGen(input_base=examples)
  hello = component.HelloComponent(
      input_data=example_gen.outputs['examples'], name='HelloWorld')
  statistics_gen = StatisticsGen(examples=hello.outputs['output_data'])
  ...
  return pipeline.Pipeline(
      ...
      components=[example_gen, hello, statistics_gen, ...],
      ...
  )

یک جزء کاملاً سفارشی را مستقر کنید

علاوه بر تغییرات کد، تمام قطعات جدید اضافه شده ( ComponentSpec ، Executor ، رابط مؤلفه) باید در محیط اجرای خط لوله قابل دسترسی باشند تا خط لوله را به درستی اجرا کنند.