<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 <!-- <table align="left">
  <td>
    <a href="https
Step1:  Install Vertex AI SDK for Python
The notebook uises Vertex AI SDK for Python to interact with Vertex AI services. The high-level google-cloud-aiplatform library is designed to simplify common data science workflows by using wrapper classes and opinionated defaults.
Step2:  Restart the kernel
After you install the additional packages, you need to restart the notebook kernel so it can find the packages.
Step3:  Check the versions of the packages you installed.  The KFP SDK version should be >=1.6.
Step4:  Before you begin
This notebook does not require a GPU runtime.
Set up your Google Cloud project
The following steps are required, regardless of your notebook environment.
Select or create a Google Cloud project. When you first create an account, you get a $300 free credit towards your compute/storage costs.
Make sure that billing is enabled for your project.
Enable following APIs in your project required for running the tutorial
Vertex AI API
Cloud Storage API
Container Registry API
Cloud Build API
If you are running this notebook locally, you will need to install the Cloud SDK.
Enter your project ID in the cell below. Then run the cell to make sure the Cloud SDK uses the right project for all the commands in this notebook.
Note
Step5:  Timestamp
If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a timestamp for each instance session, and append it onto the name of resources you create in this tutorial.
Step6:  Authenticate your Google Cloud account
If you are using Google Cloud Notebooks, your environment is already authenticated. Skip this step.
If you are using Colab, run the cell below and follow the instructions
when prompted to authenticate your account via oAuth.
Otherwise, follow these steps
Step7:  Create a Cloud Storage bucket
The following steps are required, regardless of your notebook environment.
When you submit a training job using the Cloud SDK, you upload a Python package containing your training code to a Cloud Storage bucket. Vertex AI runs the code from this package. In this tutorial, Vertex AI also saves the trained model that results from your job in the same bucket. Using this model artifact, you can then create Vertex AI model and endpoint resources in order to serve online predictions.
Set the name of your Cloud Storage bucket below. It must be unique across all Cloud Storage buckets.
You may also change the REGION variable, which is used for operations throughout the rest of this notebook. Make sure to choose a region where Vertex AI services are available. You may not use a Multi-Regional Storage bucket for training with Vertex AI.
Step8:  Only if your bucket doesn't already exist
Step9:  Finally, validate access to your Cloud Storage bucket by examining its contents
Step10:  Import libraries and define constants
Import python libraries required to run the pipeline and define constants.
Step11:  High Level Flow of Building a Pipeline
Following is the high level flow to define and submit a pipeline on Vertex AI Pipelines
Step12:  1. Component
Step13:  Copy training application code and Dockerfile from local path to GCS location
Step15:  Define custom pipeline component to build custom training container
Step17:  There are a few things to notice about the component specification
Step18:  3. Component
Step20:  Define custom pipeline component to create model archive file
Step21:  4. Component
Step22:  Copy serving Dockerfile from local path to GCS location
Step24:  Define custom pipeline component to build custom serving container
Step26:  5. Component
Step27:  Define Pipeline Specification
The pipeline definition describes how input and output parameters and artifacts are passed between the steps.
Set environment variables
These environment variables will be used to define resource specifications such as training jobs, model resource etc.
Step28:  Create pipeline configuration file
Pipeline configuration files helps in templatizing a pipeline enabling to run the same pipeline with different parameters.
Step29:  Define pipeline specification
The pipeline is defined as a standalone Python function annotated with the @kfp.dsl.pipeline decorator, specifying the pipeline's name and the root path where the pipeline's artifacts are stored.
The pipeline definition consists of both pre-built and custom defined components
Step30:  Let’s unpack this code and understand a few things
Step31:  Submit Pipeline for Execution on Vertex AI Pipelines
Pipeline is submitted to Vertex AI Pipelines by defining a PipelineJob using Vertex AI SDK for Python client, passing necessary pipeline inputs.
Step32:  When the pipeline is submitted, the logs show a link to view the pipeline run on Google Cloud Console or access the run by opening Pipelines dashboard on Vertex AI
Step33:  Monitoring the Pipeline
You can monitor the progress of a pipeline execution by navigating to the Vertex AI Pipelines dashboard.
INFO
Step34:  Cleaning up
Cleaning up training and deployment resources
To clean up all Google Cloud resources used in this notebook, you can delete the Google Cloud project you used for the tutorial.
Otherwise, you can delete the individual resources you created in this tutorial
Step35:  Define clients for jobs, models and endpoints
Step36:  Define functions to list the jobs, models and endpoints starting with APP_NAME defined earlier in the notebook
Step37:  Deleting custom training jobs
Step38:  Deleting hyperparameter tuning jobs
Step39:  Undeploy models and Delete endpoints
Step40:  Deleting models
Step41:  Deleting pipeline runs
Step42:  Delete contents from the staging bucket
NOTE
Step43:  Delete images from Container Registry
Deletes all the container images created in this tutorial with prefix defined by variable APP_NAME from the registry. All associated tags are also deleted.
<ASSISTANT_TASK:>
Python Code:
import os
# The Google Cloud Notebook product has specific requirements
IS_GOOGLE_CLOUD_NOTEBOOK = os.path.exists("/opt/deeplearning/metadata/env_version")
# Google Cloud Notebook requires dependencies to be installed with '--user'
USER_FLAG = ""
if IS_GOOGLE_CLOUD_NOTEBOOK:
    USER_FLAG = "--user"
!pip -q install {USER_FLAG} --upgrade kfp
!pip -q install {USER_FLAG} --upgrade google-cloud-pipeline-components
Explanation: <!-- <table align="left">
  <td>
    <a href="https://colab.research.google.com/github/GoogleCloudPlatform/ai-platform-samples/blob/master/ai-platform-unified/notebooks/notebook_template.ipynb"">
      <img src="https://cloud.google.com/ml-engine/images/colab-logo-32px.png" alt="Colab logo"> Run in Colab
    </a>
  </td>
  <td>
    <a href="https://github.com/GoogleCloudPlatform/ai-platform-samples/blob/master/ai-platform-unified/notebooks/notebook_template.ipynb">
      <img src="https://cloud.google.com/ml-engine/images/github-logo-32px.png" alt="GitHub logo">
      View on GitHub
    </a>
  </td>
