Введение в вершинные конвейеры

1. Обзор

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

Чему вы научитесь

Вы научитесь:

• Используйте SDK Kubeflow Pipelines для создания масштабируемых конвейеров машинного обучения.
• Создайте и запустите трехэтапный вводный конвейер, принимающий текстовый ввод.
• Создайте и запустите конвейер, который обучает, оценивает и развертывает модель классификации AutoML.
• Используйте готовые компоненты для взаимодействия с сервисами Vertex AI, предоставляемые библиотекой google_cloud_pipeline_components
• Запланируйте выполнение задания конвейера с помощью Cloud Scheduler.

Общая стоимость запуска этой лабораторной работы в Google Cloud составляет около 25 долларов .

2. Введение в Vertex AI

В этой лабораторной работе используется новейший продукт для искусственного интеллекта, доступный в Google Cloud. Vertex AI интегрирует предложения машинного обучения в Google Cloud в единый процесс разработки. Ранее модели, обученные с помощью AutoML, и пользовательские модели были доступны через отдельные сервисы. Новое предложение объединяет оба варианта в единый API, а также включает другие новые продукты. Вы также можете перенести существующие проекты в Vertex AI.

Помимо услуг по обучению и развертыванию моделей, Vertex AI также предлагает множество продуктов MLOps, включая Vertex Pipelines (на котором сосредоточено внимание в этой лабораторной работе), мониторинг моделей, хранилище признаков и многое другое. Все продукты Vertex AI представлены на диаграмме ниже.

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

Почему конвейеры машинного обучения полезны?

Прежде чем углубиться в детали, давайте сначала разберемся, зачем вам может понадобиться конвейер. Представьте, что вы создаете рабочий процесс машинного обучения, который включает обработку данных, обучение модели, настройку гиперпараметров, оценку и развертывание модели. Каждый из этих шагов может иметь различные зависимости, что может стать громоздким, если рассматривать весь рабочий процесс как монолит. По мере масштабирования вашего процесса машинного обучения вы можете захотеть поделиться своим рабочим процессом с другими членами вашей команды, чтобы они могли запускать его и вносить свой вклад в код. Без надежного, воспроизводимого процесса это может стать сложно. В конвейерах каждый шаг вашего процесса машинного обучения является собственным контейнером. Это позволяет разрабатывать шаги независимо и отслеживать входные и выходные данные каждого шага воспроизводимым образом. Вы также можете планировать или запускать запуск вашего конвейера на основе других событий в вашей облачной среде, например, запускать запуск конвейера при появлении новых обучающих данных.

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

3. Настройка облачной среды

Для выполнения этого практического задания вам потребуется проект Google Cloud Platform с включенной оплатой. Чтобы создать проект, следуйте инструкциям здесь .

Шаг 1: Запустите Cloud Shell

В этой лабораторной работе вы будете работать в сеансе Cloud Shell , который представляет собой интерпретатор команд, размещенный на виртуальной машине, работающей в облаке Google. Вы с таким же успехом могли бы выполнить этот раздел локально на своем компьютере, но использование Cloud Shell обеспечивает всем доступ к воспроизводимому опыту в согласованной среде. После завершения лабораторной работы вы можете повторить этот раздел на своем компьютере.

При первом запросе в Cloud Shell, требующем авторизации, вы увидите диалоговое окно «Авторизовать Cloud Shell», подобное показанному ниже, в котором вы можете смело нажать кнопку «Авторизовать».

Активировать Cloud Shell

В правом верхнем углу консоли Cloud Console нажмите кнопку ниже, чтобы активировать Cloud Shell :

Если вы никогда раньше не запускали Cloud Shell, вам будет показан промежуточный экран (внизу), описывающий его назначение. В этом случае нажмите «Продолжить» (и вы больше никогда его не увидите). Вот как выглядит этот одноразовый экран:

Подготовка и подключение к Cloud Shell займут всего несколько минут.

Эта виртуальная машина оснащена всеми необходимыми инструментами разработки. Она предоставляет постоянный домашний каталог размером 5 ГБ и работает в облаке Google, что значительно повышает производительность сети и аутентификацию. Большая часть, если не вся, работа в этом практическом задании может быть выполнена с помощью обычного браузера или вашего Chromebook.

После подключения к Cloud Shell вы увидите, что ваша аутентификация пройдена и проект уже настроен на ваш идентификатор проекта.

Выполните следующую команду в Cloud Shell, чтобы подтвердить свою аутентификацию:

