Tamamen Özel Bileşenler Oluşturma

Bu kılavuz, tamamen özel bir bileşen oluşturmak için TFX API'nin nasıl kullanılacağını açıklamaktadır. Tamamen özel bileşenler, bileşen spesifikasyonunu, yürütücüyü ve bileşen arayüzü sınıflarını tanımlayarak bileşenler oluşturmanıza olanak tanır. Bu yaklaşım, standart bir bileşeni ihtiyaçlarınıza uyacak şekilde yeniden kullanmanıza ve genişletmenize olanak tanır.

TFX işlem hatları konusunda yeniyseniz TFX işlem hatlarına ilişkin temel kavramlar hakkında daha fazla bilgi edinin .

Özel yürütücü veya özel bileşen

Bileşenin girişleri, çıkışları ve yürütme özellikleri mevcut bir bileşenle aynıyken yalnızca özel işleme mantığına ihtiyaç duyulursa özel bir yürütücü yeterlidir. Girişlerden, çıkışlardan veya yürütme özelliklerinden herhangi biri mevcut TFX bileşenlerinden farklı olduğunda tamamen özel bir bileşene ihtiyaç duyulur.

Özel bir bileşen nasıl oluşturulur?

Tamamen özel bir bileşen geliştirmek şunları gerektirir:

  • Yeni bileşen için tanımlanmış bir girdi ve çıktı yapay belirtimleri kümesi. Özellikle, giriş yapıtlarının türleri, yapıtları üreten bileşenlerin çıktı yapıt türleriyle tutarlı olmalı ve çıktı yapay yapılarının türleri, varsa yapıtları tüketen bileşenlerin giriş yapay yapı türleriyle tutarlı olmalıdır.
  • Yeni bileşen için gerekli olan yapay olmayan yürütme parametreleri.

Bileşen Özellikleri

ComponentSpec sınıfı, bir bileşene yönelik giriş ve çıkış yapıtlarının yanı sıra bileşenin yürütülmesi için kullanılan parametreleri tanımlayarak bileşen sözleşmesini tanımlar. Üç bölümü vardır:

  • GİRİŞLER : Bileşen yürütücüsüne iletilen giriş yapıtları için yazılan parametrelerin sözlüğü. Normalde giriş yapıtları, yukarı akış bileşenlerinden gelen çıkışlardır ve dolayısıyla aynı türü paylaşırlar.
  • ÇIKIŞLAR : Bileşenin ürettiği çıktı eserleri için yazılan parametrelerin sözlüğü.
  • PARAMETERS : Bileşen yürütücüsüne aktarılacak ek ExecutionParameter öğelerinin sözlüğü. Bunlar, DSL hattında esnek bir şekilde tanımlamak ve uygulamaya geçirmek istediğimiz yapay olmayan parametrelerdir.

İşte ComponentSpec'in bir örneği: 

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

İcracı

Daha sonra yeni bileşenin yürütücü kodunu yazın. Temel olarak, Do işlevi geçersiz kılınarak base_executor.BaseExecutor yeni bir alt sınıfının oluşturulması gerekir. Do işlevinde, sırasıyla ComponentSpec'te tanımlanan INPUTS , OUTPUTS ve PARAMETERS eşlenen input_dict , output_dict ve exec_properties argümanları. exec_properties için değer doğrudan bir sözlük araması yoluyla getirilebilir. input_dict ve output_dict içindeki yapılar için, yapıt örneğini veya yapıt uri'yi getirmek için artefakt_utils sınıfında kullanılabilen kullanışlı işlevler vardır. 

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)

Özel bir uygulayıcıyı test eden birim

Özel yürütücü için birim testleri buna benzer şekilde oluşturulabilir.

Bileşen arayüzü

Artık en karmaşık parça tamamlandığı için bir sonraki adım, bileşenin bir boru hattında kullanılmasını sağlamak için bu parçaları bir bileşen arayüzünde birleştirmektir. Birkaç adım var:

  • Bileşen arayüzünü base_component.BaseComponent bir alt sınıfı yapın
  • Daha önce tanımlanan ComponentSpec sınıfına SPEC_CLASS sınıf değişkeni atayın
  • Daha önce tanımlanan Executor sınıfına EXECUTOR_SPEC sınıf değişkeni atayın
  • ComponentSpec sınıfının bir örneğini oluşturmak için işlevin bağımsız değişkenlerini kullanarak __init__() yapıcı işlevini tanımlayın ve isteğe bağlı bir adla birlikte bu değerle süper işlevini çağırın

Bileşenin bir örneği oluşturulduğunda, aktarılan bağımsız değişkenlerin ComponentSpec sınıfında tanımlanan tür bilgileriyle uyumlu olduğundan emin olmak için base_component.BaseComponent sınıfındaki tür kontrol mantığı çağrılacaktır. 

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 ardışık düzenine birleştirme

Son adım, yeni özel bileşeni bir TFX işlem hattına takmaktır. Yeni bileşenin bir örneğini eklemenin yanı sıra aşağıdakiler de gereklidir:

  • Yeni bileşenin yukarı ve aşağı yöndeki bileşenlerini ona uygun şekilde bağlayın. Bu, yeni bileşendeki yukarı yöndeki bileşenin çıktılarına referans verilerek ve aşağı yöndeki bileşenlerde yeni bileşenin çıktılarına atıfta bulunularak yapılır.
  • İşlem hattını oluştururken yeni bileşen örneğini bileşenler listesine ekleyin.

Aşağıdaki örnek yukarıda belirtilen değişiklikleri vurgulamaktadır. Tam örnek TFX GitHub deposunda bulunabilir. 

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

Tamamen özel bir bileşen dağıtın

Kod değişikliklerinin yanı sıra, boru hattının düzgün bir şekilde çalıştırılması için yeni eklenen tüm parçalara ( ComponentSpec , Executor , bileşen arayüzü) boru hattı çalıştırma ortamında erişilebilir olması gerekir.