</table> -->
Orchestrating ML workflow to Train and Deploy a PyTorch Text Classification Model on Vertex AI Pipelines
Overview
This notebook is an extension to the previous notebook to fine-tune and deploy a pre-trained BERT model from HuggingFace Hub for sentiment classification task. This notebook shows how to automate and monitor a PyTorch based ML workflow by orchestrating the pipeline in a serverless manner using Vertex AI Pipelines.
The notebook defines a pipeline using Kubeflow Pipelines v2 (kfp.v2) SDK and submits the pipeline to Vertex AI Pipelines services.
Dataset
The notebook uses IMDB movie review dataset from Hugging Face Datasets.
Objective
How to orchestrate PyTorch ML workflows on Vertex AI and emphasize first class support for training, deploying and orchestrating PyTorch workflows on Vertex AI. 
Table of Contents
This notebook covers following sections:
High Level Flow of Building a Pipeline: Understand pipeline concepts and pipeline schematic
Define the Pipeline Components: Authoring custom pipeline components  for PyTorch based ML Workflow
Define Pipeline Specification: Author pipeline specification using KFP v2 SDK for PyTorch based ML workflow
Submit Pipeline: Compile and execute pipeline on Vertex AI Pipelines
Monitoring the Pipeline: Monitor progress of pipeline and view logs, lineage, artifacts and pipeline runs
Costs
This tutorial uses billable components of Google Cloud Platform (GCP):
Vertex AI Workbench
Vertex AI Training
Vertex AI Predictions
Vertex AI Pipelines
Cloud Storage
Container Registry
Cloud Build [Optional]
Learn about Vertex AI Pricing, Cloud Storage Pricing and Cloud Build Pricing, and use the Pricing Calculator to generate a cost estimate based on your projected usage.
NOTE: This notebook does not require a GPU runtime. However, you must have GPU quota for running the jobs with GPUs launched by pipelines. Check the quotas page to ensure that you have enough GPUs available in your project. If GPUs are not listed on the quotas page or you require additional GPU quota, request a quota increase. Free Trial accounts do not receive GPU quota by default.
Set up your local development environment
If you are using Colab or Google Cloud Notebooks, your environment already meets
all the requirements to run this notebook. You can skip this step.
Otherwise, make sure your environment meets this notebook's requirements.
You need the following:
The Google Cloud SDK
Git
Python 3
virtualenv
Jupyter notebook running in a virtual environment with Python 3
The Google Cloud guide to Setting up a Python development
environment and the Jupyter
installation guide provide detailed instructions
for meeting these requirements. The following steps provide a condensed set of
instructions:
Install and initialize the Cloud SDK.
Install Python 3.
Install virtualenv and create a virtual environment that uses Python 3. Activate the virtual environment.
To install Jupyter, run pip3 install jupyter on the command-line in a terminal shell.
To launch Jupyter, run jupyter notebook on the command-line in a terminal shell.
Open this notebook in the Jupyter Notebook Dashboard.
Install additional packages
Following are the Python dependencies required for this notebook and will be installed in the Notebooks instance itself.
Kubeflow Pipelines v2 SDK
Google Cloud Pipeline Components 
Vertex AI SDK for Python 
The notebook has been tested with the following versions of Kubeflow Pipelines SDK and Google Cloud Pipeline Components
kfp version: 1.8.10
google_cloud_pipeline_components version: 0.2.2
End of explanation
!pip -q install {USER_FLAG} --upgrade google-cloud-aiplatform
Explanation: Install Vertex AI SDK for Python
The notebook uises Vertex AI SDK for Python to interact with Vertex AI services. The high-level google-cloud-aiplatform library is designed to simplify common data science workflows by using wrapper classes and opinionated defaults.
End of explanation
# Automatically restart kernel after installs
import os
if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython
    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)
Explanation: Restart the kernel
After you install the additional packages, you need to restart the notebook kernel so it can find the packages.
End of explanation
!python3 -c "import kfp; print('kfp version: {}'.format(kfp.__version__))"
!python3 -c "import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))"
Explanation: Check the versions of the packages you installed.  The KFP SDK version should be >=1.6.
End of explanation
PROJECT_ID = "[your-project-id]"  # <---CHANGE THIS TO YOUR PROJECT
import os
# Get your Google Cloud project ID using google.auth
if not os.getenv("IS_TESTING"):
    import google.auth
    _, PROJECT_ID = google.auth.default()
    print("Project ID: ", PROJECT_ID)
# validate PROJECT_ID
if PROJECT_ID == "" or PROJECT_ID is None or PROJECT_ID == "[your-project-id]":
    print(
        f"Please set your project id before proceeding to next step. Currently it's set as {PROJECT_ID}"
    )
Explanation: Before you begin
This notebook does not require a GPU runtime.
Set up your Google Cloud project
The following steps are required, regardless of your notebook environment.
Select or create a Google Cloud project. When you first create an account, you get a $300 free credit towards your compute/storage costs.
Make sure that billing is enabled for your project.
Enable following APIs in your project required for running the tutorial
Vertex AI API
Cloud Storage API
Container Registry API
Cloud Build API
If you are running this notebook locally, you will need to install the Cloud SDK.
Enter your project ID in the cell below. Then run the cell to make sure the Cloud SDK uses the right project for all the commands in this notebook.
Note: Jupyter runs lines prefixed with ! as shell commands, and it interpolates Python variables prefixed with $ into these commands.
Set your project ID
If you don't know your project ID, you may be able to get your project ID using gcloud.
End of explanation
from datetime import datetime
def get_timestamp():
    return datetime.now().strftime("%Y%m%d%H%M%S")
TIMESTAMP = get_timestamp()
print(f"TIMESTAMP = {TIMESTAMP}")
Explanation: Timestamp
If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a timestamp for each instance session, and append it onto the name of resources you create in this tutorial.
End of explanation
import os
import sys
# If you are running this notebook in Colab, run this cell and follow the
# instructions to authenticate your GCP account. This provides access to your
# Cloud Storage bucket and lets you submit training jobs and prediction
# requests.
# The Google Cloud Notebook product has specific requirements
IS_GOOGLE_CLOUD_NOTEBOOK = os.path.exists("/opt/deeplearning/metadata/env_version")
# If on Google Cloud Notebooks, then don't execute this code
if not IS_GOOGLE_CLOUD_NOTEBOOK:
    if "google.colab" in sys.modules:
        from google.colab import auth as google_auth
        google_auth.authenticate_user()
    # If you are running this notebook locally, replace the string below with the
    # path to your service account key and run this cell to authenticate your GCP
    # account.
    elif not os.getenv("IS_TESTING"):
        %env GOOGLE_APPLICATION_CREDENTIALS ''
Explanation: Authenticate your Google Cloud account
If you are using Google Cloud Notebooks, your environment is already authenticated. Skip this step.
If you are using Colab, run the cell below and follow the instructions
when prompted to authenticate your account via oAuth.
Otherwise, follow these steps:
In the Cloud Console, go to the Create service account key page.
Click Create service account.
In the Service account name field, enter a name, and click Create.
In the Grant this service account access to project section, click the Role drop-down list. Type "Vertex AI" into the filter box, and select Vertex AI Administrator. Type "Storage Object Admin" into the filter box, and select Storage Object Admin.
Click Create. A JSON file that contains your key downloads to your local environment.
Enter the path to your service account key as the GOOGLE_APPLICATION_CREDENTIALS variable in the cell below and run the cell.
End of explanation
BUCKET_NAME = "gs://[your-bucket-name]"  # @param {type:"string"}
REGION = "us-central1"  # @param {type:"string"}
if BUCKET_NAME == "" or BUCKET_NAME is None or BUCKET_NAME == "gs://[your-bucket-name]":
    BUCKET_NAME = "gs://" + PROJECT_ID + "aip-" + TIMESTAMP