gcloud auth list

В выводе команды вы должны увидеть что-то подобное:

Примечание: Инструмент командной строки gcloud — это мощный и унифицированный инструмент командной строки в Google Cloud. Он предустановлен в Cloud Shell. Вы заметите, что он поддерживает автодополнение по клавише Tab. Для получения дополнительной информации см. обзор инструмента командной строки gcloud .

Выполните следующую команду в Cloud Shell, чтобы убедиться, что команда gcloud знает о вашем проекте:

gcloud config list project

вывод команды

[core]
project = <PROJECT_ID>

Если это не так, вы можете установить это с помощью следующей команды:

gcloud config set project <PROJECT_ID>

вывод команды

Updated property [core/project].

В Cloud Shell есть несколько переменных окружения, в том числе GOOGLE_CLOUD_PROJECT , которая содержит имя нашего текущего проекта Cloud. Мы будем использовать её в различных местах на протяжении этой лабораторной работы. Вы можете увидеть её, выполнив команду:

echo $GOOGLE_CLOUD_PROJECT

Шаг 2: Включение API

На следующих этапах вы увидите, где эти сервисы необходимы (и почему), а пока выполните эту команду, чтобы предоставить вашему проекту доступ к сервисам Compute Engine, Container Registry и Vertex AI:

gcloud services enable compute.googleapis.com         \
                       containerregistry.googleapis.com  \
                       aiplatform.googleapis.com  \
                       cloudbuild.googleapis.com \
                       cloudfunctions.googleapis.com

В результате должно появиться сообщение об успешном завершении, похожее на это:

Operation "operations/acf.cc11852d-40af-47ad-9d59-477a12847c9e" finished successfully.

Шаг 3: Создайте корзину облачного хранилища.

Для запуска задачи обучения на Vertex AI нам потребуется хранилище для сохранения ресурсов модели. Хранилище должно быть региональным. В данном случае мы используем us-central , но вы можете использовать другой регион (просто замените его в этом примере). Если у вас уже есть хранилище, вы можете пропустить этот шаг.

Для создания корзины выполните следующие команды в терминале Cloud Shell:

BUCKET_NAME=gs://$GOOGLE_CLOUD_PROJECT-bucket
gsutil mb -l us-central1 $BUCKET_NAME

Далее мы предоставим нашей учетной записи вычислительной службы доступ к этому хранилищу. Это гарантирует, что Vertex Pipelines будет иметь необходимые разрешения для записи файлов в это хранилище. Выполните следующую команду, чтобы добавить это разрешение:

gcloud projects describe $GOOGLE_CLOUD_PROJECT > project-info.txt
PROJECT_NUM=$(cat project-info.txt | sed -nre 's:.*projectNumber\: (.*):\1:p')
SVC_ACCOUNT="${PROJECT_NUM//\'/}[e-mail]"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT --member serviceAccount:$SVC_ACCOUNT --role roles/storage.objectAdmin

Шаг 4: Создайте экземпляр Vertex AI Workbench.

В разделе Vertex AI вашей облачной консоли нажмите на Workbench:

Затем в разделе «Управляемые пользователем блокноты» нажмите «Создать блокнот» :

Затем выберите тип экземпляра TensorFlow Enterprise 2.3 (с LTS) без графических процессоров :

Используйте параметры по умолчанию, а затем нажмите «Создать» .

Шаг 5: Откройте свой блокнот

После создания экземпляра выберите «Открыть JupyterLab» :

4. Настройка вершинных конвейеров

Для использования Vertex Pipelines нам потребуется установить несколько дополнительных библиотек:

• Kubeflow Pipelines : Это SDK, который мы будем использовать для создания нашего конвейера. Vertex Pipelines поддерживает запуск конвейеров, созданных как с помощью Kubeflow Pipelines, так и с помощью TFX.
• Компоненты Google Cloud Pipeline : Эта библиотека предоставляет готовые компоненты, упрощающие взаимодействие с сервисами Vertex AI на этапах вашего конвейера.

Шаг 1: Создайте блокнот Python и установите библиотеки.

Сначала в меню «Запуск» вашего экземпляра блокнота создайте блокнот, выбрав Python 3 :

Доступ к меню запуска можно получить, нажав на знак «+» в верхнем левом углу вашего экземпляра ноутбука.

Для установки обеих служб, которые мы будем использовать в этой лабораторной работе, сначала установите пользовательский флаг в ячейке блокнота:

USER_FLAG = "--user"

