Componentes de funciones de Python personalizados

La definición de componentes basada en funciones de Python le facilita la creación de componentes personalizados TFX, al ahorrarle el esfuerzo de definir una clase de especificación de componente, una clase de ejecutor y una clase de interfaz de componente. En este estilo de definición de componente, se escribe una función que está anotada con sugerencias de tipo. Las sugerencias de tipo describen los artefactos de entrada, los artefactos de salida y los parámetros de su componente.

Escribir su componente personalizado en este estilo es muy sencillo, como en el siguiente ejemplo.

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
  }

En esencia, esto define un componente personalizado que es una subclase de BaseComponent y sus clases Spec y Executor.

Si desea definir una subclase de BaseBeamComponent de modo que pueda usar una tubería de haz con configuración compartida de tubería TFX, es decir, beam_pipeline_args al compilar la tubería ( ejemplo de tubería de taxi de Chicago ), puede configurar use_beam=True en el decorador y agregar otro BeamComponentParameter con el valor predeterminado None en su función como en el siguiente ejemplo:

@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

Si es nuevo en las canalizaciones TFX, obtenga más información sobre los conceptos centrales de las canalizaciones TFX .

Entradas, salidas y parámetros.

En TFX, las entradas y salidas se rastrean como objetos Artifact que describen la ubicación y las propiedades de metadatos asociadas con los datos subyacentes; esta información se almacena en metadatos de ML. Los artefactos pueden describir tipos de datos complejos o tipos de datos simples, como: int, float, bytes o cadenas Unicode.

Un parámetro es un argumento (int, float, bytes o cadena Unicode) de un componente conocido en el momento de la construcción de la canalización. Los parámetros son útiles para especificar argumentos e hiperparámetros como el recuento de iteraciones de entrenamiento, la tasa de abandonos y otras configuraciones de su componente. Los parámetros se almacenan como propiedades de las ejecuciones de componentes cuando se rastrean en metadatos de ML.

Definición

Para crear un componente personalizado, escriba una función que implemente su lógica personalizada y decórela con el decorador @component del módulo tfx.dsl.component.experimental.decorators . Para definir el esquema de entrada y salida de su componente, anote los argumentos de su función y el valor de retorno usando anotaciones del módulo tfx.dsl.component.experimental.annotations :

  • Para cada entrada de artefacto , aplique la anotación de sugerencia de tipo InputArtifact[ArtifactType] . Reemplace ArtifactType con el tipo de artefacto, que es una subclase de tfx.types.Artifact . Estas entradas pueden ser argumentos opcionales.

  • Para cada artefacto de salida , aplique la anotación de sugerencia de tipo OutputArtifact[ArtifactType] . Reemplace ArtifactType con el tipo de artefacto, que es una subclase de tfx.types.Artifact . Los artefactos de salida de los componentes deben pasarse como argumentos de entrada de la función, de modo que su componente pueda escribir salidas en una ubicación administrada por el sistema y establecer las propiedades de metadatos de los artefactos apropiadas. Este argumento puede ser opcional o puede definirse con un valor predeterminado.

  • Para cada parámetro , utilice la anotación de sugerencia de tipo Parameter[T] . Reemplace T con el tipo de parámetro. Actualmente solo admitimos tipos primitivos de Python: bool , int , float , str o bytes .

  • Para la canalización de vigas , utilice la anotación de sugerencia de tipo BeamComponentParameter[beam.Pipeline] . Establezca el valor predeterminado en None . El valor None será reemplazado por una tubería de viga instanciada creada por _make_beam_pipeline() de BaseBeamExecutor

  • Para cada entrada de tipo de datos simple ( int , float , str o bytes ) no conocida en el momento de la construcción de la canalización, utilice la sugerencia de tipo T Tenga en cuenta que en la versión TFX 0.22, no se pueden pasar valores concretos en el momento de la construcción de la tubería para este tipo de entrada (use la anotación Parameter en su lugar, como se describe en la sección anterior). Este argumento puede ser opcional o puede definirse con un valor predeterminado. Si su componente tiene salidas de tipos de datos simples ( int , float , str o bytes ), puede devolver estas salidas utilizando TypedDict como anotación de tipo de retorno y devolviendo un objeto dict apropiado.

En el cuerpo de su función, los artefactos de entrada y salida se pasan como objetos tfx.types.Artifact ; puede inspeccionar su .uri para obtener su ubicación administrada por el sistema y leer/establecer cualquier propiedad. Los parámetros de entrada y las entradas de tipos de datos simples se pasan como objetos del tipo especificado. Las salidas de tipos de datos simples deben devolverse como un diccionario, donde las claves son los nombres de salida apropiados y los valores son los valores de retorno deseados.

El componente de función completo puede verse así:

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'])
# ...

El ejemplo anterior define MyTrainerComponent como un componente personalizado basado en funciones de Python. Este componente consume un artefacto examples como entrada y produce un artefacto model como salida. El componente utiliza artifact_instance.uri para leer o escribir el artefacto en su ubicación administrada por el sistema. El componente toma un parámetro de entrada num_iterations y un valor de tipo de datos simple dropout_hyperparameter , y el componente genera métricas loss y accuracy como valores de salida de tipo de datos simples. Luego, el componente Pusher utiliza el artefacto model de salida.