Explanation: Create a Cloud Storage bucket
The following steps are required, regardless of your notebook environment.
When you submit a training job using the Cloud SDK, you upload a Python package containing your training code to a Cloud Storage bucket. Vertex AI runs the code from this package. In this tutorial, Vertex AI also saves the trained model that results from your job in the same bucket. Using this model artifact, you can then create Vertex AI model and endpoint resources in order to serve online predictions.
Set the name of your Cloud Storage bucket below. It must be unique across all Cloud Storage buckets.
You may also change the REGION variable, which is used for operations throughout the rest of this notebook. Make sure to choose a region where Vertex AI services are available. You may not use a Multi-Regional Storage bucket for training with Vertex AI.
End of explanation
! gsutil mb -l $REGION $BUCKET_NAME
Explanation: Only if your bucket doesn't already exist: Run the following cell to create your Cloud Storage bucket.
End of explanation
! gsutil ls -al $BUCKET_NAME
Explanation: Finally, validate access to your Cloud Storage bucket by examining its contents:
End of explanation
%load_ext autoreload
%autoreload 2
import os
from typing import NamedTuple
import google_cloud_pipeline_components
import kfp
from google.cloud import aiplatform
from google.cloud.aiplatform import gapic as aip
from google.cloud.aiplatform import pipeline_jobs
from google.protobuf.json_format import MessageToDict
from google_cloud_pipeline_components import aiplatform as aip_components
from google_cloud_pipeline_components.experimental import custom_job
from kfp.v2 import compiler, dsl
from kfp.v2.dsl import Input, Metrics, Model, Output, component
APP_NAME = "finetuned-bert-classifier"
PATH = %env PATH
%env PATH={PATH}:/home/jupyter/.local/bin
# Pipeline root is the GCS path to store the artifacts from the pipeline runs
PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/{APP_NAME}"
print(f"Kubeflow Pipelines SDK version = {kfp.__version__}")
print(
    f"Google Cloud Pipeline Components version = {google_cloud_pipeline_components.__version__}"
)
print(f"Pipeline Root = {PIPELINE_ROOT}")
Explanation: Import libraries and define constants
Import python libraries required to run the pipeline and define constants.
End of explanation
!mkdir -p ./pipelines
Explanation: High Level Flow of Building a Pipeline
Following is the high level flow to define and submit a pipeline on Vertex AI Pipelines:
Define pipeline components involved in training and deploying a PyTorch model
Define a pipeline by stitching the components in the workflow including pre-built Google Cloud pipeline components and custom components
Compile and submit the pipeline to Vertex AI Pipelines service to run the workflow
Monitor the pipeline and analyze the metrics and artifacts generated
This notebook builds on the training and serving code developed in previously this notebook.
Concepts of a Pipeline
Let's look at the terminology and concepts used in Kubeflow Pipelines SDK v2.
Component: A component is a self-contained set of code performing a single task in a ML workflow, for example, training a model. A component interface is composed of inputs, outputs and a container image that the component’s code runs in - including an executable code and environment definition.
Pipeline: A pipeline is composed of modular tasks defined as components that are chained together via inputs and outputs. Pipeline definition includes configuration such as parameters required to run the pipeline. Each component in a pipeline executes independently and the data (inputs and outputs) is passed between the components in a serialized format.
Inputs & Outputs: Component’s inputs and outputs must be annotated with data type, which makes input or output a parameter or an artifact. 
Parameters: Parameters are inputs or outputs to support simple data types such as str, int, float, bool, dict, list. Input parameters are always passed by value between the components and are stored in the Vertex ML Metadata service.
Artifacts: Artifacts are references to the objects or files produced by pipeline runs that are passed as inputs or outputs. Artifacts support rich or larger data types such as datasets, models, metrics, visualizations that are written as files or objects. Artifacts are defined by name, uri and metadata which is stored automatically in the Vertex ML Metadata service and the actual content of artifacts is referred to a path in Cloud Storage bucket. Input artifacts are always passed by reference.
Learn more about KFP SDK v2 concepts here.
Pipeline schematic
Following is the high level pipeline schematic with tasks involved in the pipeline for the PyTorch based text classification model including input and outputs:
Build custom training image: This step builds a custom training container image from the training application code and associated Dockerfile with the dependencies. The output from this step is the Container or Artifact registry URI of the custom training container.
Run the custom training job to train and evaluate the model: This step downloads and preprocesses training data from IMDB sentiment classification dataset on  HuggingFace, then trains and evaluates a model on the custom training container from the previous step. The step outputs Cloud Storage path to the trained model artifacts and the model performance metrics.
Package model artifacts: This step packages trained model artifacts including custom prediction handler to create a model archive (.mar) file using Torch Model Archiver tool. The output from this step is the location of model archive (.mar) file on GCS.
Build custom serving image: The step builds a custom serving container running TorchServe HTTP server to serve prediction requests for the models mounted. The output from this step is the Container or Artifact registry URI to the custom serving container.
Upload model with custom serving container: This step creates a model resource using the custom serving image and MAR file from the previous steps.
Create an endpoint: This step creates a Vertex AI Endpoint to provide a service URL where the prediction requests are sent.
Deploy model to endpoint for serving: This step deploys the model to the endpoint created that creates necessary compute resources (based on the machine spec configured) to serve online prediction requests.
Validate deployment: This step sends test requests to the endpoint and validates the deployment.
Define the Pipeline Components for PyTorch based ML Workflow
The pipeline uses a mix of pre-built components from Google Cloud Pipeline Components SDK to interact with Google Cloud services such as Vertex AI and define custom components for some steps in the pipeline. This section of the notebook defines custom components to perform the tasks in the pipeline using KFP SDK v2 component spec. 
Create pipeline directory locally to save the component and pipeline specifications
End of explanation
%%writefile ./custom_container/Dockerfile
# Use pytorch GPU base image
# FROM gcr.io/cloud-aiplatform/training/pytorch-gpu.1-7
FROM us-docker.pkg.dev/vertex-ai/training/pytorch-gpu.1-10:latest
# set working directory
WORKDIR /app
# Install required packages
RUN pip install google-cloud-storage transformers datasets tqdm cloudml-hypertune
# Copies the trainer code to the docker image.
COPY ./trainer/__init__.py /app/trainer/__init__.py
COPY ./trainer/experiment.py /app/trainer/experiment.py
COPY ./trainer/utils.py /app/trainer/utils.py
COPY ./trainer/metadata.py /app/trainer/metadata.py
COPY ./trainer/model.py /app/trainer/model.py
COPY ./trainer/task.py /app/trainer/task.py
# Set up the entry point to invoke the trainer.
ENTRYPOINT ["python", "-m", "trainer.task"]
Explanation: 1. Component: Build Custom Training Container Image
This step builds a custom training container image using Cloud Build. The build job pulls the training application code and associated Dockerfile with the dependencies from GCS location and build/push the custom training container image to Container Registry. 
Inputs: The inputs to this component are GCS path to the training application code and Dockerfile.
Outputs: The output from this step is the Container or Artifact registry URI of the custom training container. 
Create Dockerfile from PyTorch GPU image as base, install required dependencies and copy training application code
End of explanation
# copy training Dockerfile
!gsutil cp ./custom_container/Dockerfile {BUCKET_NAME}/{APP_NAME}/train/
# copy training application code
!gsutil cp -r ./python_package/trainer/ {BUCKET_NAME}/{APP_NAME}/train/
# list copied files from GCS location
!gsutil ls -Rl {BUCKET_NAME}/{APP_NAME}/train/
print(
    f"Copied training application code and Dockerfile to {BUCKET_NAME}/{APP_NAME}/train/"
)
Explanation: Copy training application code and Dockerfile from local path to GCS location
End of explanation
@component(
    base_image="gcr.io/google.com/cloudsdktool/cloud-sdk:latest",
    packages_to_install=["google-cloud-build"],
    output_component_file="./pipelines/build_custom_train_image.yaml",
)
def build_custom_train_image(
    project: str, gs_train_src_path: str, training_image_uri: str
) -> NamedTuple("Outputs", [("training_image_uri", str)]):
    custom pipeline component to build custom training image using
    Cloud Build and the training application code and dependencies
    defined in the Dockerfile
    
    import logging
    import os
    from google.cloud.devtools import cloudbuild_v1 as cloudbuild
    from google.protobuf.duration_pb2 import Duration
    # initialize client for cloud build
    logging.getLogger().setLevel(logging.INFO)
    build_client = cloudbuild.services.cloud_build.CloudBuildClient()
    # parse step inputs to get path to Dockerfile and training application code
    gs_dockerfile_path = os.path.join(gs_train_src_path, "Dockerfile")
    gs_train_src_path = os.path.join(gs_train_src_path, "trainer/")
    logging.info(f"training_image_uri: {training_image_uri}")
    # define build steps to pull the training code and Dockerfile
    # and build/push the custom training container image
    build = cloudbuild.Build()
    build.steps = [
        {
            "name": "gcr.io/cloud-builders/gsutil",
            "args": ["cp", "-r", gs_train_src_path, "."],
        },
        {
            "name": "gcr.io/cloud-builders/gsutil",
            "args": ["cp", gs_dockerfile_path, "Dockerfile"],
        },
        # enabling Kaniko cache in a Docker build that caches intermediate
        # layers and pushes image automatically to Container Registry
        # https://cloud.google.com/build/docs/kaniko-cache
        {
            "name": "gcr.io/kaniko-project/executor:latest",
            "args": [f"--destination={training_image_uri}", "--cache=true"],
        },
    ]
    # override default timeout of 10min
    timeout = Duration()
    timeout.seconds = 7200
    build.timeout = timeout
    # create build
    operation = build_client.create_build(project_id=project, build=build)
    logging.info("IN PROGRESS:")
    logging.info(operation.metadata)
    # get build status
    result = operation.result()
    logging.info("RESULT:", result.status)
    # return step outputs
    return (training_image_uri,)