Затем выполните следующую команду на своем ноутбуке:

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.7.0 --upgrade
!pip3 install {USER_FLAG} kfp==1.8.9 google-cloud-pipeline-components==0.2.0

После установки этих пакетов потребуется перезапустить ядро:

import os

if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython

    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)

Наконец, убедитесь, что вы правильно установили пакеты. Версия KFP SDK должна быть >=1.8:

!python3 -c "import kfp; print('KFP SDK version: {}'.format(kfp.__version__))"
!python3 -c "import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))"

Шаг 2: Укажите идентификатор проекта и категорию (bucket).

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

Если вы не знаете идентификатор своего проекта, вы можете получить его, выполнив следующую команду:

import os
PROJECT_ID = ""

# Get your Google Cloud project ID from gcloud
if not os.getenv("IS_TESTING"):
    shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID: ", PROJECT_ID)

В противном случае, установите значение здесь:

if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "your-project-id"  # @param {type:"string"}

Затем создайте переменную для хранения имени вашего сегмента. Если вы создали её в этой лабораторной работе, подойдёт следующий код. В противном случае вам потребуется установить это имя вручную:

BUCKET_NAME="gs://" + PROJECT_ID + "-bucket"

Шаг 3: Импорт библиотек

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

import kfp

from kfp.v2 import compiler, dsl
from kfp.v2.dsl import component, pipeline, Artifact, ClassificationMetrics, Input, Output, Model, Metrics

from google.cloud import aiplatform
from google_cloud_pipeline_components import aiplatform as gcc_aip
from typing import NamedTuple

Шаг 4: Определение констант

Последнее, что нам нужно сделать перед созданием конвейера, — это определить несколько константных переменных. PIPELINE_ROOT — это путь к Cloud Storage, куда будут записываться артефакты, созданные нашим конвейером. В данном случае мы используем регион us-central1 , но если вы использовали другой регион при создании своего хранилища, обновите переменную REGION в приведенном ниже коде:

PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin
REGION="us-central1"

PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/"
PIPELINE_ROOT

После выполнения приведенного выше кода вы должны увидеть корневой каталог вашего конвейера. Это местоположение в Cloud Storage, куда будут записываться артефакты из вашего конвейера. Оно будет иметь формат gs://YOUR-BUCKET-NAME/pipeline_root/

5. Создание первого конвейера обработки данных

Чтобы ознакомиться с принципами работы Vertex Pipelines, сначала создадим небольшой конвейер с использованием KFP SDK. Этот конвейер не связан с машинным обучением (не волнуйтесь, мы к этому ещё вернёмся!), мы используем его для обучения:

• Как создавать пользовательские компоненты в SDK KFP
• Как запустить и отслеживать работу конвейера в Vertex Pipelines

Мы создадим конвейер обработки данных, который будет выводить предложение, используя два выходных параметра: название продукта и описание в виде эмодзи. Этот конвейер будет состоять из трех компонентов:

• product_name : Этот компонент принимает в качестве входных данных название продукта (или любое другое существительное) и возвращает эту строку в качестве выходных данных.
• emoji : Этот компонент берет текстовое описание эмодзи и преобразует его в эмодзи. Например, текстовый код для ✨ — «блестки». Этот компонент использует библиотеку эмодзи, чтобы показать вам, как управлять внешними зависимостями в вашем конвейере.
• build_sentence : Этот заключительный компонент будет использовать результаты работы двух предыдущих для построения предложения с использованием эмодзи. Например, результатом может быть "Vertex Pipelines is ✨".

Давайте начнём программировать!

Шаг 1: Создайте компонент на основе функции Python.

Используя SDK KFP, мы можем создавать компоненты на основе функций Python. Мы будем использовать это для 3 компонентов в нашем первом конвейере. Сначала мы создадим компонент product_name , который просто принимает строку в качестве входных данных и возвращает эту строку. Добавьте следующее в свой блокнот:

@component(base_image="python:3.9", output_component_file="first-component.yaml")
def product_name(text: str) -> str:
    return text

Давайте подробнее рассмотрим синтаксис:

• Декоратор @component компилирует эту функцию в компонент при запуске конвейера. Вы будете использовать это всякий раз, когда пишете собственный компонент.
• Параметр base_image указывает образ контейнера, который будет использовать этот компонент.
• Параметр output_component_file является необязательным и указывает YAML-файл, в который будет записан скомпилированный компонент. После запуска ячейки вы должны увидеть этот файл, записанный в ваш экземпляр блокнота. Если вы хотите поделиться этим компонентом с кем-либо, вы можете отправить им сгенерированный YAML-файл, и они загрузят его следующим образом:

product_name_component = kfp.components.load_component_from_file('./first-component.yaml')

• Строка -> str после определения функции указывает тип выходных данных для этого компонента.

Шаг 2: Создайте два дополнительных компонента.

Для завершения нашего конвейера мы создадим еще два компонента. Первый из них принимает строку в качестве входных данных и преобразует эту строку в соответствующий эмодзи, если таковой имеется. Он возвращает кортеж, содержащий переданный входной текст и результирующий эмодзи:

@component(packages_to_install=["emoji"])
def emoji(
    text: str,
) -> NamedTuple(
    "Outputs",
    [
        ("emoji_text", str),  # Return parameters
        ("emoji", str),
    ],
):
    import emoji

    emoji_text = text
    emoji_str = emoji.emojize(':' + emoji_text + ':', language='alias')
    print("output one: {}; output_two: {}".format(emoji_text, emoji_str))
    return (emoji_text, emoji_str)

Этот компонент немного сложнее предыдущего. Давайте разберем, что в нем нового:

• Параметр packages_to_install указывает компоненту, какие внешние библиотеки зависят от этого контейнера. В данном случае мы используем библиотеку emoji .
• Этот компонент возвращает NamedTuple с именем Outputs . Обратите внимание, что каждая из строк в этом кортеже имеет ключи: emoji_text и emoji . Мы будем использовать их в нашем следующем компоненте для доступа к выходным данным.

Последний компонент в этом конвейере обработает выходные данные первых двух компонентов и объединит их для возврата строки:

@component
def build_sentence(
    product: str,
    emoji: str,
    emojitext: str
) -> str:
    print("We completed the pipeline, hooray!")
    end_str = product + " is "
    if len(emoji) > 0:
        end_str += emoji
    else:
        end_str += emojitext
    return(end_str)

Возможно, вы задаетесь вопросом: как этот компонент узнает, что нужно использовать результаты предыдущих шагов, которые вы определили? Хороший вопрос! Мы все это объединим на следующем шаге.

Шаг 3: Объединение компонентов в конвейер.

Определения компонентов, которые мы задали выше, создали фабричные функции, которые можно использовать в определении конвейера для создания шагов. Чтобы настроить конвейер, используйте декоратор @pipeline , дайте конвейеру имя и описание, а также укажите корневой путь, куда должны записываться артефакты вашего конвейера. Под артефактами мы подразумеваем любые выходные файлы, генерируемые вашим конвейером. В этом вводном конвейере они не генерируются, но в следующем конвейере будут.

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

• product_task принимает в качестве входных данных название продукта. Здесь мы передаем `Vertex Pipelines`, но вы можете изменить это на что угодно.
• emoji_task принимает в качестве входных данных текстовый код эмодзи. Вы также можете изменить его на любой другой по своему усмотрению. Например, "party_face" относится к эмодзи 🥳. Обратите внимание, что поскольку ни этот компонент, ни компонент product_task не имеют шагов, передающих в них входные данные, мы вручную указываем входные данные для них при определении нашего конвейера обработки данных.
• Последний шаг в нашем конвейере — consumer_task имеет три входных параметра:
  • Результат выполнения product_task . Поскольку этот шаг выдает только один результат, мы можем сослаться на него через product_task.output .
  • Результат выполнения шага emoji_task в виде emoji . См. компонент emoji , определенный выше, где мы указали выходные параметры.
  • Аналогично, именованный emoji_text выводится из компонента emoji . Если в наш конвейер передается текст, не соответствующий эмодзи, он будет использовать этот текст для построения предложения.

@pipeline(
    name="hello-world",
    description="An intro pipeline",
    pipeline_root=PIPELINE_ROOT,
)

# You can change the `text` and `emoji_str` parameters here to update the pipeline output
def intro_pipeline(text: str = "Vertex Pipelines", emoji_str: str = "sparkles"):
    product_task = product_name(text)
    emoji_task = emoji(emoji_str)
    consumer_task = build_sentence(
        product_task.output,
        emoji_task.outputs["emoji"],
        emoji_task.outputs["emoji_text"],
    )

Шаг 4: Скомпилируйте и запустите конвейер.

После определения конвейера вы готовы его скомпилировать. Следующая команда сгенерирует JSON-файл, который вы будете использовать для запуска конвейера:

compiler.Compiler().compile(
    pipeline_func=intro_pipeline, package_path="intro_pipeline_job.json"
)

Далее создайте переменную типа TIMESTAMP . Мы будем использовать её в идентификаторе задания:

from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")

Затем определите задачу конвейера:

job = aiplatform.PipelineJob(
    display_name="hello-world-pipeline",
    template_path="intro_pipeline_job.json",
    job_id="hello-world-pipeline-{0}".format(TIMESTAMP),
    enable_caching=True
)

Наконец, запустите задание, чтобы создать новый этап выполнения конвейера:

job.submit()

После выполнения этой ячейки вы должны увидеть журналы со ссылкой для просмотра хода выполнения конвейера в консоли:

Перейдите по этой ссылке. После завершения ваш конвейер должен выглядеть примерно так:

Выполнение этого конвейера займет 5-6 минут . После завершения вы можете щелкнуть по компоненту build-sentence чтобы увидеть конечный результат:

Теперь, когда вы знакомы с принципами работы KFP SDK и Vertex Pipelines, вы готовы создать конвейер, который будет создавать и развертывать модель машинного обучения с использованием других сервисов Vertex AI. Давайте начнем!

6. Создание комплексного конвейера машинного обучения.

Пришло время создать свой первый конвейер машинного обучения. В этом конвейере мы будем использовать набор данных UCI Machine Learning Dry beans , взятый из статьи: KOKLU, M. и OZKAN, IA, (2020), "Многоклассовая классификация сухих бобов с использованием методов компьютерного зрения и машинного обучения". В журнале Computers and Electronics in Agriculture, 174, 105507. DOI .

Это табличный набор данных, и в нашем конвейере мы будем использовать его для обучения, оценки и развертывания модели AutoML, которая классифицирует бобы по одному из 7 типов на основе их характеристик.

Этот конвейер будет:

• Создайте набор данных в
• Обучите модель табличной классификации с помощью AutoML.
• Получите метрики оценки для этой модели.
• На основе полученных метрик оценки решите, следует ли развертывать модель с использованием условной логики в Vertex Pipelines.
• Разверните модель на конечной точке с помощью Vertex Prediction.

Каждый из описанных шагов будет представлять собой компонент. Большинство шагов конвейера будут использовать предварительно созданные компоненты для сервисов Vertex AI через библиотеку google_cloud_pipeline_components , которую мы импортировали ранее в этом практическом занятии. В этом разделе мы сначала определим один пользовательский компонент. Затем мы определим остальные шаги конвейера, используя предварительно созданные компоненты. Предварительно созданные компоненты упрощают доступ к сервисам Vertex AI, таким как обучение и развертывание моделей.

Большую часть времени на этом этапе занимает обучение AutoML, которое займет около часа.

Шаг 1: Пользовательский компонент для оценки модели.

Созданный нами пользовательский компонент будет использоваться в конце нашего конвейера, после завершения обучения модели. Этот компонент будет выполнять несколько функций:

• Получите метрики оценки обученной модели классификации AutoML.
• Проанализируйте метрики и отобразите их в пользовательском интерфейсе Vertex Pipelines.
• Сравните метрики с пороговым значением, чтобы определить, следует ли развертывать модель.

Прежде чем определить компонент, давайте разберемся с его входными и выходными параметрами. В качестве входных данных этот конвейер принимает некоторые метаданные о нашем проекте Cloud, полученную обученную модель (мы определим этот компонент позже), метрики оценки модели и параметр thresholds_dict_str . Параметр thresholds_dict_str мы определим при запуске конвейера. В случае этой модели классификации это будет значение площади под кривой ROC, при котором следует развернуть модель. Например, если мы передадим 0,95, это означает, что мы хотим, чтобы наш конвейер развертывал модель только в том случае, если этот показатель выше 95%.

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

