Тестирование конвейера с использованием Stub Executors

Введение

Чтобы продолжить работу с этим руководством, вам необходимо пройти руководство по шаблону template.ipynb до шага 6 .

В этом документе будут представлены инструкции по тестированию конвейера TensorFlow Extended (TFX) с использованием BaseStubExecuctor , который генерирует поддельные артефакты с использованием золотых тестовых данных. Это предназначено для того, чтобы пользователи могли заменять исполнителей, которые они не хотят тестировать, чтобы сэкономить время на запуске реальных исполнителей. Исполнитель-заглушка предоставляется в пакете TFX Python в разделе tfx.experimental.pipeline_testing.base_stub_executor .

Это руководство является расширением учебника template.ipynb , поэтому вы также будете использовать набор данных Taxi Trips , выпущенный властями города Чикаго. Мы настоятельно рекомендуем вам попробовать изменить компоненты перед использованием исполнителей-заглушек.

1. Запишите результаты конвейера в Google Cloud Storage.

Сначала нам нужно записать выходные данные конвейера, чтобы исполнители-заглушки могли скопировать артефакты из записанных выходных данных.

Поскольку в этом руководстве предполагается, что вы завершили работу template.ipynb до шага 6, успешный запуск конвейера должен быть сохранен в MLMD . Доступ к информации о выполнении в MLMD можно получить с помощью сервера gRPC.

Откройте терминал и выполните следующие команды:

  1. Создайте файл kubeconfig с соответствующими учетными данными: bash gcloud container clusters get-credentials $cluster_name --zone $compute_zone --project $gcp_project_id $compute_zone — это регион для механизма gcp, а $gcp_project_id — это идентификатор вашего проекта GCP.

  2. Настройте переадресацию портов для подключения к MLMD: bash nohup kubectl port-forward deployment/metadata-grpc-deployment -n $namespace $port:8080 & $namespace — это пространство имен кластера, а $port — любой неиспользуемый порт, который будет использоваться для переадресация портов.

  3. Клонируйте репозиторий tfx GitHub. Внутри каталога tfx выполните следующую команду: 

python tfx/experimental/pipeline_testing/pipeline_recorder.py \
--output_dir=gs://<gcp_project_id>-kubeflowpipelines-default/testdata \
--host=$host \
--port=$port \
--pipeline_name=$pipeline_name

$output_dir должен быть указан путь в Google Cloud Storage, куда должны записываться выходные данные конвейера, поэтому обязательно замените <gcp_project_id> идентификатором проекта GCP.

$host и $port — это имя хоста и порт сервера метаданных grpc для подключения к MLMD. $port должен быть указан номер порта, который вы использовали для переадресации портов, и вы можете установить «localhost» в качестве имени хоста.

В учебнике template.ipynb имя конвейера по умолчанию установлено как «my_pipeline», поэтому установите pipeline_name="my_pipeline" . Если вы изменили имя конвейера при запуске руководства по шаблону, вам следует соответствующим образом изменить --pipeline_name .

2. Включите исполнители-заглушки в Kubeflow DAG Runner.

Сначала убедитесь, что предопределенный шаблон скопирован в каталог вашего проекта с помощью команды CLI tfx template copy . В скопированных исходных файлах необходимо отредактировать следующие два файла.

  1. Создайте файл с именем stub_component_launcher.py в каталоге, где находится kubeflow_dag_runner.py, и поместите в него следующее содержимое.

    from tfx.experimental.pipeline_testing import base_stub_component_launcher
from pipeline import configs

class StubComponentLauncher(
    base_stub_component_launcher.BaseStubComponentLauncher):
  pass

# GCS directory where KFP outputs are recorded
test_data_dir = "gs://{}/testdata".format(configs.GCS_BUCKET_NAME)
# TODO: customize self.test_component_ids to test components, replacing other
# component executors with a BaseStubExecutor.
test_component_ids = ['Trainer']
StubComponentLauncher.initialize(
    test_data_dir=test_data_dir,
    test_component_ids=test_component_ids)

  2. Установите идентификаторы компонентов в виде списка идентификаторов компонентов, которые необходимо протестировать (другими словами, исполнители других компонентов заменяются на BaseStubExecutor).

  3. Откройте kubeflow_dag_runner.py . Добавьте следующий оператор импорта вверху, чтобы использовать только что добавленный класс StubComponentLauncher .

    import stub_component_launcher

  4. В kubeflow_dag_runner.py добавьте класс StubComponentLauncher в supported_launcher_class KubeflowDagRunnerConfig , чтобы включить запуск исполнителей-заглушек:

    runner_config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
    supported_launcher_classes=[
        stub_component_launcher.StubComponentLauncher
    ],

3. Обновите и запустите конвейер с помощью исполнителей-заглушек.

Обновите существующий конвейер, добавив в него измененное определение конвейера с помощью исполнителей-заглушек.

tfx pipeline update --pipeline-path=kubeflow_dag_runner.py \
  --endpoint=$endpoint --engine=kubeflow

$endpoint должна быть указана конечная точка вашего кластера KFP.

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

tfx run create --pipeline-name $pipeline_name --endpoint=$endpoint \
  --engine=kubeflow

Уборка

Используйте команду fg для доступа к переадресации портов в фоновом режиме, а затем ctrl-C для завершения. Вы можете удалить каталог с записанными выходными данными конвейера, используя gsutil -m rm -R $output_dir .

Чтобы очистить все ресурсы Google Cloud, используемые в этом проекте, вы можете удалить проект Google Cloud, который вы использовали в этом руководстве.

Кроме того, вы можете очистить отдельные ресурсы, посетив каждую консоль: - Google Cloud Storage - Google Container Registry - Google Kubernetes Engine .