Explanation: Define custom pipeline component to build custom training container
End of explanation
@component(
    base_image="python:3.9",
    packages_to_install=[
        "google-cloud-pipeline-components",
        "google-cloud-aiplatform",
        "pandas",
        "fsspec",
    ],
    output_component_file="./pipelines/get_training_job_details.yaml",
)
def get_training_job_details(
    project: str,
    location: str,
    job_resource: str,
    eval_metric_key: str,
    model_display_name: str,
    metrics: Output[Metrics],
    model: Output[Model],
) -> NamedTuple(
    "Outputs", [("eval_metric", float), ("eval_loss", float), ("model_artifacts", str)]
):
    custom pipeline component to get model artifacts and performance
    metrics from custom training job
    
    import logging
    import shutil
    from collections import namedtuple
    import pandas as pd
    from google.cloud.aiplatform import gapic as aip
    from google.protobuf.json_format import Parse
    from google_cloud_pipeline_components.proto.gcp_resources_pb2 import \
        GcpResources
    # parse training job resource
    logging.info(f"Custom job resource = {job_resource}")
    training_gcp_resources = Parse(job_resource, GcpResources())
    custom_job_id = training_gcp_resources.resources[0].resource_uri
    custom_job_name = "/".join(custom_job_id.split("/")[-6:])
    logging.info(f"Custom job name parsed = {custom_job_name}")
    # get custom job information
    API_ENDPOINT = "{}-aiplatform.googleapis.com".format(location)
    client_options = {"api_endpoint": API_ENDPOINT}
    job_client = aip.JobServiceClient(client_options=client_options)
    job_resource = job_client.get_custom_job(name=custom_job_name)
    job_base_dir = job_resource.job_spec.base_output_directory.output_uri_prefix
    logging.info(f"Custom job base output directory = {job_base_dir}")
    # copy model artifacts
    logging.info(f"Copying model artifacts to {model.path}")
    destination = shutil.copytree(job_base_dir.replace("gs://", "/gcs/"), model.path)
    logging.info(destination)
    logging.info(f"Model artifacts located at {model.uri}/model/{model_display_name}")
    logging.info(f"Model artifacts located at model.uri = {model.uri}")
    # set model metadata
    start, end = job_resource.start_time, job_resource.end_time
    model.metadata["model_name"] = model_display_name
    model.metadata["framework"] = "pytorch"
    model.metadata["job_name"] = custom_job_name
    model.metadata["time_to_train_in_seconds"] = (end - start).total_seconds()
    # fetch metrics from the training job run
    metrics_uri = f"{model.path}/model/{model_display_name}/all_results.json"
    logging.info(f"Reading and logging metrics from {metrics_uri}")
    metrics_df = pd.read_json(metrics_uri, typ="series")
    for k, v in metrics_df.items():
        logging.info(f"     {k} -> {v}")
        metrics.log_metric(k, v)
    # capture eval metric and log to model metadata
    eval_metric = (
        metrics_df[eval_metric_key] if eval_metric_key in metrics_df.keys() else None
    )
    eval_loss = metrics_df["eval_loss"] if "eval_loss" in metrics_df.keys() else None
    logging.info(f"     {eval_metric_key} -> {eval_metric}")
    logging.info(f'     "eval_loss" -> {eval_loss}')
    model.metadata[eval_metric_key] = eval_metric
    model.metadata["eval_loss"] = eval_loss
    # return output parameters
    outputs = namedtuple("Outputs", ["eval_metric", "eval_loss", "model_artifacts"])
    return outputs(eval_metric, eval_loss, job_base_dir)
Explanation: There are a few things to notice about the component specification:
- The standalone function defined is converted as a pipeline component using the @kfp.v2.dsl.component decorator.
- All the arguments in the standalone function must have data type annotations because KFP uses the function’s inputs and outputs to define the component’s interface.
- By default Python 3.7 is used as the base image to run the code defined. You can configure the @component decorator to override the default image by specifying base_image, install additional python packages using packages_to_install parameter and write the compiled component file as a YAML file using output_component_file to share or reuse the component.
2. Component: Get Custom Training Job Details from Vertex AI
This step gets details from a custom training job from Vertex AI including training elapsed time, model performance metrics that will be used in the next step before the model deployment. The step additionally creates Model artifact with trained model artifacts. 
NOTE: The pre-built custom job component used in the pipeline outputs CustomJob resource but not the model artifacts.
Inputs:
job_resource: Custom job resource returned by pre-built CustomJob component
project: Project ID where the job ran
region: Region where the job ran
eval_metric_key: Evaluation metric key name such as eval_accuracy
model_display_name: Model display name for saving model artifacts
Outputs: 
model: Trained model artifacts created by the training job with added model metadata
metrics: Model performance metrics captured from the training job
End of explanation
# copy custom prediction handler
!gsutil cp ./predictor/custom_handler.py ./predictor/index_to_name.json {BUCKET_NAME}/{APP_NAME}/serve/predictor/
# list copied files from GCS location
!gsutil ls -lR {BUCKET_NAME}/{APP_NAME}/serve/
print(f"Copied custom prediction handler code to {BUCKET_NAME}/{APP_NAME}/serve/")
Explanation: 3. Component: Create Model Archive (MAR) file using Torch Model Archiver
This step packages trained model artifacts and custom prediction handler (define in the earlier notebook) as a model archive (.mar) file usign Torch Model Archiver tool.
Inputs:
model_display_name: Model display name for saving model archive file
model_version: Model version  for saving model archive file
handler: Location of custom prediction handler
model: Trained model artifacts from the previous step
Outputs: 
model_mar: Packaged model archive file (artifact) on GCS
mar_env: A list of environment variables required for creating model resource
mar_export_uri: GCS path to the model archive file
Copy custom prediction handler code from local path to GCS location
NOTE: Custom prediction handler is defined in the previous notebook
End of explanation
@component(
    base_image="python:3.9",
    packages_to_install=["torch-model-archiver"],
    output_component_file="./pipelines/generate_mar_file.yaml",
)
def generate_mar_file(
    model_display_name: str,
    model_version: str,
    handler: str,
    model: Input[Model],
    model_mar: Output[Model],
) -> NamedTuple("Outputs", [("mar_env_var", list), ("mar_export_uri", str)]):
    custom pipeline component to package model artifacts and custom
    handler to a model archive file using Torch Model Archiver tool
    
    import logging
    import os
    import subprocess
    import time
    from collections import namedtuple
    from pathlib import Path
    logging.getLogger().setLevel(logging.INFO)
    # create directory to save model archive file
    model_output_root = model.path
    mar_output_root = model_mar.path
    export_path = f"{mar_output_root}/model-store"
    try:
        Path(export_path).mkdir(parents=True, exist_ok=True)
    except Exception as e:
        logging.warning(e)
        # retry after pause
        time.sleep(2)
        Path(export_path).mkdir(parents=True, exist_ok=True)
    # parse and configure paths for model archive config
    handler_path = (
        handler.replace("gs://", "/gcs/") + "predictor/custom_handler.py"
        if handler.startswith("gs://")
        else handler
    )
    model_artifacts_dir = f"{model_output_root}/model/{model_display_name}"
    extra_files = [
        os.path.join(model_artifacts_dir, f)
        for f in os.listdir(model_artifacts_dir)
        if f != "pytorch_model.bin"
    ]
    # define model archive config
    mar_config = {
        "MODEL_NAME": model_display_name,
        "HANDLER": handler_path,
        "SERIALIZED_FILE": f"{model_artifacts_dir}/pytorch_model.bin",
        "VERSION": model_version,
        "EXTRA_FILES": ",".join(extra_files),
        "EXPORT_PATH": f"{model_mar.path}/model-store",
    }
    # generate model archive command
    archiver_cmd = (
        "torch-model-archiver --force "
        f"--model-name {mar_config['MODEL_NAME']} "
        f"--serialized-file {mar_config['SERIALIZED_FILE']} "
        f"--handler {mar_config['HANDLER']} "
        f"--version {mar_config['VERSION']}"
    )
    if "EXPORT_PATH" in mar_config:
        archiver_cmd += f" --export-path {mar_config['EXPORT_PATH']}"
    if "EXTRA_FILES" in mar_config:
        archiver_cmd += f" --extra-files {mar_config['EXTRA_FILES']}"
    if "REQUIREMENTS_FILE" in mar_config:
        archiver_cmd += f" --requirements-file {mar_config['REQUIREMENTS_FILE']}"
    # run archiver command
    logging.warning("Running archiver command: %s", archiver_cmd)
    with subprocess.Popen(
        archiver_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
    ) as p:
        _, err = p.communicate()
        if err:
            raise ValueError(err)
    # set output variables
    mar_env_var = [{"name": "MODEL_NAME", "value": model_display_name}]
    mar_export_uri = f"{model_mar.uri}/model-store/"
    outputs = namedtuple("Outputs", ["mar_env_var", "mar_export_uri"])
    return outputs(mar_env_var, mar_export_uri)
