model_garden/notebooks/official/automl/automl_tabular_on_vertex_pipelines.ipynb

In [ ]:

# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Tabular Workflow: AutoML Tabular Pipeline

Run in Colab

View on GitHub

Open in Vertex AI Workbench

Overview

In this tutorial, you use two Vertex AI Tabular Workflows pipelines to train AutoML models using different configurations. You see how get_automl_tabular_pipeline_and_parameters gives you the ability to customize the default AutoML Tabular pipeline, and how get_skip_architecture_search_pipeline_and_parameters allows you to reduce the training time and cost for an AutoML model by using the tuning results from a previous pipeline run.

Objective

In this tutorial, you learn how to create two regression models using Vertex Pipelines downloaded from Google Cloud Pipeline Components (GCPC). These pipelines will be Vertex AI Tabular Workflow pipelines which are maintained by Google. These pipelines showcase different ways to customize the Vertex Tabular training process.

The steps performed are:

• Create a training pipeline that reduces the search space from the default to save time.
• Create a training pipeline that reuses the architecture search results from the previous pipeline to save time.

Dataset

The dataset you will be using is Bank Marketing. The data is for direct marketing campaigns (phone calls) of a Portuguese banking institution. The binary classification goal is to predict if a client subscribe a term deposit. For this notebook, you randomly selected 90% of the rows in the original dataset and saved them in a train.csv file hosted on Cloud Storage. To download the file, click here.

Costs

This tutorial uses billable components of Google Cloud:

• Vertex AI
• Cloud Storage

Learn about Vertex AI pricing and Cloud Storage pricing, and use the Pricing Calculator to generate a cost estimate based on your projected usage.

Set up your local development environment

If you are using Colab or Vertex AI Workbench 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 Cloud Storage SDK
• Python 3
• virtualenv
• Jupyter notebook running in a virtual environment with Python 3

The Cloud Storage 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:

1. Install and initialize the SDK.

2. Install Python 3.

3. Install virtualenv and create a virtual environment that uses Python 3. Activate the virtual environment.

4. To install Jupyter, run pip3 install jupyter on the command-line in a terminal shell.

5. To launch Jupyter, run jupyter notebook on the command-line in a terminal shell.

6. Open this notebook in the Jupyter Notebook Dashboard.

Install additional packages

Install the latest version of the Google Cloud Pipeline Components (GCPC) SDK.

In [ ]:

!pip install -U google-cloud-pipeline-components==1.0.25 -q

Restart the kernel

Once you've installed the additional packages, you need to restart the notebook kernel so it can find the packages.

Note: Once this cell has finished running, continue on. You do not need to re-run any of the cells above.

In [ ]:

import os

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

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

Before you begin

GPU runtime

This tutorial does not require a GPU runtime.

Set up your Google Cloud project

The following steps are required, regardless of your notebook environment.

1. Select or create a Google Cloud project. When you first create an account, you get a $300 free credit towards your compute/storage costs.

2. Make sure that billing is enabled for your project.

3. Enable the following APIs: Vertex AI APIs, Dataflow APIs, Compute Engine APIs, and Cloud Storage.

4. If you are running this notebook locally, you need to install the Cloud SDK.

5. 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 $.

Notes about service account and permission

By default no configuration is required, if you run into any permission related issue, please make sure the service accounts above have the required roles:

Service account email	Description	Roles
PROJECT_NUMBER-compute@developer.gserviceaccount.com	Compute Engine default service account	Dataflow Developer, Dataflow Worker, Storage Admin, BigQuery Data Editor, Vertex AI User, Service Account User
service-PROJECT_NUMBER@gcp-sa-aiplatform.iam.gserviceaccount.com	AI Platform Service Agent	Vertex AI Service Agent

1. Goto https://console.cloud.google.com/iam-admin/iam.
2. Check the "Include Google-provided role grants" checkbox.
3. Find the above emails.
4. Grant the corresponding roles.

Using data source from a different project

• For the BQ data source, grant both service accounts the "BigQuery Data Viewer" role.
• For the CSV data source, grant both service accounts the "Storage Object Viewer" role.

Set your project ID

Set your project ID below. If you know know your project ID, leave the field blank and the following cells may be able to find it. Optionally, you may also set a service account in the cell below.