@component(
    base_image="gcr.io/deeplearning-platform-release/tf2-cpu.2-3:latest",
    output_component_file="tabular_eval_component.yaml",
    packages_to_install=["google-cloud-aiplatform"],
)
def classification_model_eval_metrics(
    project: str,
    location: str,  # "us-central1",
    api_endpoint: str,  # "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str,
    model: Input[Artifact],
    metrics: Output[Metrics],
    metricsc: Output[ClassificationMetrics],
) -> NamedTuple("Outputs", [("dep_decision", str)]):  # Return parameter.

    import json
    import logging

    from google.cloud import aiplatform as aip

    # Fetch model eval info
    def get_eval_info(client, model_name):
        from google.protobuf.json_format import MessageToDict

        response = client.list_model_evaluations(parent=model_name)
        metrics_list = []
        metrics_string_list = []
        for evaluation in response:
            print("model_evaluation")
            print(" name:", evaluation.name)
            print(" metrics_schema_uri:", evaluation.metrics_schema_uri)
            metrics = MessageToDict(evaluation._pb.metrics)
            for metric in metrics.keys():
                logging.info("metric: %s, value: %s", metric, metrics[metric])
            metrics_str = json.dumps(metrics)
            metrics_list.append(metrics)
            metrics_string_list.append(metrics_str)

        return (
            evaluation.name,
            metrics_list,
            metrics_string_list,
        )

    # Use the given metrics threshold(s) to determine whether the model is
    # accurate enough to deploy.
    def classification_thresholds_check(metrics_dict, thresholds_dict):
        for k, v in thresholds_dict.items():
            logging.info("k {}, v {}".format(k, v))
            if k in ["auRoc", "auPrc"]:  # higher is better
                if metrics_dict[k] < v:  # if under threshold, don't deploy
                    logging.info("{} < {}; returning False".format(metrics_dict[k], v))
                    return False
        logging.info("threshold checks passed.")
        return True

    def log_metrics(metrics_list, metricsc):
        test_confusion_matrix = metrics_list[0]["confusionMatrix"]
        logging.info("rows: %s", test_confusion_matrix["rows"])

        # log the ROC curve
        fpr = []
        tpr = []
        thresholds = []
        for item in metrics_list[0]["confidenceMetrics"]:
            fpr.append(item.get("falsePositiveRate", 0.0))
            tpr.append(item.get("recall", 0.0))
            thresholds.append(item.get("confidenceThreshold", 0.0))
        print(f"fpr: {fpr}")
        print(f"tpr: {tpr}")
        print(f"thresholds: {thresholds}")
        metricsc.log_roc_curve(fpr, tpr, thresholds)

        # log the confusion matrix
        annotations = []
        for item in test_confusion_matrix["annotationSpecs"]:
            annotations.append(item["displayName"])
        logging.info("confusion matrix annotations: %s", annotations)
        metricsc.log_confusion_matrix(
            annotations,
            test_confusion_matrix["rows"],
        )

        # log textual metrics info as well
        for metric in metrics_list[0].keys():
            if metric != "confidenceMetrics":
                val_string = json.dumps(metrics_list[0][metric])
                metrics.log_metric(metric, val_string)
        # metrics.metadata["model_type"] = "AutoML Tabular classification"

    logging.getLogger().setLevel(logging.INFO)
    aip.init(project=project)
    # extract the model resource name from the input Model Artifact
    model_resource_path = model.metadata["resourceName"]
    logging.info("model path: %s", model_resource_path)

    client_options = {"api_endpoint": api_endpoint}
    # Initialize client that will be used to create and send requests.
    client = aip.gapic.ModelServiceClient(client_options=client_options)
    eval_name, metrics_list, metrics_str_list = get_eval_info(
        client, model_resource_path
    )
    logging.info("got evaluation name: %s", eval_name)
    logging.info("got metrics list: %s", metrics_list)
    log_metrics(metrics_list, metricsc)

    thresholds_dict = json.loads(thresholds_dict_str)
    deploy = classification_thresholds_check(metrics_list[0], thresholds_dict)
    if deploy:
        dep_decision = "true"
    else:
        dep_decision = "false"
    logging.info("deployment decision is %s", dep_decision)

    return (dep_decision,)

Шаг 2: Добавление предварительно созданных компонентов Google Cloud

На этом этапе мы определим остальные компоненты нашего конвейера и посмотрим, как они все взаимодействуют друг с другом. Сначала задайте отображаемое имя для запуска конвейера, используя метку времени:

import time
DISPLAY_NAME = 'automl-beans{}'.format(str(int(time.time())))
print(DISPLAY_NAME)

Затем скопируйте следующее в новую ячейку блокнота:

@pipeline(name="automl-tab-beans-training-v2",
                  pipeline_root=PIPELINE_ROOT)
