Создание конвейера TFX локально

TFX упрощает организацию рабочего процесса машинного обучения (ML) в виде конвейера, чтобы:

  • Автоматизируйте процесс машинного обучения, что позволит вам регулярно переобучать, оценивать и развертывать модель.
  • Создавайте конвейеры машинного обучения, которые включают глубокий анализ производительности моделей и проверку вновь обученных моделей для обеспечения производительности и надежности.
  • Отслеживайте данные обучения на предмет аномалий и устраняйте перекосы в обслуживании обучения.
  • Увеличьте скорость экспериментирования, запустив конвейер с разными наборами гиперпараметров.

Типичный процесс разработки конвейера начинается на локальном компьютере с анализа данных и настройки компонентов перед развертыванием в рабочей среде. В этом руководстве описываются два способа локального построения конвейера.

  • Настройте шаблон конвейера TFX в соответствии с потребностями вашего рабочего процесса машинного обучения. Шаблоны конвейера TFX — это готовые рабочие процессы, демонстрирующие лучшие практики использования стандартных компонентов TFX.
  • Создайте конвейер, используя TFX. В этом случае вы определяете конвейер, не начиная с шаблона.

По мере разработки конвейера вы можете запускать его с помощью LocalDagRunner . Затем, как только компоненты конвейера будут четко определены и протестированы, вы сможете использовать оркестратор промышленного уровня, такой как Kubeflow или Airflow.

Прежде чем начать

TFX — это пакет Python, поэтому вам потребуется настроить среду разработки Python, например виртуальную среду или контейнер Docker. Затем:

pip install tfx

Если вы новичок в конвейерах TFX, прежде чем продолжить , узнайте больше об основных концепциях конвейеров TFX .

Создание конвейера с использованием шаблона

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

В следующих разделах описывается, как создать копию шаблона и настроить ее в соответствии со своими потребностями.

Создайте копию шаблона конвейера.

  1. См. список доступных шаблонов конвейера TFX:

    tfx template list

  2. Выберите шаблон из списка

    tfx template copy --model=template --pipeline_name=pipeline-name \
--destination_path=destination-path

    Замените следующее:

    • template : имя шаблона, который вы хотите скопировать.
    • pipeline-name : имя создаваемого конвейера.
    • destination-path : путь для копирования шаблона.

    Узнайте больше о команде tfx template copy .

  3. Копия шаблона конвейера создана по указанному вами пути.

Изучите шаблон конвейера

В этом разделе представлен обзор строительных лесов, созданных с помощью шаблона.

  1. Изучите каталоги и файлы, которые были скопированы в корневой каталог вашего конвейера.

    • Каталог конвейера с
      • pipeline.py — определяет конвейер и перечисляет, какие компоненты используются.
      • configs.py — содержит сведения о конфигурации, например, откуда поступают данные или какой оркестратор используется.
    • Каталог данных
      • Обычно он содержит файл data.csv , который является источником по умолчанию для ExampleGen . Вы можете изменить источник данных в configs.py .

    • Каталог моделей с кодом предварительной обработки и реализациями моделей.

    • Шаблон копирует бегунов DAG для локальной среды и Kubeflow.

    • Некоторые шаблоны также включают блокноты Python, чтобы вы могли исследовать свои данные и артефакты с помощью метаданных машинного обучения.

  2. Выполните следующие команды в каталоге вашего конвейера: 

    tfx pipeline create --pipeline_path local_runner.py

    tfx run create --pipeline_name pipeline_name

    Команда создает запуск конвейера с использованием LocalDagRunner , который добавляет в ваш конвейер следующие каталоги:

    • Каталог tfx_metadata , содержащий хранилище метаданных ML, используемое локально.
    • Каталог tfx_pipeline_output , содержащий выходные файлы конвейера.

  3. Откройте файл pipeline/configs.py вашего конвейера и просмотрите его содержимое. Этот сценарий определяет параметры конфигурации, используемые конвейером и функциями компонента. Здесь вы можете указать такие вещи, как расположение источника данных или количество шагов обучения в выполнении.

  4. Откройте файл pipeline/pipeline.py вашего конвейера и просмотрите его содержимое. Этот скрипт создает конвейер TFX. Изначально конвейер содержит только компонент ExampleGen .

    • Следуйте инструкциям в комментариях TODO в pipeline.py , чтобы добавить дополнительные шаги в конвейер.

  5. Откройте файл local_runner.py и просмотрите его содержимое. Этот сценарий создает запуск конвейера и указывает параметры запуска, такие как data_path и preprocessing_fn .

  6. Вы просмотрели структуру, созданную с помощью шаблона, и создали конвейер, запускаемый с помощью LocalDagRunner . Далее настройте шаблон в соответствии со своими требованиями.

Настройте свой конвейер

В этом разделе представлен обзор того, как приступить к настройке шаблона.

  1. Спроектируйте свой трубопровод. Формы, предоставляемые шаблоном, помогают реализовать конвейер для табличных данных с использованием стандартных компонентов TFX. Если вы перемещаете существующий рабочий процесс машинного обучения в конвейер, вам может потребоваться пересмотреть свой код, чтобы в полной мере использовать стандартные компоненты TFX . Вам также может потребоваться создать пользовательские компоненты , реализующие функции, уникальные для вашего рабочего процесса или еще не поддерживаемые стандартными компонентами TFX.

  2. После того как вы спроектировали свой конвейер, итеративно настройте его, используя следующий процесс. Начните с компонента, который принимает данные в ваш конвейер, обычно это компонент ExampleGen .

    1. Настройте конвейер или компонент в соответствии со своим вариантом использования. Эти настройки могут включать в себя такие изменения, как:

      • Изменение параметров трубопровода.
      • Добавление компонентов в конвейер или их удаление.
      • Замена источника ввода данных. Этим источником данных может быть файл или запросы к таким сервисам, как BigQuery.
      • Изменение конфигурации компонента в конвейере.
      • Изменение функции настройки компонента.

    2. Запустите компонент локально с помощью сценария local_runner.py или другого подходящего средства запуска DAG, если вы используете другой оркестратор. Если сценарий завершается сбоем, отладьте ошибку и повторите запуск сценария.

    3. Как только эта настройка заработает, переходите к следующей настройке.

  3. Работая итеративно, вы можете настроить каждый шаг рабочего процесса шаблона в соответствии со своими потребностями.

