mirror of
https://github.com/GoogleCloudPlatform/vertex-ai-samples.git
synced 2026-09-26 14:42:04 +00:00
* Reduced boilerplate Ran linter and fixed install Added pyarrow Fixed pip install Reverted unneeded changes Fixed pip install Linted code and added missing preprocessor call Fixed preprocessor Default to us-central1 * Ran linter * Simplified service account section
49 KiB
49 KiB
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.In [ ]:
# Install the packages
! pip3 install --upgrade google-cloud-aiplatform \
google-cloud-storage \
google-cloud-bigquery \
pyarrowIn [ ]:
# Automatically restart kernel after installs so that your environment can access the new packages
# import IPython
# app = IPython.Application.instance()
# app.kernel.do_shutdown(True)In [ ]:
PROJECT_ID = "[your-project-id]" # @param {type:"string"}
# Set the project id
! gcloud config set project {PROJECT_ID}In [ ]:
REGION = "us-central1" # @param {type: "string"}In [ ]:
# ! gcloud auth loginIn [ ]:
# from google.colab import auth
# auth.authenticate_user()In [ ]:
BUCKET_URI = "gs://your-bucket-name-unique" # @param {type:"string"}In [ ]:
! gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URIIn [ ]:
from google.cloud import aiplatform, bigqueryIn [ ]:
# Initialize the Vertex AI SDK
aiplatform.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)In [ ]:
# Set up BigQuery client
bqclient = bigquery.Client(project=PROJECT_ID)In [ ]:
TRAIN_VERSION = "tf-cpu.2-8"
DEPLOY_VERSION = "tf2-cpu.2-8"
TRAIN_IMAGE = "us-docker.pkg.dev/vertex-ai/training/{}:latest".format(TRAIN_VERSION)
DEPLOY_IMAGE = "us-docker.pkg.dev/vertex-ai/prediction/{}:latest".format(DEPLOY_VERSION)In [ ]:
import json
import numpy as np
# Calculate mean and std across all rows
# Define the BigQuery source dataset
BQ_SOURCE = "bq://bigquery-public-data.ml_datasets.penguins"
# Define NA values
NA_VALUES = ["NA", "."]
# Download a table
def download_table(bq_table_uri: str):
# Remove bq:// prefix if present
prefix = "bq://"
if bq_table_uri.startswith(prefix):
bq_table_uri = bq_table_uri[len(prefix) :]
table = bigquery.TableReference.from_string(bq_table_uri)
rows = bqclient.list_rows(
table,
)
return rows.to_dataframe()
# Remove NA values
def clean_dataframe(df):
return df.replace(to_replace=NA_VALUES, value=np.NaN).dropna()
def calculate_mean_and_std(df):
# Calculate mean and std for each applicable column
mean_and_std = {}
dtypes = list(zip(df.dtypes.index, map(str, df.dtypes)))
# Normalize numeric columns.
for column, dtype in dtypes:
if dtype == "float32" or dtype == "float64":
mean_and_std[column] = {
"mean": df[column].mean(),
"std": df[column].std(),
}
return mean_and_std
dataframe = download_table(BQ_SOURCE)
dataframe = clean_dataframe(dataframe)
mean_and_std = calculate_mean_and_std(dataframe)
print("The mean and stds for each column are: " + str(mean_and_std))
# Write to a file
MEAN_AND_STD_JSON_FILE = "mean_and_std.json"
with open(MEAN_AND_STD_JSON_FILE, "w") as outfile:
json.dump(mean_and_std, outfile)
# Save to the staging bucket
! gsutil cp {MEAN_AND_STD_JSON_FILE} {BUCKET_URI}In [ ]:
dataset = aiplatform.TabularDataset.create(
display_name="sample-penguins", bq_source=BQ_SOURCE
)In [ ]:
JOB_NAME = "custom_job_unique"
TRAIN_STRATEGY = "single"
EPOCHS = 20
BATCH_SIZE = 10
CMDARGS = [
"--epochs=" + str(EPOCHS),
"--batch_size=" + str(BATCH_SIZE),
"--distribute=" + TRAIN_STRATEGY,
"--mean_and_std_json_file=" + f"{BUCKET_URI}/{MEAN_AND_STD_JSON_FILE}",
]In [ ]:
%%writefile task.py
import argparse
import tensorflow as tf
import numpy as np
import os
import pandas as pd
import tensorflow as tf
from google.cloud import bigquery
from google.cloud import storage
# Read environmental variables
training_data_uri = os.getenv("AIP_TRAINING_DATA_URI")
validation_data_uri = os.getenv("AIP_VALIDATION_DATA_URI")
test_data_uri = os.getenv("AIP_TEST_DATA_URI")
# Read args
parser = argparse.ArgumentParser()
parser.add_argument('--epochs', dest='epochs',
default=10, type=int,
help='Number of epochs.')
parser.add_argument('--batch_size', dest='batch_size',
default=10, type=int,
help='Batch size.')
parser.add_argument('--distribute', dest='distribute', type=str, default='single',
help='Distributed training strategy.')
parser.add_argument('--mean_and_std_json_file', dest='mean_and_std_json_file', type=str,
help='GCS URI to the JSON file with pre-calculated column means and standard deviations.')
args = parser.parse_args()
def download_blob(bucket_name, source_blob_name, destination_file_name):
"""Downloads a blob from the bucket."""
# bucket_name = "your-bucket-name"
# source_blob_name = "storage-object-name"
# destination_file_name = "local/path/to/file"
storage_client = storage.Client()
bucket = storage_client.bucket(bucket_name)
# Construct a client side representation of a blob.
# Note `Bucket.blob` differs from `Bucket.get_blob` as it doesn't retrieve
# any content from Cloud Storage. As we don't need additional data,
# using `Bucket.blob` is preferred here.
blob = bucket.blob(source_blob_name)
blob.download_to_filename(destination_file_name)
print(
"Blob {} downloaded to {}.".format(
source_blob_name, destination_file_name
)
)
def extract_bucket_and_prefix_from_gcs_path(gcs_path: str):
"""Given a complete GCS path, return the bucket name and prefix as a tuple.
Example Usage:
bucket, prefix = extract_bucket_and_prefix_from_gcs_path(
"gs://example-bucket/path/to/folder"
)
# bucket = "example-bucket"
# prefix = "path/to/folder"
Args:
gcs_path (str):
Required. A full path to a Cloud Storage folder or resource.
Can optionally include "gs://" prefix or end in a trailing slash "/".
Returns:
Tuple[str, Optional[str]]
A (bucket, prefix) pair from provided GCS path. If a prefix is not
present, None is returned in its place.
"""
if gcs_path.startswith("gs://"):
gcs_path = gcs_path[5:]
if gcs_path.endswith("/"):
gcs_path = gcs_path[:-1]
gcs_parts = gcs_path.split("/", 1)
gcs_bucket = gcs_parts[0]
gcs_blob_prefix = None if len(gcs_parts) == 1 else gcs_parts[1]
return (gcs_bucket, gcs_blob_prefix)
# Download means and std
def download_mean_and_std(mean_and_std_json_file):
"""Download mean and std for each column"""
import json
bucket, file_path = extract_bucket_and_prefix_from_gcs_path(mean_and_std_json_file)
download_blob(bucket_name=bucket, source_blob_name=file_path, destination_file_name=file_path)
with open(file_path, 'r') as file:
return json.loads(file.read())
mean_and_std = download_mean_and_std(args.mean_and_std_json_file)
# Single Machine, single compute device
if args.distribute == 'single':
if tf.test.is_gpu_available():
strategy = tf.distribute.OneDeviceStrategy(device="/gpu:0")
else:
strategy = tf.distribute.OneDeviceStrategy(device="/cpu:0")
# Single Machine, multiple compute device
elif args.distribute == 'mirror':
strategy = tf.distribute.MirroredStrategy()
# Multiple Machine, multiple compute device
elif args.distribute == 'multi':
strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
# Set up training variables
LABEL_COLUMN = "species"
UNUSED_COLUMNS = []
NA_VALUES = ["NA", "."]
# Possible categorical values
SPECIES = ['Adelie Penguin (Pygoscelis adeliae)',
'Chinstrap penguin (Pygoscelis antarctica)',
'Gentoo penguin (Pygoscelis papua)']
ISLANDS = ['Dream', 'Biscoe', 'Torgersen']
SEXES = ['FEMALE', 'MALE']
# Set up BigQuery clients
bqclient = bigquery.Client()
# Download a table
def download_table(bq_table_uri: str):
# Remove bq:// prefix if present
prefix = "bq://"
if bq_table_uri.startswith(prefix):
bq_table_uri = bq_table_uri[len(prefix):]
table = bigquery.TableReference.from_string(bq_table_uri)
rows = bqclient.list_rows(
table,
)
return rows.to_dataframe(create_bqstorage_client=False)
df_train = download_table(training_data_uri)
df_validation = download_table(validation_data_uri)
df_test = download_table(test_data_uri)
# Remove NA values
def clean_dataframe(df):
return df.replace(to_replace=NA_VALUES, value=np.NaN).dropna()
df_train = clean_dataframe(df_train)
df_validation = clean_dataframe(df_validation)
_CATEGORICAL_TYPES = {
"island": pd.api.types.CategoricalDtype(categories=ISLANDS),
"species": pd.api.types.CategoricalDtype(categories=SPECIES),
"sex": pd.api.types.CategoricalDtype(categories=SEXES),
}
def standardize(df, mean_and_std):
"""Scales numerical columns using their means and standard deviation to get
z-scores: the mean of each numerical column becomes 0, and the standard
deviation becomes 1. This can help the model converge during training.
Args:
df: Pandas df
Returns:
Input df with the numerical columns scaled to z-scores
"""
dtypes = list(zip(df.dtypes.index, map(str, df.dtypes)))
# Normalize numeric columns.
for column, dtype in dtypes:
if dtype == "float32":
df[column] -= mean_and_std[column]["mean"]
df[column] /= mean_and_std[column]["std"]
return df
def preprocess(df):
"""Converts categorical features to numeric. Removes unused columns.
Args:
df: Pandas df with raw data
Returns:
df with preprocessed data
"""
df = df.drop(columns=UNUSED_COLUMNS)
# Drop rows with NaN's
df = df.dropna()
# Convert integer valued (numeric) columns to floating point
numeric_columns = df.select_dtypes(["int32", "float32", "float64"]).columns
df[numeric_columns] = df[numeric_columns].astype("float32")
# Convert categorical columns to numeric
cat_columns = df.select_dtypes(["object"]).columns
df[cat_columns] = df[cat_columns].apply(
lambda x: x.astype(_CATEGORICAL_TYPES[x.name])
)
df[cat_columns] = df[cat_columns].apply(lambda x: x.cat.codes)
return df
def convert_dataframe_to_dataset(
df_train,
df_validation,
mean_and_std
):
df_train = preprocess(df_train)
df_validation = preprocess(df_validation)
df_train_x, df_train_y = df_train, df_train.pop(LABEL_COLUMN)
df_validation_x, df_validation_y = df_validation, df_validation.pop(LABEL_COLUMN)
# Join train_x and eval_x to normalize on overall means and standard
# deviations. Then separate them again.
all_x = pd.concat([df_train_x, df_validation_x], keys=["train", "eval"])
all_x = standardize(all_x, mean_and_std)
df_train_x, df_validation_x = all_x.xs("train"), all_x.xs("eval")
y_train = np.asarray(df_train_y).astype("float32")
y_validation = np.asarray(df_validation_y).astype("float32")
# Convert to numpy representation
x_train = np.asarray(df_train_x)
x_test = np.asarray(df_validation_x)
# Convert to one-hot representation
y_train = tf.keras.utils.to_categorical(y_train, num_classes=len(SPECIES))
y_validation = tf.keras.utils.to_categorical(y_validation, num_classes=len(SPECIES))
dataset_train = tf.data.Dataset.from_tensor_slices((x_train, y_train))
dataset_validation = tf.data.Dataset.from_tensor_slices((x_test, y_validation))
return (dataset_train, dataset_validation)
# Create datasets
dataset_train, dataset_validation = convert_dataframe_to_dataset(df_train, df_validation, mean_and_std)
# Shuffle train set
dataset_train = dataset_train.shuffle(len(df_train))
def create_model(num_features):
# Create model
Dense = tf.keras.layers.Dense
model = tf.keras.Sequential(
[
Dense(
100,
activation=tf.nn.relu,
kernel_initializer="uniform",
input_dim=num_features,
),
Dense(75, activation=tf.nn.relu),
Dense(50, activation=tf.nn.relu),
Dense(25, activation=tf.nn.relu),
Dense(3, activation=tf.nn.softmax),
]
)
# Compile Keras model
optimizer = tf.keras.optimizers.RMSprop(lr=0.001)
model.compile(
loss="categorical_crossentropy", metrics=["accuracy"], optimizer=optimizer
)
return model
# Create the model
with strategy.scope():
model = create_model(num_features=dataset_train._flat_shapes[0].dims[0].value)
# Set up datasets
NUM_WORKERS = strategy.num_replicas_in_sync
# Here the batch size scales up by number of workers since
# `tf.data.Dataset.batch` expects the global batch size.
GLOBAL_BATCH_SIZE = args.batch_size * NUM_WORKERS
dataset_train = dataset_train.batch(GLOBAL_BATCH_SIZE)
dataset_validation = dataset_validation.batch(GLOBAL_BATCH_SIZE)
# Train the model
model.fit(dataset_train, epochs=args.epochs, validation_data=dataset_validation)
tf.saved_model.save(model, os.getenv("AIP_MODEL_DIR"))
df_test.head()In [ ]:
job = aiplatform.CustomTrainingJob(
display_name=JOB_NAME,
script_path="task.py",
container_uri=TRAIN_IMAGE,
requirements=["google-cloud-bigquery>=2.20.0", "db-dtypes"],
model_serving_container_image_uri=DEPLOY_IMAGE,
)
MODEL_DISPLAY_NAME = "penguins_model_unique"
# Start the training
model = job.run(
dataset=dataset,
model_display_name=MODEL_DISPLAY_NAME,
bigquery_destination=f"bq://{PROJECT_ID}",
args=CMDARGS,
)In [ ]:
DEPLOYED_NAME = "penguins_deployed_unique"
endpoint = model.deploy(deployed_model_display_name=DEPLOYED_NAME)In [ ]:
import pandas as pd
from google.cloud import bigquery
UNUSED_COLUMNS = []
LABEL_COLUMN = "species"
# Possible categorical values
SPECIES = [
"Adelie Penguin (Pygoscelis adeliae)",
"Chinstrap penguin (Pygoscelis antarctica)",
"Gentoo penguin (Pygoscelis papua)",
]
ISLANDS = ["Dream", "Biscoe", "Torgersen"]
SEXES = ["FEMALE", "MALE"]
_CATEGORICAL_TYPES = {
"island": pd.api.types.CategoricalDtype(categories=ISLANDS),
"species": pd.api.types.CategoricalDtype(categories=SPECIES),
"sex": pd.api.types.CategoricalDtype(categories=SEXES),
}
def standardize(df, mean_and_std):
"""Scales numerical columns using their means and standard deviation to get
z-scores: the mean of each numerical column becomes 0, and the standard
deviation becomes 1. This can help the model converge during training.
Args:
df: Pandas df
Returns:
Input df with the numerical columns scaled to z-scores
"""
dtypes = list(zip(df.dtypes.index, map(str, df.dtypes)))
# Normalize numeric columns.
for column, dtype in dtypes:
if dtype == "float32":
df[column] -= mean_and_std[column]["mean"]
df[column] /= mean_and_std[column]["std"]
return df
def preprocess(df, mean_and_std):
"""Converts categorical features to numeric. Removes unused columns.
Args:
df: Pandas df with raw data
Returns:
df with preprocessed data
"""
df = df.drop(columns=UNUSED_COLUMNS)
# Drop rows with NaN's
df = df.dropna()
# Convert integer valued (numeric) columns to floating point
numeric_columns = df.select_dtypes(["int32", "float32", "float64"]).columns
df[numeric_columns] = df[numeric_columns].astype("float32")
# Convert categorical columns to numeric
cat_columns = df.select_dtypes(["object"]).columns
df[cat_columns] = df[cat_columns].apply(
lambda x: x.astype(_CATEGORICAL_TYPES[x.name])
)
df[cat_columns] = df[cat_columns].apply(lambda x: x.cat.codes)
return df
def convert_dataframe_to_list(df, mean_and_std):
df = preprocess(df, mean_and_std)
df_x, df_y = df, df.pop(LABEL_COLUMN)
# Normalize on overall means and standard deviations.
df = standardize(df, mean_and_std)
y = np.asarray(df_y).astype("float32")
# Convert to numpy representation
x = np.asarray(df_x)
# Convert to one-hot representation
return x.tolist(), y.tolist()
x_test, y_test = convert_dataframe_to_list(dataframe, mean_and_std)In [ ]:
predictions = endpoint.predict(instances=x_test)
y_predicted = np.argmax(predictions.predictions, axis=1)
correct = sum(y_predicted == np.array(y_test))
accuracy = len(y_predicted)
print(
f"Correct predictions = {correct}, Total predictions = {accuracy}, Accuracy = {correct/accuracy}"
)In [ ]:
endpoint.undeploy_all()In [ ]:
import os
# Delete the training job
job.delete()
# Delete the model
model.delete()
# Delete the endpoint
endpoint.delete()
# Warning: Setting this to true deletes everything in your bucket
delete_bucket = False
if delete_bucket or os.getenv("IS_TESTING"):
! gsutil rm -r $BUCKET_URI
Run in Colab
View on GitHub