Tutorial del componente de función TFX Python

Este cuaderno contiene ejemplos sobre cómo crear y ejecutar componentes de funciones de Python dentro de TFX InteractiveContext y en una canalización TFX orquestada localmente.

Para obtener más información de contexto y, ver el pitón personalizada componentes de función página en el sitio de la documentación TFX.

Configuración

Primero instalaremos TFX e importaremos los módulos necesarios. TFX requiere Python 3.

Verifique la versión de Python del sistema

import sys
sys.version
'3.7.5 (default, Feb 23 2021, 13:22:40) \n[GCC 8.4.0]'

Actualizar Pip

Para evitar actualizar Pip en un sistema cuando se ejecuta localmente, verifique que estemos ejecutando en Colab. Por supuesto, los sistemas locales se pueden actualizar por separado.

try:
  import colab
  !pip install --upgrade pip
except:
  pass

Instalar TFX

pip install -U tfx

¿Reinició el tiempo de ejecución?

Si está utilizando Google Colab, la primera vez que ejecuta la celda anterior, debe reiniciar el tiempo de ejecución (Tiempo de ejecución> Reiniciar tiempo de ejecución ...). Esto se debe a la forma en que Colab carga los paquetes.

Importar paquetes

Importamos TFX y comprobamos su versión.

# Check version
from tfx import v1 as tfx
tfx.__version__
'1.4.0'

Componentes personalizados de la función de Python

En esta sección, crearemos componentes a partir de funciones de Python. No haremos ningún problema de ML real; estas funciones simples solo se utilizan para ilustrar el proceso de desarrollo de componentes de funciones de Python.

Ver función de Python guía componente a base de más documentación.

Crear componentes personalizados de Python

Comenzamos escribiendo una función que genera algunos datos ficticios. Esto se escribe en su propio archivo de módulo de Python.

%%writefile my_generator.py

import os
import tensorflow as tf  # Used for writing files.

from tfx import v1 as tfx

# Non-public APIs, just for showcase.
from tfx.types.experimental.simple_artifacts import Dataset

@tfx.dsl.components.component
def MyGenerator(data: tfx.dsl.components.OutputArtifact[Dataset]):
  """Create a file with dummy data in the output artifact."""
  with tf.io.gfile.GFile(os.path.join(data.uri, 'data_file.txt'), 'w') as f:
    f.write('Dummy data')

  # Set metadata and ensure that it gets passed to downstream components.
  data.set_string_custom_property('my_custom_field', 'my_custom_value')
Writing my_generator.py

A continuación, escribimos un segundo componente que utiliza los datos ficticios producidos. Simplemente calcularemos el hash de los datos y lo devolveremos.

%%writefile my_consumer.py

import hashlib
import os
import tensorflow as tf

from tfx import v1 as tfx

# Non-public APIs, just for showcase.
from tfx.types.experimental.simple_artifacts import Dataset
from tfx.types.standard_artifacts import String

@tfx.dsl.components.component
def MyConsumer(data: tfx.dsl.components.InputArtifact[Dataset],
               hash: tfx.dsl.components.OutputArtifact[String],
               algorithm: tfx.dsl.components.Parameter[str] = 'sha256'):
  """Reads the contents of data and calculate."""
  with tf.io.gfile.GFile(
      os.path.join(data.uri, 'data_file.txt'), 'r') as f:
    contents = f.read()
  h = hashlib.new(algorithm)
  h.update(tf.compat.as_bytes(contents))
  hash.value = h.hexdigest()

  # Read a custom property from the input artifact and set to the output.
  custom_value = data.get_string_custom_property('my_custom_field')
  hash.set_string_custom_property('input_custom_field', custom_value)
Writing my_consumer.py

Ejecutar en el cuaderno con InteractiveContext

Ahora, demostraremos el uso de nuestros nuevos componentes en TFX InteractiveContext.

Para obtener más información sobre lo que puede hacer con el bloc de notas TFX InteractiveContext, ver el bloc de notas en TFX Keras Componente Tutorial .

from my_generator import MyGenerator
from my_consumer import MyConsumer

Construya el InteractiveContext

# Here, we create an InteractiveContext using default parameters. This will
# use a temporary directory with an ephemeral ML Metadata database instance.
# To use your own pipeline root or database, the optional properties
# `pipeline_root` and `metadata_connection_config` may be passed to
# InteractiveContext. Calls to InteractiveContext are no-ops outside of the
# notebook.
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext
context = InteractiveContext()
WARNING:absl:InteractiveContext pipeline_root argument not provided: using temporary directory /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m as root for pipeline outputs.
WARNING:absl:InteractiveContext metadata_connection_config not provided: using SQLite ML Metadata database at /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m/metadata.sqlite.