def pipeline(
    bq_source: str = "bq://aju-dev-demos.beans.beans1",
    display_name: str = DISPLAY_NAME,
    project: str = PROJECT_ID,
    gcp_region: str = "us-central1",
    api_endpoint: str = "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str = '{"auRoc": 0.95}',
):
    dataset_create_op = gcc_aip.TabularDatasetCreateOp(
        project=project, display_name=display_name, bq_source=bq_source
    )

    training_op = gcc_aip.AutoMLTabularTrainingJobRunOp(
        project=project,
        display_name=display_name,
        optimization_prediction_type="classification",
        budget_milli_node_hours=1000,
        column_transformations=[
            {"numeric": {"column_name": "Area"}},
            {"numeric": {"column_name": "Perimeter"}},
            {"numeric": {"column_name": "MajorAxisLength"}},
            {"numeric": {"column_name": "MinorAxisLength"}},
            {"numeric": {"column_name": "AspectRation"}},
            {"numeric": {"column_name": "Eccentricity"}},
            {"numeric": {"column_name": "ConvexArea"}},
            {"numeric": {"column_name": "EquivDiameter"}},
            {"numeric": {"column_name": "Extent"}},
            {"numeric": {"column_name": "Solidity"}},
            {"numeric": {"column_name": "roundness"}},
            {"numeric": {"column_name": "Compactness"}},
            {"numeric": {"column_name": "ShapeFactor1"}},
            {"numeric": {"column_name": "ShapeFactor2"}},
            {"numeric": {"column_name": "ShapeFactor3"}},
            {"numeric": {"column_name": "ShapeFactor4"}},
            {"categorical": {"column_name": "Class"}},
        ],
        dataset=dataset_create_op.outputs["dataset"],
        target_column="Class",
    )
    model_eval_task = classification_model_eval_metrics(
        project,
        gcp_region,
        api_endpoint,
        thresholds_dict_str,
        training_op.outputs["model"],
    )

    with dsl.Condition(
        model_eval_task.outputs["dep_decision"] == "true",
        name="deploy_decision",
    ):

        endpoint_op = gcc_aip.EndpointCreateOp(
            project=project,
            location=gcp_region,
            display_name="train-automl-beans",
        )

        gcc_aip.ModelDeployOp(
            model=training_op.outputs["model"],
            endpoint=endpoint_op.outputs["endpoint"],
            dedicated_resources_min_replica_count=1,
            dedicated_resources_max_replica_count=1,
            dedicated_resources_machine_type="n1-standard-4",
        )

Давайте посмотрим, что происходит в этом коде:

• Во-первых, как и в предыдущем конвейере, мы определяем входные параметры, которые принимает этот конвейер. Нам необходимо установить их вручную, поскольку они не зависят от результатов других шагов конвейера.
• Остальная часть конвейера использует несколько предварительно созданных компонентов для взаимодействия с сервисами Vertex AI:
  • TabularDatasetCreateOp создает табличный набор данных в Vertex AI, используя в качестве источника данных либо Cloud Storage, либо BigQuery. В этом конвейере мы передаем данные через URL-адрес таблицы BigQuery.
  • AutoMLTabularTrainingJobRunOp запускает задачу обучения AutoML для табличного набора данных. Мы передаем этому компоненту несколько параметров конфигурации, включая тип модели (в данном случае, классификация), некоторые данные о столбцах, продолжительность обучения и указатель на набор данных. Обратите внимание, что для передачи набора данных этому компоненту мы предоставляем выходные данные предыдущего компонента через dataset_create_op.outputs["dataset"]
  • EndpointCreateOp создает конечную точку в Vertex AI. Конечная точка, созданная на этом шаге, будет передана в качестве входных данных следующему компоненту.
  • ModelDeployOp развертывает заданную модель на конечной точке в Vertex AI. В данном случае мы используем конечную точку, созданную на предыдущем шаге. Доступны дополнительные параметры конфигурации, но здесь мы указываем тип машины конечной точки и модель, которую хотим развернуть. Мы передаем модель, получая доступ к результатам этапа обучения в нашем конвейере.
• Этот конвейер также использует условную логику — функцию Vertex Pipelines, которая позволяет определять условие, а также различные ветви в зависимости от результата этого условия. Напомним, что при определении нашего конвейера мы передали параметр thresholds_dict_str . Это пороговое значение точности, которое мы используем для определения того, следует ли развертывать нашу модель на конечной точке. Для реализации этого мы используем класс Condition из KFP SDK. Передаваемое условие — это результат работы пользовательского компонента eval, который мы определили ранее в этом практическом задании. Если это условие истинно, конвейер продолжит выполнение компонента deploy_op . Если точность не соответствует нашему предопределенному порогу, конвейер остановится здесь и не будет развертывать модель.

