หน้านี้ได้รับการแปลโดย Cloud Translation API
Switch to English

การสร้างส่วนประกอบที่กำหนดเองทั้งหมด

คู่มือนี้อธิบายถึงวิธีการใช้ TFX API เพื่อสร้างคอมโพเนนต์แบบกำหนดเองทั้งหมด คอมโพเนนต์แบบกำหนดเองทั้งหมดช่วยให้คุณสามารถสร้างคอมโพเนนต์ได้โดยกำหนดคุณสมบัติของคอมโพเนนต์ตัวดำเนินการและคลาสอินเทอร์เฟซของคอมโพเนนต์ วิธีนี้ช่วยให้คุณสามารถนำมาใช้ซ้ำและขยายส่วนประกอบมาตรฐานให้เหมาะกับความต้องการของคุณ

หากคุณเพิ่งเริ่มใช้งานท่อ TFX โปรด เรียนรู้เพิ่มเติมเกี่ยวกับแนวคิดหลักของไปป์ไลน์ TFX

ตัวดำเนินการที่กำหนดเองหรือส่วนประกอบที่กำหนดเอง

หากต้องการเฉพาะตรรกะการประมวลผลแบบกำหนดเองในขณะที่คุณสมบัติอินพุตเอาต์พุตและการดำเนินการของคอมโพเนนต์เหมือนกับส่วนประกอบที่มีอยู่ตัวดำเนินการแบบกำหนดเองก็เพียงพอแล้ว จำเป็นต้องใช้คอมโพเนนต์แบบกำหนดเองโดยสมบูรณ์เมื่อคุณสมบัติอินพุตเอาต์พุตหรือการดำเนินการใด ๆ แตกต่างจากส่วนประกอบ TFX ที่มีอยู่

จะสร้างส่วนประกอบที่กำหนดเองได้อย่างไร?

การพัฒนาองค์ประกอบที่กำหนดเองทั้งหมดต้องการ:

  • ชุดข้อมูลจำเพาะของสิ่งประดิษฐ์อินพุตและเอาต์พุตที่กำหนดไว้สำหรับส่วนประกอบใหม่ โดยเฉพาะอย่างยิ่งประเภทของสิ่งประดิษฐ์อินพุตควรสอดคล้องกับประเภทอาร์ติแฟกต์เอาต์พุตของคอมโพเนนต์ที่สร้างอาร์ติแฟกต์และประเภทของอาร์ติแฟกต์เอาต์พุตควรสอดคล้องกับประเภทอินพุตอาร์ติแฟกต์ของส่วนประกอบที่ใช้อาร์ติแฟกต์หากมี
  • พารามิเตอร์การดำเนินการที่ไม่ใช่อาร์ติแฟกต์ที่จำเป็นสำหรับคอมโพเนนต์ใหม่

ComponentSpec

คลาส ComponentSpec กำหนดสัญญาคอมโพเนนต์โดยการกำหนดส่วนของอินพุตและเอาต์พุตให้กับคอมโพเนนต์ตลอดจนพารามิเตอร์ที่ใช้สำหรับการเรียกใช้คอมโพเนนต์ มีสามส่วนในนั้น:

  • ปัจจัยการผลิต: พจนานุกรมของพารามิเตอร์พิมพ์สำหรับสิ่งประดิษฐ์การป้อนข้อมูลที่มีเข้ามาในปฏิบัติการส่วนประกอบ โดยปกติอินพุตอาร์ติแฟกต์คือเอาต์พุตจากคอมโพเนนต์ต้นน้ำและแบ่งประเภทเดียวกัน
  • เอาท์พุท : พจนานุกรมของพารามิเตอร์ที่พิมพ์สำหรับสิ่งประดิษฐ์เอาต์พุตที่คอมโพเนนต์สร้างขึ้น
  • 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 คุณสามารถดึงค่าได้โดยตรงผ่านการค้นหาพจนานุกรม สำหรับ output_dict ใน input_dict และ output_dict มีฟังก์ชันที่สะดวกพร้อมใช้งานในคลาส artifact_utils ที่สามารถใช้เพื่อดึงอินสแตนซ์ output_dict หรือสิ่งประดิษฐ์ 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 = os.path.join(artifact.uri, 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 นอกจากการเพิ่มอินสแตนซ์ขององค์ประกอบใหม่แล้วยังจำเป็นต้องมีสิ่งต่อไปนี้:

  • เชื่อมต่อส่วนประกอบต้นน้ำและปลายน้ำของส่วนประกอบใหม่อย่างถูกต้อง สิ่งนี้ทำได้โดยการอ้างอิงเอาต์พุตขององค์ประกอบต้นน้ำในส่วนประกอบใหม่และอ้างอิงผลลัพธ์ของส่วนประกอบใหม่ในส่วนประกอบปลายน้ำ
  • เพิ่มอินสแตนซ์คอมโพเนนต์ใหม่ในรายการคอมโพเนนต์เมื่อสร้างไปป์ไลน์

ตัวอย่างด้านล่างเน้นการเปลี่ยนแปลงที่กล่าวมา ตัวอย่างเต็มสามารถพบได้ใน repo 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 , ส่วนต่อประสานคอมโพเนนต์) จะต้องสามารถเข้าถึงได้ในสภาพแวดล้อมที่ทำงานไปป์ไลน์เพื่อให้รันไปป์ไลน์ได้อย่างถูกต้อง