Простой конвейер TFX для конвейеров вершин

Посмотреть на TensorFlow.org Запустить в Google Colab Посмотреть исходный код на GitHub Скачать блокнот Запуск в Google Cloud Vertex AI Workbench

В этом учебном пособии на основе записной книжки будет создан простой конвейер TFX и запущен с помощью Google Cloud Vertex Pipelines. Эта записная книжка основана на конвейере TFX, который мы создали в учебнике Simple TFX Pipeline Tutorial . Если вы не знакомы с TFX и еще не читали этот учебник, вам следует прочитать его, прежде чем приступить к работе с этой записной книжкой.

Google Cloud Vertex Pipelines помогает автоматизировать, отслеживать и управлять системами машинного обучения, организуя рабочий процесс машинного обучения без сервера. Вы можете определить свои конвейеры машинного обучения с помощью Python с TFX, а затем выполнить свои конвейеры в Google Cloud. См. Введение в Vertex Pipelines , чтобы узнать больше о Vertex Pipelines.

Эта записная книжка предназначена для работы в Google Colab или записных книжках на платформе AI . Если вы не используете один из них, вы можете просто нажать кнопку «Запустить в Google Colab» выше.

Настраивать

Перед запуском этой записной книжки убедитесь, что у вас есть следующее:

См. документацию Vertex для дальнейшей настройки проекта GCP.

Установить пакеты питона

Мы установим необходимые пакеты Python, включая TFX и KFP, для создания конвейеров ML и отправки заданий в Vertex Pipelines.

# Use the latest version of pip.
pip install --upgrade pip
pip install --upgrade "tfx[kfp]<2"

Вы перезапустили среду выполнения?

Если вы используете Google Colab, при первом запуске указанной выше ячейки необходимо перезапустить среду выполнения, нажав кнопку «RESTART RUNTIME» выше или используя меню «Runtime > Restart runtime...». Это связано с тем, как Colab загружает пакеты.

Если вы не используете Colab, вы можете перезапустить среду выполнения со следующей ячейкой.

# docs_infra: no_execute
import sys
if not 'google.colab' in sys.modules:
  # Automatically restart kernel after installs
  import IPython
  app = IPython.Application.instance()
  app.kernel.do_shutdown(True)

Войдите в Google для этого блокнота

Если вы используете этот блокнот в Colab, выполните аутентификацию с помощью своей учетной записи пользователя:

import sys
if 'google.colab' in sys.modules:
  from google.colab import auth
  auth.authenticate_user()

Если вы используете ноутбуки на платформе AI , выполните аутентификацию в Google Cloud перед запуском следующего раздела, запустив

gcloud auth login

в окне терминала (которое можно открыть через меню « Файл» > « Создать »). Это нужно сделать только один раз для каждого экземпляра записной книжки.

Проверьте версии пакетов.

import tensorflow as tf
print('TensorFlow version: {}'.format(tf.__version__))
from tfx import v1 as tfx
print('TFX version: {}'.format(tfx.__version__))
import kfp
print('KFP version: {}'.format(kfp.__version__))

TensorFlow version: 2.7.1
TFX version: 1.6.0
KFP version: 1.8.11

Настройка переменных

Ниже мы настроим некоторые переменные, используемые для настройки конвейеров. Требуется следующая информация:

Введите требуемые значения в ячейку ниже перед запуском .

GOOGLE_CLOUD_PROJECT = ''     # <--- ENTER THIS
GOOGLE_CLOUD_REGION = ''      # <--- ENTER THIS
GCS_BUCKET_NAME = ''          # <--- ENTER THIS

if not (GOOGLE_CLOUD_PROJECT and GOOGLE_CLOUD_REGION and GCS_BUCKET_NAME):
    from absl import logging
    logging.error('Please set all required parameters.')

ERROR:absl:Please set all required parameters.

Настройте gcloud для использования вашего проекта.

gcloud config set project {GOOGLE_CLOUD_PROJECT}

ERROR: (gcloud.config.set) argument VALUE: Must be specified.
Usage: gcloud config set SECTION/PROPERTY VALUE [optional flags]
  optional flags may be  --help | --installation

For detailed information on this command and its flags, run:
  gcloud config set --help

PIPELINE_NAME = 'penguin-vertex-pipelines'

# Path to various pipeline artifact.
PIPELINE_ROOT = 'gs://{}/pipeline_root/{}'.format(
    GCS_BUCKET_NAME, PIPELINE_NAME)

# Paths for users' Python module.
MODULE_ROOT = 'gs://{}/pipeline_module/{}'.format(
    GCS_BUCKET_NAME, PIPELINE_NAME)

# Paths for input data.
DATA_ROOT = 'gs://{}/data/{}'.format(GCS_BUCKET_NAME, PIPELINE_NAME)