Explanation: Define custom pipeline component to create model archive file
End of explanation
%%bash -s $APP_NAME
APP_NAME=$1
cat << EOF > ./predictor/Dockerfile.serve
FROM pytorch/torchserve:latest-cpu
USER root
# run and update some basic packages software packages, including security libs
RUN apt-get update && \
    apt-get install -y software-properties-common && \
    add-apt-repository -y ppa:ubuntu-toolchain-r/test && \
    apt-get update && \
    apt-get install -y gcc-9 g++-9 apt-transport-https ca-certificates gnupg curl
# Install gcloud tools for gsutil as well as debugging
RUN echo "deb [signed-by=/usr/share/keyrings/cloud.google.gpg] http://packages.cloud.google.com/apt cloud-sdk main" | \
    tee -a /etc/apt/sources.list.d/google-cloud-sdk.list && \
    curl https://packages.cloud.google.com/apt/doc/apt-key.gpg | \
    apt-key --keyring /usr/share/keyrings/cloud.google.gpg add - && \
    apt-get update -y && \
    apt-get install google-cloud-sdk -y
USER model-server
# install dependencies
RUN python3 -m pip install --upgrade pip
RUN pip3 install transformers
ARG MODEL_NAME=$APP_NAME
ENV MODEL_NAME="\${MODEL_NAME}"
# health and prediction listener ports
ARG AIP_HTTP_PORT=7080
ENV AIP_HTTP_PORT="\${AIP_HTTP_PORT}"
ARG MODEL_MGMT_PORT=7081
# expose health and prediction listener ports from the image
EXPOSE "\${AIP_HTTP_PORT}"
EXPOSE "\${MODEL_MGMT_PORT}"
EXPOSE 8080 8081 8082 7070 7071
# create torchserve configuration file
USER root
RUN echo "service_envelope=json\n" \
    "inference_address=http://0.0.0.0:\${AIP_HTTP_PORT}\n" \
    "management_address=http://0.0.0.0:\${MODEL_MGMT_PORT}" >> \
    /home/model-server/config.properties
USER model-server
# run Torchserve HTTP serve to respond to prediction requests
CMD ["echo", "AIP_STORAGE_URI=\${AIP_STORAGE_URI}", ";", \
    "gsutil", "cp", "-r", "\${AIP_STORAGE_URI}/\${MODEL_NAME}.mar", "/home/model-server/model-store/", ";", \
    "ls", "-ltr", "/home/model-server/model-store/", ";", \
    "torchserve", "--start", "--ts-config=/home/model-server/config.properties", \
    "--models", "\${MODEL_NAME}=\${MODEL_NAME}.mar", \
    "--model-store", "/home/model-server/model-store"]
EOF
echo "Writing ./predictor/Dockerfile"
Explanation: 4. Component: Create custom serving container running TorchServe
The step builds a custom serving container running TorchServe HTTP server to serve prediction requests for the models mounted. The output from this step is the Container registry URI to the custom serving container.
Inputs:
project:  Project ID to run 
serving_image_uri: Custom serving container URI from Container registry
gs_serving_dependencies_path: Location of serving dependencies - Dockerfile
Outputs: 
serving_image_uri: Custom serving container URI from Container registry
Create Dockerfile from TorchServe CPU image as base, install required dependencies and run TorchServe serve command
End of explanation
# copy serving Dockerfile
!gsutil cp ./predictor/Dockerfile.serve {BUCKET_NAME}/{APP_NAME}/serve/
# list copied files from GCS location
!gsutil ls -lR {BUCKET_NAME}/{APP_NAME}/serve/
print(f"Copied serving Dockerfile to {BUCKET_NAME}/{APP_NAME}/serve/")
Explanation: Copy serving Dockerfile from local path to GCS location
End of explanation
@component(
    base_image="python:3.9",
    packages_to_install=["google-cloud-build"],
    output_component_file="./pipelines/build_custom_serving_image.yaml",
)
def build_custom_serving_image(
    project: str, gs_serving_dependencies_path: str, serving_image_uri: str
) -> NamedTuple("Outputs", [("serving_image_uri", str)],):
    custom pipeline component to build custom serving image using
    Cloud Build and dependencies defined in the Dockerfile
    
    import logging
    import os
    from google.cloud.devtools import cloudbuild_v1 as cloudbuild
    from google.protobuf.duration_pb2 import Duration
    logging.getLogger().setLevel(logging.INFO)
    build_client = cloudbuild.services.cloud_build.CloudBuildClient()
    logging.info(f"gs_serving_dependencies_path: {gs_serving_dependencies_path}")
    gs_dockerfile_path = os.path.join(gs_serving_dependencies_path, "Dockerfile.serve")
    logging.info(f"serving_image_uri: {serving_image_uri}")
    build = cloudbuild.Build()
    build.steps = [
        {
            "name": "gcr.io/cloud-builders/gsutil",
            "args": ["cp", gs_dockerfile_path, "Dockerfile"],
        },
        # enabling Kaniko cache in a Docker build that caches intermediate
        # layers and pushes image automatically to Container Registry
        # https://cloud.google.com/build/docs/kaniko-cache
        {
            "name": "gcr.io/kaniko-project/executor:latest",
            "args": [f"--destination={serving_image_uri}", "--cache=true"],
        },
    ]
    # override default timeout of 10min
    timeout = Duration()
    timeout.seconds = 7200
    build.timeout = timeout
    # create build
    operation = build_client.create_build(project_id=project, build=build)
    logging.info("IN PROGRESS:")
    logging.info(operation.metadata)
    # get build status
    result = operation.result()
    logging.info("RESULT:", result.status)
    # return step outputs
    return (serving_image_uri,)
Explanation: Define custom pipeline component to build custom serving container
End of explanation
@component(
    base_image="python:3.9",
    packages_to_install=["google-cloud-aiplatform", "google-cloud-pipeline-components"],
    output_component_file="./pipelines/make_prediction_request.yaml",
)
def make_prediction_request(project: str, bucket: str, endpoint: str, instances: list):
    custom pipeline component to pass prediction requests to Vertex AI
    endpoint and get responses
    
    import base64
    import logging
    from google.cloud import aiplatform
    from google.protobuf.json_format import Parse
    from google_cloud_pipeline_components.proto.gcp_resources_pb2 import \
        GcpResources
    logging.getLogger().setLevel(logging.INFO)
    aiplatform.init(project=project, staging_bucket=bucket)
    # parse endpoint resource
    logging.info(f"Endpoint = {endpoint}")
    gcp_resources = Parse(endpoint, GcpResources())
    endpoint_uri = gcp_resources.resources[0].resource_uri
    endpoint_id = "/".join(endpoint_uri.split("/")[-8:-2])
    logging.info(f"Endpoint ID = {endpoint_id}")
    # define endpoint client
    _endpoint = aiplatform.Endpoint(endpoint_id)
    # call prediction endpoint for each instance
    for instance in instances:
        if not isinstance(instance, (bytes, bytearray)):
            instance = instance.encode()
        logging.info(f"Input text: {instance.decode('utf-8')}")
        b64_encoded = base64.b64encode(instance)
        test_instance = [{"data": {"b64": f"{str(b64_encoded.decode('utf-8'))}"}}]
        response = _endpoint.predict(instances=test_instance)
        logging.info(f"Prediction response: {response.predictions}")