In [ ]:

PROJECT_ID = "[your-project-id]"  # @param {type:"string"}

In [ ]:

if PROJECT_ID == "" or PROJECT_ID is None or PROJECT_ID == "[your-project-id]":
    # Get your GCP project id from gcloud
    shell_output = !gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID:", PROJECT_ID)

In [ ]:

! gcloud config set project $PROJECT_ID

Region

You may change the REGION variable, which is used for Vertex Forecasting operations throughout the rest of this notebook. Below are regions supported for Vertex AI. We recommend that you choose the region closest to you.

• Americas: us-central1
• Europe: europe-west4
• Asia Pacific: asia-east1

Learn more about Vertex AI regions

In [ ]:

REGION = "[your-region]"  # @param {type: "string"}

if REGION == "[your-region]":
    REGION = "us-central1"

Authenticate your Google Cloud account

If you are using Vertex AI Workbench Notebooks, your environment is already authenticated.

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" into the filter box, and select Vertex 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.

In [ ]:

# 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.

import os
import sys

# If on Vertex AI Workbench, then don't execute this code
IS_COLAB = "google.colab" in sys.modules
if not os.path.exists("/opt/deeplearning/metadata/env_version") and not os.getenv(
    "DL_ANACONDA_HOME"
):
    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 '[your-service-account-key-path]'

Create a Cloud Storage bucket

The following steps are required, regardless of your notebook environment.

All training related files (TF model checkpoint, TensorBoard file, etc) saved to the GCS bucket. The pipeline not clean up the files since some of them might be useful for you, please make sure to clean up the files. For easy cleanup, you can set GCS bucket level TTL.

Set the name of your Cloud Storage bucket below. Bucket names must be globally unique across all Google Cloud projects, including those outside of your organization.

In [ ]:

BUCKET_URI = "gs://[your-bucket-name]"  # @param {type:"string"}
GENERATE_BUCKET_URI = True  # @param {type:"boolean"}

Create the bucket if it doesn't already exist.

In [ ]:

import uuid

if GENERATE_BUCKET_URI:
    bucket_name = "gs://test-{}".format(uuid.uuid4())
    !gsutil mb -p {PROJECT_ID} -l {REGION} {bucket_name}

    # set GCS bucket object TTL to 7 days
    !echo '{"rule":[{"action": {"type": "Delete"},"condition": {"age": 7}}]}' > gcs_lifecycle.tmp
    !gsutil lifecycle set gcs_lifecycle.tmp {bucket_name}
    !rm gcs_lifecycle.tmp

    BUCKET_URI = bucket_name
    print(f"changed BUCKET_URI to {BUCKET_URI} due to GENERATE_BUCKET_URI is True")

if BUCKET_URI == "" or BUCKET_URI is None or BUCKET_URI == "gs://[your-bucket-name]":
    BUCKET_URI = "gs://" + PROJECT_ID + "aip-" + uuid.uuid4()

! gsutil ls -b $BUCKET_URI || gsutil mb -l $DATA_REGION $BUCKET_URI

Finally, validate access to your Cloud Storage bucket by examining its contents:

In [ ]:

! gsutil ls -al $BUCKET_URI

Service Account

You use a service account to create Vertex AI Pipeline jobs. If you do not want to use your project's Compute Engine service account, set SERVICE_ACCOUNT to another service account ID.

In [ ]:

SERVICE_ACCOUNT = "[your-service-account]"  # @param {type:"string"}

In [ ]:

if (
    SERVICE_ACCOUNT == ""
    or SERVICE_ACCOUNT is None
    or SERVICE_ACCOUNT == "[your-service-account]"
):
    # Get your service account from gcloud
    if not IS_COLAB:
        shell_output = !gcloud auth list 2>/dev/null
        SERVICE_ACCOUNT = shell_output[2].replace("*", "").strip()

    else:  # IS_COLAB:
        shell_output = ! gcloud projects describe  $PROJECT_ID
        project_number = shell_output[-1].split(":")[1].strip().replace("'", "")
        SERVICE_ACCOUNT = f"{project_number}-compute@developer.gserviceaccount.com"

    print("Service Account:", SERVICE_ACCOUNT)

Set service account access for Vertex AI Pipelines

