Componentes de função Python personalizados

A definição de componente baseada em função Python facilita a criação de componentes personalizados do TFX, economizando o esforço de definir uma classe de especificação de componente, uma classe de executor e uma classe de interface de componente. Neste estilo de definição de componente, você escreve uma função anotada com dicas de tipo. As dicas de tipo descrevem os artefatos de entrada, os artefatos de saída e os parâmetros do seu componente.

Escrever seu componente personalizado neste estilo é muito simples, como no exemplo a seguir.

class MyOutput(TypedDict):
  accuracy: float

@component
def MyValidationComponent(
    model: InputArtifact[Model],
    blessing: OutputArtifact[Model],
    accuracy_threshold: Parameter[int] = 10,
) -> MyOutput:
  '''My simple custom model validation component.'''

  accuracy = evaluate_model(model)
  if accuracy >= accuracy_threshold:
    write_output_blessing(blessing)

  return {
    'accuracy': accuracy
  }

Nos bastidores, isso define um componente personalizado que é uma subclasse de BaseComponent e suas classes Spec e Executor.

Se você deseja definir uma subclasse de BaseBeamComponent de modo que possa usar um pipeline de feixe com configuração compartilhada do pipeline TFX, ou seja, beam_pipeline_args ao compilar o pipeline ( Chicago Taxi Pipeline Exemplo ), você pode definir use_beam=True no decorador e adicionar outro BeamComponentParameter com valor padrão None em sua função como no exemplo a seguir:

@component(use_beam=True)
def MyDataProcessor(
    examples: InputArtifact[Example],
    processed_examples: OutputArtifact[Example],
    beam_pipeline: BeamComponentParameter[beam.Pipeline] = None,
    ) -> None:
  '''My simple custom model validation component.'''

  with beam_pipeline as p:
    # data pipeline definition with beam_pipeline begins
    ...
    # data pipeline definition with beam_pipeline ends

Se você é novo nos pipelines do TFX, saiba mais sobre os principais conceitos dos pipelines do TFX .

Entradas, saídas e parâmetros

No TFX, as entradas e saídas são rastreadas como objetos de artefato que descrevem a localização e as propriedades de metadados associadas aos dados subjacentes; essas informações são armazenadas em metadados de ML. Os artefatos podem descrever tipos de dados complexos ou tipos de dados simples, como: int, float, bytes ou strings unicode.

Um parâmetro é um argumento (int, float, bytes ou string unicode) para um componente conhecido no momento da construção do pipeline. Os parâmetros são úteis para especificar argumentos e hiperparâmetros, como contagem de iterações de treinamento, taxa de abandono e outras configurações para seu componente. Os parâmetros são armazenados como propriedades de execuções de componentes quando rastreados em metadados de ML.

Definição

Para criar um componente personalizado, escreva uma função que implemente sua lógica personalizada e decore-a com o decorador @component do módulo tfx.dsl.component.experimental.decorators . Para definir o esquema de entrada e saída do seu componente, anote os argumentos da sua função e o valor de retorno usando anotações do módulo tfx.dsl.component.experimental.annotations :

  • Para cada entrada de artefato , aplique a anotação de dica de tipo InputArtifact[ArtifactType] . Substitua ArtifactType pelo tipo do artefato, que é uma subclasse de tfx.types.Artifact . Essas entradas podem ser argumentos opcionais.

  • Para cada artefato de saída , aplique a anotação de dica de tipo OutputArtifact[ArtifactType] . Substitua ArtifactType pelo tipo do artefato, que é uma subclasse de tfx.types.Artifact . Os artefatos de saída do componente devem ser passados ​​como argumentos de entrada da função, para que seu componente possa gravar saídas em um local gerenciado pelo sistema e configurar propriedades apropriadas de metadados do artefato. Este argumento pode ser opcional ou pode ser definido com um valor padrão.

  • Para cada parâmetro , use a anotação de dica de tipo Parameter[T] . Substitua T pelo tipo do parâmetro. Atualmente, oferecemos suporte apenas a tipos python primitivos: bool , int , float , str ou bytes .

  • Para beam pipeline , use a anotação de dica de tipo BeamComponentParameter[beam.Pipeline] . Defina o valor padrão como None . O valor None será substituído por um pipeline de feixe instanciado criado por _make_beam_pipeline() do BaseBeamExecutor

  • Para cada entrada de tipo de dados simples ( int , float , str ou bytes ) não conhecida no momento da construção do pipeline, use a dica de tipo T . Observe que na versão TFX 0.22, valores concretos não podem ser transmitidos no momento da construção do pipeline para esse tipo de entrada (em vez disso, use a anotação Parameter , conforme descrito na seção anterior). Este argumento pode ser opcional ou pode ser definido com um valor padrão. Se o seu componente tiver saídas de tipo de dados simples ( int , float , str ou bytes ), você poderá retornar essas saídas usando um TypedDict como uma anotação de tipo de retorno e retornando um objeto dict apropriado.

No corpo da sua função, os artefatos de entrada e saída são passados ​​como objetos tfx.types.Artifact ; você pode inspecionar seu .uri para obter sua localização gerenciada pelo sistema e ler/definir quaisquer propriedades. Parâmetros de entrada e entradas de tipo de dados simples são passados ​​como objetos do tipo especificado. As saídas de tipo de dados simples devem ser retornadas como um dicionário, onde as chaves são os nomes de saída apropriados e os valores são os valores de retorno desejados.

O componente de função concluído pode ter esta aparência:

from typing import TypedDict
import tfx.v1 as tfx
from tfx.dsl.component.experimental.decorators import component

class MyOutput(TypedDict):
  loss: float
  accuracy: float

@component
def MyTrainerComponent(
    training_data: tfx.dsl.components.InputArtifact[tfx.types.standard_artifacts.Examples],
    model: tfx.dsl.components.OutputArtifact[tfx.types.standard_artifacts.Model],
    dropout_hyperparameter: float,
    num_iterations: tfx.dsl.components.Parameter[int] = 10
) -> MyOutput:
  '''My simple trainer component.'''

  records = read_examples(training_data.uri)
  model_obj = train_model(records, num_iterations, dropout_hyperparameter)
  model_obj.write_to(model.uri)

  return {
    'loss': model_obj.loss,
    'accuracy': model_obj.accuracy
  }

# Example usage in a pipeline graph definition:
# ...
trainer = MyTrainerComponent(
    examples=example_gen.outputs['examples'],
    dropout_hyperparameter=other_component.outputs['dropout'],
    num_iterations=1000)
pusher = Pusher(model=trainer.outputs['model'])
# ...

O exemplo anterior define MyTrainerComponent como um componente personalizado baseado em função Python. Este componente consome um artefato examples como entrada e produz um artefato model como saída. O componente usa o artifact_instance.uri para ler ou gravar o artefato em seu local gerenciado pelo sistema. O componente usa um parâmetro de entrada num_iterations e um valor de tipo de dados simples dropout_hyperparameter , e o componente gera métricas de loss e accuracy como valores de saída de tipo de dados simples. O artefato model de saída é então usado pelo componente Pusher .