Explanation: 5. Component: Test model deployment making online prediction requests
This step sends test requests to the Vertex AI Endpoint and validates the deployment by sending test prediction requests. Deployment is considered successful when the response from model server returns text sentiment.
Inputs:
project:  Project ID to run 
bucket: Staging GCS bucket path
endpoint: Location of Vertex AI Endpoint from the Endpoint creation task
instances: List of test prediction requests
Outputs: 
None
End of explanation
os.environ["PROJECT_ID"] = PROJECT_ID
os.environ["BUCKET"] = BUCKET_NAME
os.environ["REGION"] = REGION
os.environ["APP_NAME"] = APP_NAME
Explanation: Define Pipeline Specification
The pipeline definition describes how input and output parameters and artifacts are passed between the steps.
Set environment variables
These environment variables will be used to define resource specifications such as training jobs, model resource etc.
End of explanation
%%writefile ./pipelines/pipeline_config.py
import os
from datetime import datetime
PROJECT_ID = os.getenv("PROJECT_ID", "")
BUCKET = os.getenv("BUCKET", "")
REGION = os.getenv("REGION", "us-central1")
APP_NAME = os.getenv("APP_NAME", "finetuned-bert-classifier")
VERSION = datetime.now().strftime("%Y%m%d%H%M%S")
MODEL_NAME = APP_NAME
MODEL_DISPLAY_NAME = f"{MODEL_NAME}-{VERSION}"
PIPELINE_NAME = f"pytorch-{APP_NAME}"
PIPELINE_ROOT = f"{BUCKET}/pipeline_root/{MODEL_NAME}"
GCS_STAGING = f"{BUCKET}/pipeline_root/{MODEL_NAME}"
TRAIN_IMAGE_URI = f"gcr.io/{PROJECT_ID}/pytorch_gpu_train_{MODEL_NAME}"
SERVE_IMAGE_URI = f"gcr.io/{PROJECT_ID}/pytorch_cpu_predict_{MODEL_NAME}"
MACHINE_TYPE = "n1-standard-8"
REPLICA_COUNT = "1"
ACCELERATOR_TYPE = "NVIDIA_TESLA_T4"
ACCELERATOR_COUNT = "1"
NUM_WORKERS = 1
SERVING_HEALTH_ROUTE = "/ping"
SERVING_PREDICT_ROUTE = f"/predictions/{MODEL_NAME}"
SERVING_CONTAINER_PORT= [{"containerPort": 7080}]
SERVING_MACHINE_TYPE = "n1-standard-4"
SERVING_MIN_REPLICA_COUNT = 1
SERVING_MAX_REPLICA_COUNT=1
SERVING_TRAFFIC_SPLIT='{"0": 100}'
Explanation: Create pipeline configuration file
Pipeline configuration files helps in templatizing a pipeline enabling to run the same pipeline with different parameters.
End of explanation
from pipelines import pipeline_config as cfg
@dsl.pipeline(
    name=cfg.PIPELINE_NAME,
    pipeline_root=cfg.PIPELINE_ROOT,
)
def pytorch_text_classifier_pipeline(
    pipeline_job_id: str,
    gs_train_script_path: str,
    gs_serving_dependencies_path: str,
    eval_acc_threshold: float,
    is_hp_tuning_enabled: str = "n",
):
    # ========================================================================
    # build custom training container image
    # ========================================================================
    # build custom container for training job passing the
    # GCS location of the training application code
    build_custom_train_image_task = (
        build_custom_train_image(
            project=cfg.PROJECT_ID,
            gs_train_src_path=gs_train_script_path,
            training_image_uri=cfg.TRAIN_IMAGE_URI,
        )
        .set_caching_options(True)
        .set_display_name("Build custom training image")
    )
    # ========================================================================
    # model training
    # ========================================================================
    # train the model on Vertex AI by submitting a CustomJob
    # using the custom container (no hyper-parameter tuning)
    # define training code arguments
    training_args = ["--num-epochs", "2", "--model-name", cfg.MODEL_NAME]
    # define job name
    JOB_NAME = f"{cfg.MODEL_NAME}-train-pytorch-cstm-cntr-{TIMESTAMP}"
    GCS_BASE_OUTPUT_DIR = f"{cfg.GCS_STAGING}/{TIMESTAMP}"
    # define worker pool specs
    worker_pool_specs = [
        {
            "machine_spec": {
                "machine_type": cfg.MACHINE_TYPE,
                "accelerator_type": cfg.ACCELERATOR_TYPE,
                "accelerator_count": cfg.ACCELERATOR_COUNT,
            },
            "replica_count": cfg.REPLICA_COUNT,
            "container_spec": {"image_uri": cfg.TRAIN_IMAGE_URI, "args": training_args},
        }
    ]
    run_train_task = (
        custom_job.CustomTrainingJobOp(
            project=cfg.PROJECT_ID,
            location=cfg.REGION,
            display_name=JOB_NAME,
            base_output_directory=GCS_BASE_OUTPUT_DIR,
            worker_pool_specs=worker_pool_specs,
        )
        .set_display_name("Run custom training job")
        .after(build_custom_train_image_task)
    )
    # ========================================================================
    # get training job details
    # ========================================================================
    training_job_details_task = get_training_job_details(
        project=cfg.PROJECT_ID,
        location=cfg.REGION,
        job_resource=run_train_task.output,
        eval_metric_key="eval_accuracy",
        model_display_name=cfg.MODEL_NAME,
    ).set_display_name("Get custom training job details")
    # ========================================================================
    # model deployment when condition is met
    # ========================================================================
    with dsl.Condition(
        training_job_details_task.outputs["eval_metric"] > eval_acc_threshold,
        name="model-deploy-decision",
    ):
        # ===================================================================
        # create model archive file
        # ===================================================================
        create_mar_task = generate_mar_file(
            model_display_name=cfg.MODEL_NAME,
            model_version=cfg.VERSION,
            handler=gs_serving_dependencies_path,
            model=training_job_details_task.outputs["model"],
        ).set_display_name("Create MAR file")
        # ===================================================================
        # build custom serving container running TorchServe
        # ===================================================================
        # build custom container for serving predictions using
        # the trained model artifacts served by TorchServe
        build_custom_serving_image_task = build_custom_serving_image(
            project=cfg.PROJECT_ID,
            gs_serving_dependencies_path=gs_serving_dependencies_path,
            serving_image_uri=cfg.SERVE_IMAGE_URI,
        ).set_display_name("Build custom serving image")
        # ===================================================================
        # create model resource
        # ===================================================================
        # upload model to vertex ai
        model_upload_task = (
            aip_components.ModelUploadOp(
                project=cfg.PROJECT_ID,
                display_name=cfg.MODEL_DISPLAY_NAME,
                serving_container_image_uri=cfg.SERVE_IMAGE_URI,
                serving_container_predict_route=cfg.SERVING_PREDICT_ROUTE,
                serving_container_health_route=cfg.SERVING_HEALTH_ROUTE,
                serving_container_ports=cfg.SERVING_CONTAINER_PORT,
                serving_container_environment_variables=create_mar_task.outputs[
                    "mar_env_var"
                ],
                artifact_uri=create_mar_task.outputs["mar_export_uri"],
            )
            .set_display_name("Upload model")
            .after(build_custom_serving_image_task)
        )
        # ===================================================================
        # create Vertex AI Endpoint
        # ===================================================================
        # create endpoint to deploy one or more models
        # An endpoint provides a service URL where the prediction requests are sent
        endpoint_create_task = (
            aip_components.EndpointCreateOp(
                project=cfg.PROJECT_ID,
                display_name=cfg.MODEL_NAME + "-endpoint",
            )
            .set_display_name("Create endpoint")
            .after(create_mar_task)
        )
        # ===================================================================
        # deploy model to Vertex AI Endpoint
        # ===================================================================
        # deploy models to endpoint to associates physical resources with the model
        # so it can serve online predictions
        model_deploy_task = aip_components.ModelDeployOp(
            endpoint=endpoint_create_task.outputs["endpoint"],
            model=model_upload_task.outputs["model"],
            deployed_model_display_name=cfg.MODEL_NAME,
            dedicated_resources_machine_type=cfg.SERVING_MACHINE_TYPE,
            dedicated_resources_min_replica_count=cfg.SERVING_MIN_REPLICA_COUNT,
            dedicated_resources_max_replica_count=cfg.SERVING_MAX_REPLICA_COUNT,
            traffic_split=cfg.SERVING_TRAFFIC_SPLIT,
        ).set_display_name("Deploy model to endpoint")
        # ===================================================================
        # test model deployment
        # ===================================================================
        # test model deployment by making online prediction requests
        test_instances = [
            "Jaw dropping visual affects and action! One of the best I have seen to date.",
            "Take away the CGI and the A-list cast and you end up with film with less punch.",
        ]
        predict_test_instances_task = make_prediction_request(
            project=cfg.PROJECT_ID,
            bucket=cfg.BUCKET,
            endpoint=model_deploy_task.outputs["gcp_resources"],
            instances=test_instances,
        ).set_display_name("Test model deployment making online predictions")
        predict_test_instances_task