Run the following commands to grant your service account access to read and write pipeline artifacts in the bucket that you created in the previous step. You only need to run this step once per service account.

In [ ]:

! gsutil iam ch serviceAccount:{SERVICE_ACCOUNT}:roles/storage.objectCreator $BUCKET_URI

! gsutil iam ch serviceAccount:{SERVICE_ACCOUNT}:roles/storage.objectViewer $BUCKET_URI

Import libraries and define constants

In [ ]:

# Import required modules
import json
from typing import Any, Dict, List

from google.cloud import aiplatform, storage
from google_cloud_pipeline_components.experimental.automl.tabular import \
    utils as automl_tabular_utils

Initialize Vertex SDK for Python

Initialize the Vertex SDK for Python for your project.

In [ ]:

aiplatform.init(project=PROJECT_ID, location=REGION)

Define helper functions

In [ ]:

#Get the bucket name and path.
def get_bucket_name_and_path(uri):
    no_prefix_uri = uri[len("gs://") :]
    splits = no_prefix_uri.split("/")
    return splits[0], "/".join(splits[1:])

#Download from the bucket.
def download_from_gcs(uri):
    bucket_name, path = get_bucket_name_and_path(uri)
    storage_client = storage.Client(project=PROJECT_ID)
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(path)
    return blob.download_as_string()

#Write content in to the bucket.
def write_to_gcs(uri: str, content: str):
    bucket_name, path = get_bucket_name_and_path(uri)
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(path)
    blob.upload_from_string(content)

#Generate auto transformations.
def generate_auto_transformation(column_names: List[str]) -> List[Dict[str, Any]]:
    transformations = []
    for column_name in column_names:
        transformations.append({"auto": {"column_name": column_name}})
    return transformations

#Write auto transformations in to the bucket.
def write_auto_transformations(uri: str, column_names: List[str]):
    transformations = generate_auto_transformation(column_names)
    write_to_gcs(uri, json.dumps(transformations))

#Get the task details filter with task name.
def get_task_detail(
    task_details: List[Dict[str, Any]], task_name: str
) -> List[Dict[str, Any]]:
    for task_detail in task_details:
        if task_detail.task_name == task_name:
            return task_detail

#Get the deployed model uri from task details.
def get_deployed_model_uri(
    task_details,
):
    ensemble_task = get_task_detail(task_details, "model-upload")
    return ensemble_task.outputs["model"].artifacts[0].uri

#Get the model uri from the task details.
def get_no_custom_ops_model_uri(task_details):
    ensemble_task = get_task_detail(task_details, "automl-tabular-ensemble")
    return download_from_gcs(
        ensemble_task.outputs["model_without_custom_ops"].artifacts[0].uri
    )

#Get the feature attributions from task details.
def get_feature_attributions(
    task_details,
):
    ensemble_task = get_task_detail(task_details, "model-evaluation-2")
    return download_from_gcs(
        ensemble_task.outputs["evaluation_metrics"]
        .artifacts[0]
        .metadata["explanation_gcs_path"]
    )

#Get the evaluation metrics from task details.
def get_evaluation_metrics(
    task_details,
):
    ensemble_task = get_task_detail(task_details, "model-evaluation")
    return download_from_gcs(
        ensemble_task.outputs["evaluation_metrics"].artifacts[0].uri
    )

#Print the parsed json.
def load_and_print_json(s):
    parsed = json.loads(s)
    print(json.dumps(parsed, indent=2, sort_keys=True))

Define training specification

In [ ]:

run_evaluation = True  # @param {type:"boolean"}
run_distillation = False  # @param {type:"boolean"}
root_dir = os.path.join(BUCKET_URI, "automl_tabular_pipeline")
prediction_type = "classification"
optimization_objective = "minimize-log-loss"
target_column = "deposit"
data_source_csv_filenames = (
    "gs://cloud-samples-data/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv"
)
data_source_bigquery_table_path = None  # format: bq://bq_project.bq_dataset.bq_table

timestamp_split_key = None  # timestamp column name when using timestamp split
stratified_split_key = None  # target column name when using stratified split
training_fraction = 0.8
validation_fraction = 0.1
test_fraction = 0.1

predefined_split_key = None
if predefined_split_key:
    training_fraction = None
    validation_fraction = None
    test_fraction = None