# This is the path where your model will be pushed for serving.
SERVING_MODEL_DIR = 'gs://{}/serving_model/{}'.format(
    GCS_BUCKET_NAME, PIPELINE_NAME)

print('PIPELINE_ROOT: {}'.format(PIPELINE_ROOT))

PIPELINE_ROOT: gs:///pipeline_root/penguin-vertex-pipelines

Подготовьте пример данных

Мы будем использовать тот же набор данных Palmer Penguins , что и Simple TFX Pipeline Tutorial .

В этом наборе данных есть четыре числовых признака, которые уже были нормализованы, чтобы иметь диапазон [0,1]. Мы построим классификационную модель, которая предсказывает species пингвинов.

Нам нужно сделать собственную копию набора данных. Поскольку TFX ExampleGen считывает входные данные из каталога, нам нужно создать каталог и скопировать в него набор данных в GCS.

gsutil cp gs://download.tensorflow.org/data/palmer_penguins/penguins_processed.csv {DATA_ROOT}/

InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///data/penguin-vertex-pipelines/".

Взгляните на файл CSV.

gsutil cat {DATA_ROOT}/penguins_processed.csv | head

InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///data/penguin-vertex-pipelines/penguins_processed.csv".

Создать конвейер

Конвейеры TFX определяются с помощью API Python. Мы определим конвейер, который состоит из трех компонентов: CsvExampleGen, Trainer и Pusher. Определение пайплайна и модели почти такое же, как и в Simple TFX Pipeline Tutorial .

Единственное отличие состоит в том, что нам не нужно устанавливать metadata_connection_config , который используется для поиска базы данных метаданных ML . Поскольку Vertex Pipelines использует службу управляемых метаданных, пользователям не нужно заботиться об этом, и нам не нужно указывать параметр.

Прежде чем определить конвейер, нам нужно сначала написать код модели для компонента Trainer.

Напишите код модели.

Мы будем использовать тот же код модели, что и в Simple TFX Pipeline Tutorial .

_trainer_module_file = 'penguin_trainer.py'

%%writefile {_trainer_module_file}

# Copied from https://www.tensorflow.org/tfx/tutorials/tfx/penguin_simple

from typing import List
from absl import logging
import tensorflow as tf
from tensorflow import keras
from tensorflow_transform.tf_metadata import schema_utils


from tfx import v1 as tfx
from tfx_bsl.public import tfxio

from tensorflow_metadata.proto.v0 import schema_pb2

_FEATURE_KEYS = [
    'culmen_length_mm', 'culmen_depth_mm', 'flipper_length_mm', 'body_mass_g'
]
_LABEL_KEY = 'species'

_TRAIN_BATCH_SIZE = 20
_EVAL_BATCH_SIZE = 10

# Since we're not generating or creating a schema, we will instead create
# a feature spec.  Since there are a fairly small number of features this is
# manageable for this dataset.
_FEATURE_SPEC = {
    **{
        feature: tf.io.FixedLenFeature(shape=[1], dtype=tf.float32)
           for feature in _FEATURE_KEYS
       },
    _LABEL_KEY: tf.io.FixedLenFeature(shape=[1], dtype=tf.int64)
}


def _input_fn(file_pattern: List[str],
              data_accessor: tfx.components.DataAccessor,
              schema: schema_pb2.Schema,
              batch_size: int) -> tf.data.Dataset:
  """Generates features and label for training.

  Args:
    file_pattern: List of paths or patterns of input tfrecord files.
    data_accessor: DataAccessor for converting input to RecordBatch.
    schema: schema of the input data.
    batch_size: representing the number of consecutive elements of returned
      dataset to combine in a single batch

  Returns:
    A dataset that contains (features, indices) tuple where features is a
      dictionary of Tensors, and indices is a single Tensor of label indices.
  """
  return data_accessor.tf_dataset_factory(
      file_pattern,
      tfxio.TensorFlowDatasetOptions(
          batch_size=batch_size, label_key=_LABEL_KEY),
      schema=schema).repeat()


def _make_keras_model() -> tf.keras.Model:
  """Creates a DNN Keras model for classifying penguin data.

  Returns:
    A Keras Model.
  """
  # The model below is built with Functional API, please refer to
  # https://www.tensorflow.org/guide/keras/overview for all API options.
  inputs = [keras.layers.Input(shape=(1,), name=f) for f in _FEATURE_KEYS]
  d = keras.layers.concatenate(inputs)
  for _ in range(2):
    d = keras.layers.Dense(8, activation='relu')(d)
  outputs = keras.layers.Dense(3)(d)

  model = keras.Model(inputs=inputs, outputs=outputs)
  model.compile(
      optimizer=keras.optimizers.Adam(1e-2),
      loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
      metrics=[keras.metrics.SparseCategoricalAccuracy()])

  model.summary(print_fn=logging.info)
  return model


