La journée communautaire ML est le 9 novembre ! Rejoignez - nous pour les mises à jour de tensorflow, JAX et plus En savoir plus

Construire des composants entièrement personnalisés

Ce guide décrit comment utiliser l'API TFX pour créer un composant entièrement personnalisé. Les composants entièrement personnalisés vous permettent de créer des composants en définissant la spécification de composant, l'exécuteur et les classes d'interface de composant. Cette approche vous permet de réutiliser et d'étendre un composant standard pour l'adapter à vos besoins.

Si vous êtes nouveau aux pipelines TFX, en savoir plus sur les concepts de base de pipelines TFX .

Exécuteur personnalisé ou composant personnalisé

Si seule une logique de traitement personnalisée est nécessaire alors que les entrées, les sorties et les propriétés d'exécution du composant sont les mêmes qu'un composant existant, un exécuteur personnalisé est suffisant. Un composant entièrement personnalisé est nécessaire lorsque l'une des entrées, sorties ou propriétés d'exécution est différente de tout composant TFX existant.

Comment créer un composant personnalisé ?

Le développement d'un composant entièrement personnalisé nécessite :

  • Un ensemble défini de spécifications d'artefact d'entrée et de sortie pour le nouveau composant. En particulier, les types des artefacts d'entrée doivent être cohérents avec les types d'artefacts de sortie des composants qui produisent les artefacts et les types des artefacts de sortie doivent être cohérents avec les types d'artefacts d'entrée des composants qui consomment les artefacts, le cas échéant.
  • Les paramètres d'exécution non-artefact qui sont nécessaires pour le nouveau composant.

Spécification de composant

La ComponentSpec classe définit le contrat composant en définissant les objets d' entrée et de sortie à un composant ainsi que les paramètres qui sont utilisés pour l'exécution du composant. Il comporte trois parties :

  • ENTREES: un dictionnaire de paramètres typés pour les artefacts d'entrée qui sont dans l'exécuteur composant. Normalement, les artefacts d'entrée sont les sorties des composants en amont et partagent donc le même type.
  • SORTIES: Un dictionnaire des paramètres typés pour les artefacts de sortie qui produit le composant.
  • PARAMETRES: Un dictionnaire de supplémentaires ExecutionParameter éléments qui seront transmis à l'exécuteur testamentaire composant. Ce sont des paramètres non-artefact que nous voulons définir de manière flexible dans le pipeline DSL et passer en exécution.

Voici un exemple de 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),
  }

Exécuteur

Ensuite, écrivez le code d'exécution pour le nouveau composant. En fait, une nouvelle sous - classe de base_executor.BaseExecutor doit être créé avec son Do fonction overriden. Dans la Do fonction, les arguments input_dict , output_dict et exec_properties qui sont passés sur la carte à INPUTS , OUTPUTS et PARAMETERS qui sont définis dans ComponentSpec respectivement. Pour exec_properties , la valeur peut être récupérée directement par une recherche dans le dictionnaire. Pour des artefacts dans input_dict et output_dict , il existe des fonctions pratiques disponibles dans artifact_utils classe qui peut être utilisé pour chercher uri exemple des artefacts ou artefact.

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)

Test unitaire d'un exécuteur personnalisé

Les tests unitaires pour l'exécuteur personnalisé peut être créé semblable à celui - ci .

Interface de composant

Maintenant que la partie la plus complexe est terminée, l'étape suivante consiste à assembler ces pièces dans une interface de composant, pour permettre au composant d'être utilisé dans un pipeline. Il y a plusieurs étapes :

  • Faire le composant Interface une sous - classe de base_component.BaseComponent
  • Affectation d' une variable de classe SPEC_CLASS avec la ComponentSpec classe qui a été défini précédemment
  • Affectation d' une variable de classe EXECUTOR_SPEC avec la classe Executor qui a été défini précédemment
  • Définir le __init__() la fonction de constructeur en utilisant les arguments de la fonction pour construire une instance de la classe ComponentSpec et invoquer la fonction de super avec cette valeur, avec un nom facultatif

Lorsqu'une instance du composant est créé, tapez la logique de vérification dans la base_component.BaseComponent classe sera invoquée pour faire en sorte que les arguments qui ont été adoptés en sont compatibles avec les informations de type défini dans la ComponentSpec classe.

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)

Assembler dans un pipeline TFX

La dernière étape consiste à brancher le nouveau composant personnalisé dans un pipeline TFX. Outre l'ajout d'une instance du nouveau composant, les éléments suivants sont également nécessaires :

  • Câblez correctement les composants amont et aval du nouveau composant. Cela se fait en référençant les sorties du composant en amont dans le nouveau composant et en référençant les sorties du nouveau composant dans les composants en aval
  • Ajoutez la nouvelle instance de composant à la liste des composants lors de la construction du pipeline.

L'exemple ci-dessous met en évidence les modifications susmentionnées. Exemple complet se trouve dans le 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, ...],
      ...
  )

Déployer un composant entièrement personnalisé

A côté des changements de code, toutes les parties nouvellement ajoutés ( ComponentSpec , Executor , interface de composant) doivent être accessibles dans un environnement en cours d' exécution du pipeline afin d'exécuter le pipeline correctement.