Шаг 3: Скомпилируйте и запустите сквозной конвейер машинного обучения.

После того, как мы определили весь наш конвейер обработки данных, пришло время его скомпилировать:

compiler.Compiler().compile(
    pipeline_func=pipeline, package_path="tab_classif_pipeline.json"
)

Далее, определите задачу:

ml_pipeline_job = aiplatform.PipelineJob(
    display_name="automl-tab-beans-training",
    template_path="tab_classif_pipeline.json",
    pipeline_root=PIPELINE_ROOT,
    parameter_values={"project": PROJECT_ID, "display_name": DISPLAY_NAME},
    enable_caching=True
)

И наконец, запустите задание:

ml_pipeline_job.submit()

Перейдите по ссылке, указанной в логах после выполнения приведенной выше ячейки, чтобы увидеть свой конвейер в консоли. Выполнение этого конвейера займет чуть больше часа. Большая часть времени уходит на этап обучения AutoML. Завершенный конвейер будет выглядеть примерно так:

Если вы нажмете кнопку «Развернуть артефакты» вверху, вы сможете увидеть подробную информацию о различных артефактах, созданных в ходе вашего конвейера. Например, если вы щелкнете по артефакту dataset , вы увидите подробную информацию о созданном наборе данных Vertex AI. Вы можете перейти по этой ссылке на страницу этого набора данных:

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

Чтобы увидеть модель и конечную точку, созданные в результате выполнения этого конвейера, перейдите в раздел моделей и щелкните по модели с именем automl-beans Там вы должны увидеть эту модель, развернутую на конечной точке:

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

Помимо просмотра графика конвейера в консоли, вы также можете использовать Vertex Pipelines для отслеживания происхождения данных . Под отслеживанием происхождения данных мы подразумеваем отслеживание артефактов, созданных на протяжении всего конвейера. Это может помочь нам понять, где были созданы артефакты и как они используются в рабочем процессе машинного обучения. Например, чтобы просмотреть отслеживание происхождения данных для набора данных, созданного в этом конвейере, щелкните по артефакту набора данных, а затем выберите «Просмотреть происхождение данных» :

Это показывает нам все места, где используется этот артефакт:

Шаг 4: Сравнение метрик между запусками конвейера.

Если вы запускаете этот конвейер несколько раз, вам может потребоваться сравнить метрики между запусками. Для доступа к метаданным запуска можно использовать метод aiplatform.get_pipeline_df() . Здесь мы получим метаданные для всех запусков этого конвейера и загрузим их в DataFrame Pandas:

pipeline_df = aiplatform.get_pipeline_df(pipeline="automl-tab-beans-training-v2")
small_pipeline_df = pipeline_df.head(2)
small_pipeline_df

На этом лабораторная работа завершена!

🎉 Поздравляем! 🎉

Вы научились использовать Vertex AI для:

• Используйте SDK Kubeflow Pipelines для создания сквозных конвейеров с пользовательскими компонентами.
• Запускайте свои конвейеры на Vertex Pipelines и запускайте их выполнение с помощью SDK.
• Просматривайте и анализируйте граф Vertex Pipelines в консоли.
• Используйте готовые компоненты конвейера для добавления сервисов Vertex AI в ваш конвейер.
• Планирование повторяющихся заданий конвейера

Чтобы узнать больше о различных компонентах Vertex, ознакомьтесь с документацией .

7. Уборка

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

Шаг 1: Остановите или удалите свой экземпляр Notebooks.

Если вы хотите продолжить использовать созданный в этой лабораторной работе блокнот, рекомендуется выключать его, когда он не используется. В интерфейсе блокнотов в вашей облачной консоли выберите блокнот, а затем выберите «Остановить» . Если вы хотите полностью удалить экземпляр, выберите «Удалить» .

Шаг 2: Удалите свою конечную точку

Чтобы удалить развернутую конечную точку, перейдите в раздел «Конечные точки» в консоли Vertex AI и нажмите значок удаления:

Затем нажмите кнопку «Отменить развертывание» в появившемся окне:

Наконец, перейдите в раздел «Модели» на вашей консоли, найдите нужную модель и в меню с тремя точками справа нажмите «Удалить модель» .

Шаг 3: Удалите свой сегмент облачного хранилища.

Чтобы удалить сегмент хранилища, воспользуйтесь меню навигации в консоли Cloud Console, перейдите в раздел «Хранилище», выберите свой сегмент и нажмите «Удалить».