Explanation: Define pipeline specification
The pipeline is defined as a standalone Python function annotated with the @kfp.dsl.pipeline decorator, specifying the pipeline's name and the root path where the pipeline's artifacts are stored.
The pipeline definition consists of both pre-built and custom defined components:
- Pre-built components from Google Cloud Pipeline Components SDK are defined for tasks calling Vertex AI services such as submitting custom training job (custom_job.CustomTrainingJobOp), uploading a model (ModelUploadOp), creating an endpoint (EndpointCreateOp) and deploying a model to the endpoint (ModelDeployOp)
- Custom components are defined for tasks to build custom containers for training (build_custom_train_image), get training job details (get_training_job_details), create mar file (generate_mar_file) and serving (build_custom_serving_image) and validating the model deployment task (ake_prediction_request). Refer to the notebook for custom component specification for these tasks.
End of explanation
PIPELINE_JSON_SPEC_PATH = "./pipelines/pytorch_text_classifier_pipeline_spec.json"
compiler.Compiler().compile(
    pipeline_func=pytorch_text_classifier_pipeline, package_path=PIPELINE_JSON_SPEC_PATH
)
Explanation: Let’s unpack this code and understand a few things:
A component’s inputs can be set from the pipeline's inputs (passed as arguments) or they can depend on the output of other components within this pipeline. For example, ModelUploadOp depends on custom serving container image URI from build_custom_serving_image task along with the pipeline’s inputs such as project id.
kfp.dsl.Condition is a control structure with a group of tasks which runs only when the condition is met. In this pipeline, model deployment steps run only when the trained model performance exceeds the set threshold. If not, those steps are skipped.
Each component in the pipeline runs within its own container image. You can specify machine type for each pipeline step such as CPU, GPU and memory limits. By default, each component runs as a Vertex AI CustomJob using an e2-standard-4 machine.
By default, pipeline execution caching is enabled. Vertex AI Pipelines service checks to see whether an execution of each pipeline step exists in Vertex ML metadata. It uses a combination of pipeline name, step’s inputs, output and component specification. When a matching execution already exists, the step is skipped and thereby reducing costs. Execution caching can be turned off at task level or at pipeline level.
Following is the runtime graph generated for this pipeline
To learn more about building pipelines, refer to the building Kubeflow pipelines section, and follow the pipelines samples and tutorials.
Submit Pipeline
Compile Pipeline Specification as JSON
After defining the pipeline, it must be compiled for executing on Vertex AI Pipeline services. When the pipeline is compiled, the KFP SDK analyzes the data dependencies between the components to create a directed acyclic graph. The compiled pipeline is in JSON format with all information required to run the pipeline.
End of explanation
# initialize Vertex AI SDK
aiplatform.init(project=PROJECT_ID, location=REGION)
# define pipeline parameters
# NOTE: These parameters can be included in the pipeline config file as needed
PIPELINE_JOB_ID = f"pipeline-{APP_NAME}-{get_timestamp()}"
TRAIN_APP_CODE_PATH = f"{BUCKET_NAME}/{APP_NAME}/train/"
SERVE_DEPENDENCIES_PATH = f"{BUCKET_NAME}/{APP_NAME}/serve/"
pipeline_params = {
    "pipeline_job_id": PIPELINE_JOB_ID,
    "gs_train_script_path": TRAIN_APP_CODE_PATH,
    "gs_serving_dependencies_path": SERVE_DEPENDENCIES_PATH,
    "eval_acc_threshold": 0.87,
    "is_hp_tuning_enabled": "n",
}
# define pipeline job
pipeline_job = pipeline_jobs.PipelineJob(
    display_name=cfg.PIPELINE_NAME,
    job_id=PIPELINE_JOB_ID,
    template_path=PIPELINE_JSON_SPEC_PATH,
    pipeline_root=PIPELINE_ROOT,
    parameter_values=pipeline_params,
    enable_caching=True,
)
Explanation: Submit Pipeline for Execution on Vertex AI Pipelines
Pipeline is submitted to Vertex AI Pipelines by defining a PipelineJob using Vertex AI SDK for Python client, passing necessary pipeline inputs.
End of explanation
# submit pipeline job for execution
response = pipeline_job.run(sync=True)
response
Explanation: When the pipeline is submitted, the logs show a link to view the pipeline run on Google Cloud Console or access the run by opening Pipelines dashboard on Vertex AI
End of explanation
# underscores are not supported in the pipeline name, so
# replace underscores with hyphen
df_pipeline = aiplatform.get_pipeline_df(pipeline=cfg.PIPELINE_NAME.replace("_", "-"))
df_pipeline
Explanation: Monitoring the Pipeline
You can monitor the progress of a pipeline execution by navigating to the Vertex AI Pipelines dashboard.
INFO:google.cloud.aiplatform.pipeline_jobs:Creating PipelineJob
INFO:google.cloud.aiplatform.pipeline_jobs:PipelineJob created. Resource name: projects/&lt;project-id&gt;/locations/&lt;region&gt;/pipelineJobs/pipeline-finetuned-bert-classifier-20220119061941
INFO:google.cloud.aiplatform.pipeline_jobs:To use this PipelineJob in another session:
INFO:google.cloud.aiplatform.pipeline_jobs:pipeline_job = aiplatform.PipelineJob.get('projects/&lt;project-id&gt;/locations/&lt;region&gt;/pipelineJobs/pipeline-finetuned-bert-classifier-20220119061941')
INFO:google.cloud.aiplatform.pipeline_jobs:View Pipeline Job:
https://console.cloud.google.com/vertex-ai/locations/region/pipelines/runs/pipeline-finetuned-bert-classifier-20220119061941?project=&lt;project-id&gt;
Component Execution Logs
Since every step in the pipeline runs in its own container or as a remote job (such as Dataflow, Dataproc job), you can view the step logs by clicking on "View Logs" button on a step.
Artifacts and Lineage
In the pipeline graph, you can notice the small boxes after each step. Those are artifacts generated from the step. For example, "Create MAR file" step generates MAR file as an artifact. Click on the artifact to know more details about it.
You can track lineage of an artifact describing its relationship with the steps in the pipeline. Vertex AI Pipelines automatically tracks the metadata and lineage. This lineage aids in establishing model governance and reproducibility. Click on "View Lineage" button on an artifact and it shows you the lineage graph as below.
Comparing Pipeline runs with the Vertex AI SDK
When running pipeline executions for different experiments, you may want to compare the metrics across the pipeline runs. You can compare pipeline runs from the Vertex AI Pipelines dashboard.
Alternatively, you can use aiplatform.get_pipeline_df() method from Vertex AI SDK for Python that fetches pipeline execution metadata for a pipeline and returns a Pandas dataframe.
End of explanation
delete_custom_job = False
delete_hp_tuning_job = False
delete_endpoint = False
delete_model = False
delete_bucket = False
delete_image = False
delete_pipeline_job = False
Explanation: Cleaning up
Cleaning up training and deployment resources
To clean up all Google Cloud resources used in this notebook, you can delete the Google Cloud project you used for the tutorial.
Otherwise, you can delete the individual resources you created in this tutorial:
Training Jobs
Model
Endpoint
Cloud Storage Bucket
Container Images
Pipeline runs
Set flags for the resource type to be deleted
End of explanation
# API Endpoint
API_ENDPOINT = "{}-aiplatform.googleapis.com".format(REGION)
# Vertex AI location root path for your dataset, model and endpoint resources
PARENT = f"projects/{PROJECT_ID}/locations/{REGION}"
client_options = {"api_endpoint": API_ENDPOINT}
# Initialize Vertex SDK
aiplatform.init(project=PROJECT_ID, staging_bucket=BUCKET_NAME)
# functions to create client
def create_job_client():
    client = aip.JobServiceClient(client_options=client_options)
    return client