Ejecutar el componente de forma interactiva con context.run()

A continuación, se ejecuta de forma interactiva los componentes dentro del cuaderno con context.run() . Nuestro componente consumidor utiliza las salidas del componente generador.

generator = MyGenerator()
context.run(generator)
WARNING: Logging before InitGoogleLogging() is written to STDERR
I1205 10:37:04.765872 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
consumer = MyConsumer(
    data=generator.outputs['data'],
    algorithm='md5')
context.run(consumer)
I1205 10:37:04.808555 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type

Después de la ejecución, podemos inspeccionar el contenido del artefacto de salida "hash" del componente consumidor en el disco.

tail -v {consumer.outputs['hash'].get()[0].uri}
==> /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m/MyConsumer/hash/2/value <==
0015fe7975d1a2794b59aa12635703f1

Eso es todo, ¡y ahora ha escrito y ejecutado sus propios componentes personalizados!

Escribe una definición de canalización

A continuación, crearemos una canalización utilizando estos mismos componentes. Durante el uso de la InteractiveContext dentro de un portátil funciona bien para la experimentación, la definición de una tubería le permite desplegar su tubería de corredores locales o remotos para su uso en producción.

Aquí, demostraremos el uso de LocalDagRunner que se ejecuta localmente en su máquina. Para la ejecución de la producción, los canales Airflow o Kubeflow pueden ser más adecuados.

Construye una tubería

import os
import tempfile
from tfx import v1 as tfx

# Select a persistent TFX root directory to store your output artifacts.
# For demonstration purposes only, we use a temporary directory.
PIPELINE_ROOT = tempfile.mkdtemp()
# Select a pipeline name so that multiple runs of the same logical pipeline
# can be grouped.
PIPELINE_NAME = "function-based-pipeline"
# We use a ML Metadata configuration that uses a local SQLite database in
# the pipeline root directory. Other backends for ML Metadata are available
# for production usage.
METADATA_CONNECTION_CONFIG = tfx.orchestration.metadata.sqlite_metadata_connection_config(
    os.path.join(PIPELINE_ROOT, 'metadata.sqlite'))

def function_based_pipeline():
  # Here, we construct our generator and consumer components in the same way.
  generator = MyGenerator()
  consumer = MyConsumer(
      data=generator.outputs['data'],
      algorithm='md5')

  return tfx.dsl.Pipeline(
      pipeline_name=PIPELINE_NAME,
      pipeline_root=PIPELINE_ROOT,
      components=[generator, consumer],
      metadata_connection_config=METADATA_CONNECTION_CONFIG)

my_pipeline = function_based_pipeline()

Ejecutar su tubería con la LocalDagRunner

tfx.orchestration.LocalDagRunner().run(my_pipeline)
I1205 10:37:04.983860 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:04.990442 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:04.996665 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:05.003470 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:05.013659 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:05.031374 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:05.048280 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type
I1205 10:37:05.067972 28682 rdbms_metadata_access_object.cc:686] No property is defined for the Type

Podemos inspeccionar los artefactos de salida generados por esta ejecución de canalización.

find {PIPELINE_ROOT}
/tmp/tmpydmun02b
/tmp/tmpydmun02b/metadata.sqlite
/tmp/tmpydmun02b/MyConsumer
/tmp/tmpydmun02b/MyConsumer/.system
/tmp/tmpydmun02b/MyConsumer/.system/executor_execution
/tmp/tmpydmun02b/MyConsumer/.system/executor_execution/2
/tmp/tmpydmun02b/MyConsumer/hash
/tmp/tmpydmun02b/MyConsumer/hash/2
/tmp/tmpydmun02b/MyConsumer/hash/2/value
/tmp/tmpydmun02b/MyGenerator
/tmp/tmpydmun02b/MyGenerator/data
/tmp/tmpydmun02b/MyGenerator/data/1
/tmp/tmpydmun02b/MyGenerator/data/1/data_file.txt
/tmp/tmpydmun02b/MyGenerator/.system
/tmp/tmpydmun02b/MyGenerator/.system/executor_execution
/tmp/tmpydmun02b/MyGenerator/.system/executor_execution/1

¡Ya ha escrito sus propios componentes personalizados y ha orquestado su ejecución en LocalDagRunner! Para los pasos siguientes, echa un vistazo a los tutoriales y guías adicionales en el sitio web de TFX .