# TFX Trainer will call this function.
def run_fn(fn_args: tfx.components.FnArgs):
  """Train the model based on given args.

  Args:
    fn_args: Holds args used to train the model as name/value pairs.
  """

  # This schema is usually either an output of SchemaGen or a manually-curated
  # version provided by pipeline author. A schema can also derived from TFT
  # graph if a Transform component is used. In the case when either is missing,
  # `schema_from_feature_spec` could be used to generate schema from very simple
  # feature_spec, but the schema returned would be very primitive.
  schema = schema_utils.schema_from_feature_spec(_FEATURE_SPEC)

  train_dataset = _input_fn(
      fn_args.train_files,
      fn_args.data_accessor,
      schema,
      batch_size=_TRAIN_BATCH_SIZE)
  eval_dataset = _input_fn(
      fn_args.eval_files,
      fn_args.data_accessor,
      schema,
      batch_size=_EVAL_BATCH_SIZE)

  model = _make_keras_model()
  model.fit(
      train_dataset,
      steps_per_epoch=fn_args.train_steps,
      validation_data=eval_dataset,
      validation_steps=fn_args.eval_steps)

  # The result of the training should be saved in `fn_args.serving_model_dir`
  # directory.
  model.save(fn_args.serving_model_dir, save_format='tf')

Writing penguin_trainer.py

Скопируйте файл модуля в GCS, к которому можно получить доступ из компонентов конвейера. Поскольку обучение модели происходит в GCP, нам необходимо загрузить это определение модели.

В противном случае вы можете создать образ контейнера, включающий файл модуля, и использовать этот образ для запуска конвейера.

gsutil cp {_trainer_module_file} {MODULE_ROOT}/

InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///pipeline_module/penguin-vertex-pipelines/".

Напишите определение конвейера

Мы определим функцию для создания конвейера TFX.

# Copied from https://www.tensorflow.org/tfx/tutorials/tfx/penguin_simple and
# slightly modified because we don't need `metadata_path` argument.

def _create_pipeline(pipeline_name: str, pipeline_root: str, data_root: str,
                     module_file: str, serving_model_dir: str,
                     ) -> tfx.dsl.Pipeline:
  """Creates a three component penguin pipeline with TFX."""
  # Brings data into the pipeline.
  example_gen = tfx.components.CsvExampleGen(input_base=data_root)

  # Uses user-provided Python function that trains a model.
  trainer = tfx.components.Trainer(
      module_file=module_file,
      examples=example_gen.outputs['examples'],
      train_args=tfx.proto.TrainArgs(num_steps=100),
      eval_args=tfx.proto.EvalArgs(num_steps=5))

  # Pushes the model to a filesystem destination.
  pusher = tfx.components.Pusher(
      model=trainer.outputs['model'],
      push_destination=tfx.proto.PushDestination(
          filesystem=tfx.proto.PushDestination.Filesystem(
              base_directory=serving_model_dir)))

  # Following three components will be included in the pipeline.
  components = [
      example_gen,
      trainer,
      pusher,
  ]

  return tfx.dsl.Pipeline(
      pipeline_name=pipeline_name,
      pipeline_root=pipeline_root,
      components=components)

Запустите конвейер на Vertex Pipelines.

Мы использовали LocalDagRunner , который работает в локальной среде, в учебнике Simple TFX Pipeline Tutorial . TFX предоставляет несколько оркестраторов для запуска конвейера. В этом уроке мы будем использовать Vertex Pipelines вместе с бегунком dag Kubeflow V2.

Нам нужно определить бегун для фактического запуска конвейера. Вы скомпилируете свой конвейер в наш формат определения конвейера, используя TFX API.

import os

PIPELINE_DEFINITION_FILE = PIPELINE_NAME + '_pipeline.json'

runner = tfx.orchestration.experimental.KubeflowV2DagRunner(
    config=tfx.orchestration.experimental.KubeflowV2DagRunnerConfig(),
    output_filename=PIPELINE_DEFINITION_FILE)
# Following function will write the pipeline definition to PIPELINE_DEFINITION_FILE.
_ = runner.run(
    _create_pipeline(
        pipeline_name=PIPELINE_NAME,
        pipeline_root=PIPELINE_ROOT,
        data_root=DATA_ROOT,
        module_file=os.path.join(MODULE_ROOT, _trainer_module_file),
        serving_model_dir=SERVING_MODEL_DIR))

Сгенерированный файл определения можно отправить с помощью клиента kfp.

# docs_infra: no_execute
from google.cloud import aiplatform
from google.cloud.aiplatform import pipeline_jobs

aiplatform.init(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_REGION)

job = pipeline_jobs.PipelineJob(template_path=PIPELINE_DEFINITION_FILE,
                                display_name=PIPELINE_NAME)
job.run(sync=False)

Теперь вы можете посетить Vertex AI > Pipelines в Google Cloud Console , чтобы увидеть прогресс.