Xây dựng các thành phần hoàn toàn tùy chỉnh

Hướng dẫn này mô tả cách sử dụng API TFX để xây dựng thành phần tùy chỉnh hoàn toàn. Các thành phần hoàn toàn tùy chỉnh cho phép bạn xây dựng các thành phần bằng cách xác định các lớp giao diện thành phần, bộ thực thi và đặc tả thành phần. Cách tiếp cận này cho phép bạn tái sử dụng và mở rộng một thành phần tiêu chuẩn để phù hợp với nhu cầu của mình.

Nếu bạn chưa quen với quy trình TFX, hãy tìm hiểu thêm về các khái niệm cốt lõi của quy trình TFX .

Trình thực thi tùy chỉnh hoặc thành phần tùy chỉnh

Nếu chỉ cần logic xử lý tùy chỉnh trong khi các thuộc tính đầu vào, đầu ra và thực thi của thành phần giống với thành phần hiện có thì một trình thực thi tùy chỉnh là đủ. Cần có một thành phần tùy chỉnh hoàn toàn khi bất kỳ thuộc tính đầu vào, đầu ra hoặc thực thi nào khác với bất kỳ thành phần TFX hiện có nào.

Làm thế nào để tạo một thành phần tùy chỉnh?

Việc phát triển một thành phần tùy chỉnh hoàn toàn yêu cầu:

  • Một tập hợp các thông số kỹ thuật tạo tác đầu vào và đầu ra được xác định cho thành phần mới. Đặc biệt, các loại cho tạo phẩm đầu vào phải nhất quán với các loại tạo phẩm đầu ra của các thành phần tạo ra tạo phẩm đó và các loại cho tạo phẩm đầu ra phải nhất quán với các loại tạo phẩm đầu vào của các thành phần sử dụng tạo phẩm đó nếu có.
  • Các tham số thực thi không phải cấu phần giả cần thiết cho thành phần mới.

Thông số thành phần

Lớp ComponentSpec xác định hợp đồng thành phần bằng cách xác định các tạo phẩm đầu vào và đầu ra cho một thành phần cũng như các tham số được sử dụng để thực thi thành phần. Nó có ba phần:

  • ĐẦU VÀO : Một từ điển gồm các tham số đã nhập cho các tạo phẩm đầu vào được chuyển vào bộ thực thi thành phần. Thông thường các tạo phẩm đầu vào là đầu ra từ các thành phần ngược dòng và do đó có cùng loại.
  • ĐẦU RA : Một từ điển các tham số đã nhập cho các tạo phẩm đầu ra mà thành phần tạo ra.
  • PARAMETERS : Một từ điển gồm các mục ExecutionParameter bổ sung sẽ được chuyển vào bộ thực thi thành phần. Đây là các tham số không phải cấu phần giả mà chúng tôi muốn xác định một cách linh hoạt trong đường ống DSL và chuyển vào thực thi.

Đây là một ví dụ về 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),
  }

Người thi hành

Tiếp theo, viết mã thực thi cho thành phần mới. Về cơ bản, một lớp con mới của base_executor.BaseExecutor cần được tạo với chức năng Do được ghi đè của nó. Trong hàm Do , các đối số input_dict , output_dict và exec_properties được chuyển trong bản đồ tới INPUTS , OUTPUTS và PARAMETERS tương ứng được xác định trong ComponentSpec. Đối với exec_properties , giá trị có thể được tìm nạp trực tiếp thông qua tra cứu từ điển. Đối với các tạo phẩm trong input_dict và output_dict , có các hàm tiện lợi có sẵn trong lớp Artifact_utils có thể được sử dụng để tìm nạp phiên bản tạo tác hoặc uri tạo tác.

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)

Đơn vị kiểm tra một người thực thi tùy chỉnh

Các bài kiểm tra đơn vị cho trình thực thi tùy chỉnh có thể được tạo tương tự như bài kiểm tra này .

Giao diện thành phần

Bây giờ phần phức tạp nhất đã hoàn tất, bước tiếp theo là lắp ráp các phần này thành một giao diện thành phần để cho phép thành phần đó có thể được sử dụng trong quy trình. Có một số bước:

  • Biến giao diện thành phần thành một lớp con của base_component.BaseComponent
  • Gán một biến lớp SPEC_CLASS với lớp ComponentSpec đã được xác định trước đó
  • Gán một biến lớp EXECUTOR_SPEC với lớp Executor đã được xác định trước đó
  • Xác định hàm xây dựng __init__() bằng cách sử dụng các đối số của hàm để xây dựng một phiên bản của lớp ElementSpec và gọi siêu hàm với giá trị đó, cùng với một tên tùy chọn

Khi một phiên bản của thành phần được tạo, logic kiểm tra kiểu trong lớp base_component.BaseComponent sẽ được gọi để đảm bảo rằng các đối số được truyền vào tương thích với thông tin kiểu được xác định trong lớp 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)

Lắp ráp vào một đường ống TFX

Bước cuối cùng là cắm thành phần tùy chỉnh mới vào đường dẫn TFX. Bên cạnh việc thêm một phiên bản của thành phần mới, cũng cần có những điều sau:

  • Đấu dây đúng cách các thành phần ngược dòng và hạ lưu của thành phần mới vào nó. Điều này được thực hiện bằng cách tham chiếu các đầu ra của thành phần thượng nguồn trong thành phần mới và tham chiếu các đầu ra của thành phần mới trong các thành phần hạ nguồn
  • Thêm phiên bản thành phần mới vào danh sách thành phần khi xây dựng quy trình.

Ví dụ dưới đây nêu bật những thay đổi nói trên. Ví dụ đầy đủ có thể được tìm thấy trong 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, ...],
      ...
  )

Triển khai một thành phần hoàn toàn tùy chỉnh

Bên cạnh những thay đổi về mã, tất cả các phần mới được thêm vào ( ComponentSpec , Executor , giao diện thành phần) cần phải có thể truy cập được trong môi trường chạy đường ống để chạy đường ống đúng cách.