def create_model_client():
    client = aip.ModelServiceClient(client_options=client_options)
    return client
def create_endpoint_client():
    client = aip.EndpointServiceClient(client_options=client_options)
    return client
def create_pipeline_client():
    client = aip.PipelineServiceClient(client_options=client_options)
    return client
clients = {}
clients["job"] = create_job_client()
clients["model"] = create_model_client()
clients["endpoint"] = create_endpoint_client()
clients["pipeline"] = create_pipeline_client()
Explanation: Define clients for jobs, models and endpoints
End of explanation
def list_custom_jobs():
    client = clients["job"]
    jobs = []
    response = client.list_custom_jobs(parent=PARENT)
    for row in response:
        _row = MessageToDict(row._pb)
        if _row["displayName"].startswith(APP_NAME):
            jobs.append((_row["name"], _row["displayName"]))
    return jobs
def list_hp_tuning_jobs():
    client = clients["job"]
    jobs = []
    response = client.list_hyperparameter_tuning_jobs(parent=PARENT)
    for row in response:
        _row = MessageToDict(row._pb)
        if _row["displayName"].startswith(APP_NAME):
            jobs.append((_row["name"], _row["displayName"]))
    return jobs
def list_models():
    client = clients["model"]
    models = []
    response = client.list_models(parent=PARENT)
    for row in response:
        _row = MessageToDict(row._pb)
        if _row["displayName"].startswith(APP_NAME):
            models.append((_row["name"], _row["displayName"]))
    return models
def list_endpoints():
    client = clients["endpoint"]
    endpoints = []
    response = client.list_endpoints(parent=PARENT)
    for row in response:
        _row = MessageToDict(row._pb)
        if _row["displayName"].startswith(APP_NAME):
            endpoints.append((_row["name"], _row["displayName"]))
    return endpoints
def list_pipelines():
    client = clients["pipeline"]
    pipelines = []
    request = aip.ListPipelineJobsRequest(
        parent=PARENT, filter=f'display_name="{cfg.PIPELINE_NAME}"', order_by="end_time"
    )
    response = client.list_pipeline_jobs(request=request)
    for row in response:
        _row = MessageToDict(row._pb)
        pipelines.append(_row["name"])
    return pipelines
Explanation: Define functions to list the jobs, models and endpoints starting with APP_NAME defined earlier in the notebook
End of explanation
# Delete the custom training using the Vertex AI fully qualified identifier for the custom training
try:
    if delete_custom_job:
        custom_jobs = list_custom_jobs()
        for job_id, job_name in custom_jobs:
            print(f"Deleting job {job_id} [{job_name}]")
            clients["job"].delete_custom_job(name=job_id)
except Exception as e:
    print(e)
Explanation: Deleting custom training jobs
End of explanation
# Delete the hyperparameter tuning jobs using the Vertex AI fully qualified identifier for the hyperparameter tuning job
try:
    if delete_hp_tuning_job:
        hp_tuning_jobs = list_hp_tuning_jobs()
        for job_id, job_name in hp_tuning_jobs:
            print(f"Deleting job {job_id} [{job_name}]")
            clients["job"].delete_hyperparameter_tuning_job(name=job_id)
except Exception as e:
    print(e)
Explanation: Deleting hyperparameter tuning jobs
End of explanation
# Delete the endpoint using the Vertex AI fully qualified identifier for the endpoint
try:
    if delete_endpoint:
        endpoints = list_endpoints()
        for endpoint_id, endpoint_name in endpoints:
            endpoint = aiplatform.Endpoint(endpoint_id)
            # undeploy models from the endpoint
            print(f"Undeploying all deployed models from the endpoint {endpoint_name}")
            endpoint.undeploy_all(sync=True)
            # deleting endpoint
            print(f"Deleting endpoint {endpoint_id} [{endpoint_name}]")
            clients["endpoint"].delete_endpoint(name=endpoint_id)
except Exception as e:
    print(e)
Explanation: Undeploy models and Delete endpoints
End of explanation
# Delete the model using the Vertex AI fully qualified identifier for the model
try:
    if delete_model:
        models = list_models()
        for model_id, model_name in models:
            print(f"Deleting model {model_id} [{model_name}]")
            clients["model"].delete_model(name=model_id)
except Exception as e:
    print(e)
Explanation: Deleting models
End of explanation
# Delete the pipeline execution using the Vertex AI fully qualified identifier for the pipeline job
try:
    if delete_pipeline_job:
        pipelines = list_pipelines()
        for pipeline_name in pipelines[:1]:
            print(f"Deleting pipeline run {pipeline_name}")
            if delete_custom_job:
                print("\t Deleting underlying custom jobs")
                pipeline_job = clients["pipeline"].get_pipeline_job(name=pipeline_name)
                pipeline_job = MessageToDict(pipeline_job._pb)
                task_details = pipeline_job["jobDetail"]["taskDetails"]
                for task in tasks:
                    if "containerDetail" in task["executorDetail"]:
                        custom_job_id = task["executorDetail"]["containerDetail"][
                            "mainJob"
                        ]
                        print(
                            f"\t Deleting custom job {custom_job_id} for task {task['taskName']}"
                        )
                        clients["job"].delete_custom_job(name=custom_job_id)
            clients["pipeline"].delete_pipeline_job(name=pipeline_name)
except Exception as e:
    print(e)
Explanation: Deleting pipeline runs
End of explanation
if delete_bucket and "BUCKET_NAME" in globals():
    print(f"Deleting all contents from the bucket {BUCKET_NAME}")
    shell_output = ! gsutil du -as $BUCKET_NAME
    print(
        f"Size of the bucket {BUCKET_NAME} before deleting = {shell_output[0].split()[0]} bytes"
    )
    # uncomment below line to delete contents of the bucket
    # ! gsutil rm -r $BUCKET_NAME
    shell_output = ! gsutil du -as $BUCKET_NAME
    if float(shell_output[0].split()[0]) > 0:
        print(
            "PLEASE UNCOMMENT LINE TO DELETE BUCKET. CONTENT FROM THE BUCKET NOT DELETED"
        )
    print(
        f"Size of the bucket {BUCKET_NAME} after deleting = {shell_output[0].split()[0]} bytes"
    )
Explanation: Delete contents from the staging bucket
NOTE: Everything in this Cloud Storage bucket will be DELETED. Please run it with caution.
End of explanation
gcr_images = !gcloud container images list --repository=gcr.io/$PROJECT_ID --filter="name~"$APP_NAME
if delete_image:
    for image in gcr_images:
        if image != "NAME":  # skip header line
            print(f"Deleting image {image} including all tags")
            !gcloud container images delete $image --force-delete-tags --quiet
Explanation: Delete images from Container Registry
Deletes all the container images created in this tutorial with prefix defined by variable APP_NAME from the registry. All associated tags are also deleted.
End of explanation
<END_TASK>