weight_column = None

features = [
    "age",
    "job",
    "marital",
    "education",
    "default",
    "balance",
    "housing",
    "loan",
    "contact",
    "day",
    "month",
    "duration",
    "campaign",
    "pdays",
    "previous",
    "poutcome",
]
transformations = generate_auto_transformation(features)
transform_config_path = os.path.join(root_dir, f"transform_config_{uuid.uuid4()}.json")
write_to_gcs(transform_config_path, json.dumps(transformations))

VPC related config

If you need to use a custom Dataflow subnetwork, you can set it through the dataflow_subnetwork parameter. The requirements are:

1. dataflow_subnetwork must be fully qualified subnetwork name. [reference]
2. The following service accounts must have Compute Network User role assigned on the specified dataflow subnetwork [reference]:
   1. Compute Engine default service account: PROJECT_NUMBER-compute@developer.gserviceaccount.com
   2. Dataflow service account: service-PROJECT_NUMBER@dataflow-service-producer-prod.iam.gserviceaccount.com

If your project has VPC-SC enabled, please make sure:

1. The dataflow subnetwork used in VPC-SC is configured properly for Dataflow. [reference]
2. dataflow_use_public_ips is set to False.

In [ ]:

# Dataflow's fully qualified subnetwork name, when empty the default subnetwork will be used.
# Fully qualified subnetwork name is in the form of
# https://www.googleapis.com/compute/v1/projects/HOST_PROJECT_ID/regions/REGION_NAME/subnetworks/SUBNETWORK_NAME
# reference: https://cloud.google.com/dataflow/docs/guides/specifying-networks#example_network_and_subnetwork_specifications
dataflow_subnetwork = None  # @param {type:"string"}
# Specifies whether Dataflow workers use public IP addresses.
dataflow_use_public_ips = True  # @param {type:"boolean"}

Customize search space and change training configuration

you create a skip evaluation AutoML Tables pipeline with the following customizations:

• Limit the hyperparameter search space
• Change machine type and tuning / training parallelism

In [ ]:

study_spec_parameters_override = [
    {
        "parameter_id": "model_type",
        "categorical_value_spec": {
            "values": [
                "nn"
            ]  # The default value is ["nn", "boosted_trees"], this reduces the search space
        },
    }
]

worker_pool_specs_override = [
    {"machine_spec": {"machine_type": "n1-standard-8"}},  # override for TF chief node
    {},  # override for TF worker node, since it's not used, leave it empty
    {},  # override for TF ps node, since it's not used, leave it empty
    {
        "machine_spec": {
            "machine_type": "n1-standard-4"  # override for TF evaluator node
        }
    },
]

# Number of weak models in the final ensemble model is
# stage_2_num_selected_trials * 5. If unspecified, 5 is the default value for
# stage_2_num_selected_trials.
stage_2_num_selected_trials = 5

# The pipeline output a TF saved model contains the following TF custom op:
# - https://github.com/google/struct2tensor
#
# There are a few ways to run the model:
# - Official prediction server docker image
#   Please follow the "Run the model server" section in
#   https://cloud.google.com/vertex-ai/docs/export/export-model-tabular#run-server
# - Python or cpp runtimes like TF serving
#   Please set export_additional_model_without_custom_ops so the pipeline
#   outputs an additional model does does not depend on struct2tensor.
#   - `get_no_custom_ops_model_uri` shows how to get the model artifact URI.
#   - The input to the model is a dictionary of feature name to tensor. Use
#     `saved_model_cli show --dir {saved_model.pb's path} --signature_def serving_default --tag serve`
#     to find out more details.
export_additional_model_without_custom_ops = False

train_budget_milli_node_hours = 1000  # 1 hour

