 Посмотреть на TensorFlow.org |  Запускаем в Google Colab |  Посмотреть исходный код на GitHub |  Скачать блокнот |
Эта записная книжка содержит примеры того, как создавать и запускать функциональные компоненты Python в TFX InteractiveContext и в локально управляемом конвейере TFX.
Для получения дополнительной информации и контекста см компонентов пользовательской функции Python страница на сайте TFX документации.
Настраивать
Сначала мы установим TFX и импортируем необходимые модули. TFX требует Python 3.
Проверьте системную версию Python
import sys
sys.version
'3.7.5 (default, Feb 23 2021, 13:22:40) \n[GCC 8.4.0]'
Обновить Pip
Чтобы избежать обновления Pip в системе при локальном запуске, убедитесь, что мы работаем в Colab. Конечно, локальные системы можно модернизировать отдельно.
try:
import colab
!pip install --upgrade pip
except:
pass
Установить TFX
pip install -U tfx
Вы перезапускали среду выполнения?
Если вы используете Google Colab, при первом запуске указанной выше ячейки вы должны перезапустить среду выполнения (Runtime> Restart runtime ...). Это связано с тем, как Colab загружает пакеты.
Импортировать пакеты
Импортируем TFX и проверяем его версию.
# Check version
from tfx import v1 as tfx
tfx.__version__
'1.4.0'
Пользовательские компоненты функций Python
В этом разделе мы создадим компоненты из функций Python. Мы не будем заниматься реальной проблемой машинного обучения - эти простые функции используются только для иллюстрации процесса разработки компонентов функций Python.
См функция Python на основе руководства компонента для получения дополнительной документации.
Создание пользовательских компонентов Python
Начнем с написания функции, которая генерирует фиктивные данные. Это записывается в собственный файл модуля 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
Затем мы пишем второй компонент, который использует созданные фиктивные данные. Мы просто вычислим хэш данных и вернем его.
%%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
Запуск в записной книжке с InteractiveContext
Теперь мы продемонстрируем использование наших новых компонентов в TFX InteractiveContext.
Для получения дополнительной информации о том , что вы можете сделать с TFX ноутбука InteractiveContext см в-ноутбук TFX Keras компонентов Учебник .
from my_generator import MyGenerator
from my_consumer import MyConsumer
Создайте 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.
Запустите компонент в интерактивном режиме с context.run()
Далее, мы проводим наши компоненты в интерактивном режиме в записной книжке с context.run() . Наш потребительский компонент использует выходные данные генератора.
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
После выполнения мы можем проверить содержимое выходного артефакта "хеша" потребительского компонента на диске.
tail -v {consumer.outputs['hash'].get()[0].uri}
==> /tmp/tfx-interactive-2021-12-05T10_37_04.715534-3q0k1y0m/MyConsumer/hash/2/value <== 0015fe7975d1a2794b59aa12635703f1
Вот и все, и теперь вы написали и выполнили свои собственные пользовательские компоненты!
Напишите определение конвейера
Затем мы создадим конвейер, используя те же компоненты. При использовании InteractiveContext в ноутбуке работает хорошо для экспериментов, определение трубопровода позволяет развертывать трубопровод на локальных или удаленные полозьях для использования продукции.
Здесь мы продемонстрируем использование LocalDagRunner, работающего локально на вашем компьютере. Для производственного исполнения могут быть более подходящими бегуны Airflow или Kubeflow.
Построить трубопровод
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()
Запустите конвейер с 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
Мы можем проверить выходные артефакты, генерируемые этим конвейером.
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
Теперь вы написали свои собственные пользовательские компоненты и организовали их выполнение на LocalDagRunner! Для последующих шагов, проверьте дополнительные учебники и руководства на веб - сайте TFX .