Создайте собственный конвейер

Используйте следующие инструкции, чтобы узнать больше о создании пользовательского конвейера без использования шаблона.

  1. Спроектируйте свой трубопровод. Стандартные компоненты TFX обеспечивают проверенную функциональность, которая поможет вам реализовать полный рабочий процесс машинного обучения. Если вы перемещаете существующий рабочий процесс машинного обучения в конвейер, вам может потребоваться пересмотреть свой код, чтобы в полной мере использовать стандартные компоненты TFX. Вам также может потребоваться создать собственные компоненты , реализующие такие функции, как увеличение данных.

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

    import os
from typing import Optional, Text, List
from absl import logging
from ml_metadata.proto import metadata_store_pb2
import tfx.v1 as tfx

PIPELINE_NAME = 'my_pipeline'
PIPELINE_ROOT = os.path.join('.', 'my_pipeline_output')
METADATA_PATH = os.path.join('.', 'tfx_metadata', PIPELINE_NAME, 'metadata.db')
ENABLE_CACHE = True

def create_pipeline(
  pipeline_name: Text,
  pipeline_root:Text,
  enable_cache: bool,
  metadata_connection_config: Optional[
    metadata_store_pb2.ConnectionConfig] = None,
  beam_pipeline_args: Optional[List[Text]] = None
):
  components = []

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

def run_pipeline():
  my_pipeline = create_pipeline(
      pipeline_name=PIPELINE_NAME,
      pipeline_root=PIPELINE_ROOT,
      enable_cache=ENABLE_CACHE,
      metadata_connection_config=tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)
      )

  tfx.orchestration.LocalDagRunner().run(my_pipeline)

if __name__ == '__main__':
  logging.set_verbosity(logging.INFO)
  run_pipeline()

    На следующих шагах вы определите свой конвейер в create_pipeline и запустите его локально с помощью локального средства запуска.

    Итеративно создайте свой конвейер, используя следующий процесс.

    1. Настройте конвейер или компонент в соответствии со своим вариантом использования. Эти настройки могут включать в себя такие изменения, как:

      • Изменение параметров трубопровода.
      • Добавление компонентов в конвейер или их удаление.
      • Замена файла ввода данных.
      • Изменение конфигурации компонента в конвейере.
      • Изменение функции настройки компонента.

    2. Запустите компонент локально с помощью локального средства запуска или напрямую запустив скрипт. Если сценарий завершается сбоем, отладьте ошибку и повторите запуск сценария.

    3. Как только эта настройка заработает, переходите к следующей настройке.

    Начните с первого узла рабочего процесса вашего конвейера, обычно первый узел принимает данные в ваш конвейер.

  3. Добавьте первый узел рабочего процесса в свой конвейер. В этом примере конвейер использует стандартный компонент ExampleGen для загрузки CSV-файла из каталога ./data .

    from tfx.components import CsvExampleGen

DATA_PATH = os.path.join('.', 'data')

def create_pipeline(
  pipeline_name: Text,
  pipeline_root:Text,
  data_path: Text,
  enable_cache: bool,
  metadata_connection_config: Optional[
    metadata_store_pb2.ConnectionConfig] = None,
  beam_pipeline_args: Optional[List[Text]] = None
):
  components = []

  example_gen = tfx.components.CsvExampleGen(input_base=data_path)
  components.append(example_gen)

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

def run_pipeline():
  my_pipeline = create_pipeline(
    pipeline_name=PIPELINE_NAME,
    pipeline_root=PIPELINE_ROOT,
    data_path=DATA_PATH,
    enable_cache=ENABLE_CACHE,
    metadata_connection_config=tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)
    )

  tfx.orchestration.LocalDagRunner().run(my_pipeline)

    CsvExampleGen создает сериализованные примеры записей, используя данные в формате CSV по указанному пути к данным. Путем установки параметра input_base компонента CsvExampleGen с корнем данных.

  4. Создайте каталог data в том же каталоге, что и my_pipeline.py . Добавьте небольшой файл CSV в каталог data .

  5. Используйте следующую команду для запуска сценария my_pipeline.py .

    python my_pipeline.py

    Результат должен быть примерно следующим:

    INFO:absl:Component CsvExampleGen depends on [].
INFO:absl:Component CsvExampleGen is scheduled.
INFO:absl:Component CsvExampleGen is running.
INFO:absl:Running driver for CsvExampleGen
INFO:absl:MetadataStore with DB connection initialized
INFO:absl:Running executor for CsvExampleGen
INFO:absl:Generating examples.
INFO:absl:Using 1 process(es) for Local pipeline execution.
INFO:absl:Processing input csv data ./data/* to TFExample.
WARNING:root:Couldn't find python-snappy so the implementation of _TFRecordUtil._masked_crc32c is not as fast as it could be.
INFO:absl:Examples generated.
INFO:absl:Running publisher for CsvExampleGen
INFO:absl:MetadataStore with DB connection initialized
INFO:absl:Component CsvExampleGen is finished.

  6. Продолжайте итеративно добавлять компоненты в свой конвейер.