(
    template_path,
    parameter_values,
) = automl_tabular_utils.get_automl_tabular_pipeline_and_parameters(
    PROJECT_ID,
    REGION,
    root_dir,
    target_column,
    prediction_type,
    optimization_objective,
    transform_config_path,
    train_budget_milli_node_hours,
    data_source_csv_filenames=data_source_csv_filenames,
    data_source_bigquery_table_path=data_source_bigquery_table_path,
    weight_column=weight_column,
    predefined_split_key=predefined_split_key,
    timestamp_split_key=timestamp_split_key,
    stratified_split_key=stratified_split_key,
    training_fraction=training_fraction,
    validation_fraction=validation_fraction,
    test_fraction=test_fraction,
    study_spec_parameters_override=study_spec_parameters_override,
    stage_1_tuner_worker_pool_specs_override=worker_pool_specs_override,
    cv_trainer_worker_pool_specs_override=worker_pool_specs_override,
    run_evaluation=run_evaluation,
    run_distillation=run_distillation,
    dataflow_subnetwork=dataflow_subnetwork,
    dataflow_use_public_ips=dataflow_use_public_ips,
    export_additional_model_without_custom_ops=export_additional_model_without_custom_ops,
)

job_id = "automl-tabular-{}".format(uuid.uuid4())
job = aiplatform.PipelineJob(
    display_name=job_id,
    location=REGION,  # launches the pipeline job in the specified region
    template_path=template_path,
    job_id=job_id,
    pipeline_root=root_dir,
    parameter_values=parameter_values,
    enable_caching=False,
)

job.run(service_account=SERVICE_ACCOUNT)

pipeline_task_details = job.gca_resource.job_detail.task_details

if export_additional_model_without_custom_ops:
    print(
        "trained model without custom TF ops:",
        get_no_custom_ops_model_uri(pipeline_task_details),
    )

if run_evaluation:
    print("evaluation metrics:")
    load_and_print_json(get_evaluation_metrics(pipeline_task_details))

    print("feature attributions:")
    load_and_print_json(get_feature_attributions(pipeline_task_details))

Skip architecture search

Instead of doing architecture search everytime, you can reuse the existing architecture search result. This could help:

1. reducing the variation of the output model
2. reducing training cost

The existing architecture search result is stored in the tuning_result_output output of the automl-tabular-stage-1-tuner component. you can manually input it or get it programmatically.

In [ ]:

stage_1_tuner_task = get_task_detail(
    pipeline_task_details, "automl-tabular-stage-1-tuner"
)
stage_1_tuning_result_artifact_uri = (
    stage_1_tuner_task.outputs["tuning_result_output"].artifacts[0].uri
)

Run the skip architecture search pipeline

In [ ]:

(
    template_path,
    parameter_values,
) = automl_tabular_utils.get_skip_architecture_search_pipeline_and_parameters(
    PROJECT_ID,
    REGION,
    root_dir,
    target_column,
    prediction_type,
    optimization_objective,
    transform_config_path,
    train_budget_milli_node_hours,
    data_source_csv_filenames=data_source_csv_filenames,
    data_source_bigquery_table_path=data_source_bigquery_table_path,
    weight_column=weight_column,
    predefined_split_key=predefined_split_key,
    timestamp_split_key=timestamp_split_key,
    stratified_split_key=stratified_split_key,
    training_fraction=training_fraction,
    validation_fraction=validation_fraction,
    test_fraction=test_fraction,
    stage_1_tuning_result_artifact_uri=stage_1_tuning_result_artifact_uri,
    run_evaluation=run_evaluation,
    dataflow_subnetwork=dataflow_subnetwork,
    dataflow_use_public_ips=dataflow_use_public_ips,
)

job_id = "automl-tabular-skip-architecture-search-{}".format(uuid.uuid4())
job = aiplatform.PipelineJob(
    display_name=job_id,
    location=REGION,  # launches the pipeline job in the specified region
    template_path=template_path,
    job_id=job_id,
    pipeline_root=root_dir,
    parameter_values=parameter_values,
    enable_caching=False,
)

job.run(service_account=SERVICE_ACCOUNT)

# Get model URI
skip_architecture_search_pipeline_task_details = (
    job.gca_resource.job_detail.task_details
)

if export_additional_model_without_custom_ops:
    print(
        "trained model without custom TF ops:",
        get_no_custom_ops_model_uri(pipeline_task_details),
    )

Clean up Vertex and BigQuery resources

To clean up all Google Cloud resources used in this project, you can delete the Google Cloud project you used for the tutorial.

Otherwise, you can delete the individual resources you created in this tutorial:

• Cloud Storage Bucket

In [ ]:

if os.getenv("IS_TESTING"):
    ! gsutil rm -r $BUCKET_URI