mirror of
https://github.com/GoogleCloudPlatform/vertex-ai-samples.git
synced 2026-09-26 14:42:04 +00:00
77 KiB
77 KiB
In [ ]:
# Copyright 2021 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 [ ]:
! pip3 install -U google-cloud-aiplatform --userIn [ ]:
! pip3 install google-cloud-storageIn [ ]:
import os
if not os.getenv("AUTORUN"):
# Automatically restart kernel after installs
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)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_IDIn [ ]:
REGION = "us-central1" # @param {type: "string"}In [ ]:
from datetime import datetime
TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")In [ ]:
import os
import sys
# If you are running this notebook in Colab, run this cell and follow the
# instructions to authenticate your Google Cloud account. This provides access
# to your Cloud Storage bucket and lets you submit training jobs and prediction
# requests.
# If on Vertex, then don't execute this code
if not os.path.exists("/opt/deeplearning/metadata/env_version"):
if "google.colab" in sys.modules:
from google.colab import auth as google_auth
google_auth.authenticate_user()
# If you are running this tutorial in a notebook locally, replace the string
# below with the path to your service account key and run this cell to
# authenticate your Google Cloud account.
else:
%env GOOGLE_APPLICATION_CREDENTIALS your_path_to_credentials.json
# Log in to your account on Google Cloud
! gcloud auth loginIn [ ]:
BUCKET_NAME = "[your-bucket-name]" # @param {type:"string"}In [ ]:
if BUCKET_NAME == "" or BUCKET_NAME is None or BUCKET_NAME == "[your-bucket-name]":
BUCKET_NAME = PROJECT_ID + "aip-" + TIMESTAMPIn [ ]:
! gcloud storage buckets create --location $REGION gs://$BUCKET_NAMEIn [ ]:
! gcloud storage ls --all-versions --long gs://$BUCKET_NAMEIn [ ]:
import time
from google.cloud.aiplatform import gapic as aip
from google.protobuf import json_format
from google.protobuf.json_format import MessageToJson, ParseDict
from google.protobuf.struct_pb2 import ValueIn [ ]:
# API Endpoint
API_ENDPOINT = "{}-aiplatform.googleapis.com".format(REGION)
# Vertex AI location root path for your dataset, model and endpoint resources
PARENT = "projects/" + PROJECT_ID + "/locations/" + REGIONIn [ ]:
# client options same for all services
client_options = {"api_endpoint": API_ENDPOINT}
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_prediction_client():
client = aip.PredictionServiceClient(client_options=client_options)
return client
def create_job_client():
client = aip.JobServiceClient(client_options=client_options)
return client
clients = {}
clients["model"] = create_model_client()
clients["endpoint"] = create_endpoint_client()
clients["prediction"] = create_prediction_client()
clients["job"] = create_job_client()
for client in clients.items():
print(client)In [ ]:
! rm -rf cifar
! mkdir cifar
! touch cifar/README.md
setup_cfg = "[egg_info]\n\
tag_build =\n\
tag_date = 0"
! echo "$setup_cfg" > cifar/setup.cfg
setup_py = "import setuptools\n\
# Requires TensorFlow Datasets\n\
setuptools.setup(\n\
install_requires=[\n\
'tensorflow_datasets==1.3.0',\n\
],\n\
packages=setuptools.find_packages())"
! echo "$setup_py" > cifar/setup.py
pkg_info = "Metadata-Version: 1.0\n\
Name: Custom Training CIFAR-10\n\
Version: 0.0.0\n\
Summary: Demonstration training script\n\
Home-page: www.google.com\n\
Author: Google\n\
Author-email: aferlitsch@google.com\n\
License: Public\n\
Description: Demo\n\
Platform: Vertex AI"
! echo "$pkg_info" > cifar/PKG-INFO
! mkdir cifar/trainer
! touch cifar/trainer/__init__.pyIn [ ]:
%%writefile cifar/trainer/task.py
import tensorflow_datasets as tfds
import tensorflow as tf
from tensorflow.python.client import device_lib
import argparse
import os
import sys
tfds.disable_progress_bar()
parser = argparse.ArgumentParser()
parser.add_argument('--model-dir', dest='model_dir',
default='/tmp/saved_model', type=str, help='Model dir.')
parser.add_argument('--lr', dest='lr',
default=0.01, type=float,
help='Learning rate.')
parser.add_argument('--epochs', dest='epochs',
default=10, type=int,
help='Number of epochs.')
parser.add_argument('--steps', dest='steps',
default=200, type=int,
help='Number of steps per epoch.')
parser.add_argument('--distribute', dest='distribute', type=str, default='single',
help='distributed training strategy')
args = parser.parse_args()
print('Python Version = {}'.format(sys.version))
print('TensorFlow Version = {}'.format(tf.__version__))
print('TF_CONFIG = {}'.format(os.environ.get('TF_CONFIG', 'Not found')))
print('DEVICES', device_lib.list_local_devices())
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")
elif args.distribute == 'mirror':
strategy = tf.distribute.MirroredStrategy()
elif args.distribute == 'multi':
strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
print('num_replicas_in_sync = {}'.format(strategy.num_replicas_in_sync))
BUFFER_SIZE = 10000
BATCH_SIZE = 64
def make_datasets_unbatched():
def scale(image, label):
image = tf.cast(image, tf.float32)
image /= 255.0
return image, label
datasets, info = tfds.load(name='cifar10',
with_info=True,
as_supervised=True)
return datasets['train'].map(scale).cache().shuffle(BUFFER_SIZE).repeat()
def build_and_compile_cnn_model():
model = tf.keras.Sequential([
tf.keras.layers.Conv2D(32, 3, activation='relu', input_shape=(32, 32, 3)),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Conv2D(32, 3, activation='relu'),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(10, activation='softmax')
])
model.compile(
loss=tf.keras.losses.sparse_categorical_crossentropy,
optimizer=tf.keras.optimizers.SGD(learning_rate=args.lr),
metrics=['accuracy'])
return model
NUM_WORKERS = strategy.num_replicas_in_sync
GLOBAL_BATCH_SIZE = BATCH_SIZE * NUM_WORKERS
train_dataset = make_datasets_unbatched().batch(GLOBAL_BATCH_SIZE)
with strategy.scope():
model = build_and_compile_cnn_model()
model.fit(x=train_dataset, epochs=args.epochs, steps_per_epoch=args.steps)
model.save(args.model_dir)
In [ ]:
! rm -f cifar.tar cifar.tar.gz
! tar cvf cifar.tar cifar
! gzip cifar.tar
! gcloud storage cp cifar.tar.gz gs://$BUCKET_NAME/trainer_cifar.tar.gzIn [ ]:
JOB_NAME = "custom_job_TF_" + TIMESTAMP
TRAIN_IMAGE = "gcr.io/cloud-aiplatform/training/tf-gpu.2-1:latest"
TRAIN_NGPU = 1
TRAIN_GPU = aip.AcceleratorType.NVIDIA_TESLA_K80
worker_pool_specs = [
{
"replica_count": 1,
"machine_spec": {
"machine_type": "n1-standard-4",
"accelerator_type": TRAIN_GPU,
"accelerator_count": TRAIN_NGPU,
},
"python_package_spec": {
"executor_image_uri": TRAIN_IMAGE,
"package_uris": ["gs://" + BUCKET_NAME + "/trainer_cifar.tar.gz"],
"python_module": "trainer.task",
"args": [
"--model-dir=" + "gs://{}/{}".format(BUCKET_NAME, JOB_NAME),
"--epochs=" + str(20),
"--steps=" + str(100),
"--distribute=" + "single",
],
},
}
]
training_job = {
"display_name": JOB_NAME,
"job_spec": {"worker_pool_specs": worker_pool_specs},
}
print(
MessageToJson(
aip.CreateCustomJobRequest(parent=PARENT, custom_job=training_job).__dict__[
"_pb"
]
)
)In [ ]:
request = clients["job"].create_custom_job(parent=PARENT, custom_job=training_job)In [ ]:
print(MessageToJson(request.__dict__["_pb"]))In [ ]:
# The full unique ID for the custom training job
custom_training_id = request.name
# The short numeric ID for the custom training job
custom_training_short_id = custom_training_id.split("/")[-1]
print(custom_training_id)In [ ]:
request = clients["job"].get_custom_job(name=custom_training_id)In [ ]:
print(MessageToJson(request.__dict__["_pb"]))In [ ]:
while True:
response = clients["job"].get_custom_job(name=custom_training_id)
if response.state != aip.PipelineState.PIPELINE_STATE_SUCCEEDED:
print("Training job has not completed:", response.state)
if response.state == aip.PipelineState.PIPELINE_STATE_FAILED:
break
else:
print("Training Time:", response.end_time - response.start_time)
break
time.sleep(20)
# model artifact output directory on Google Cloud Storage
model_artifact_dir = (
response.job_spec.worker_pool_specs[0].python_package_spec.args[0].split("=")[-1]
)
print("artifact location " + model_artifact_dir)In [ ]:
import tensorflow as tf
model = tf.keras.models.load_model(model_artifact_dir)In [ ]:
CONCRETE_INPUT = "numpy_inputs"
def _preprocess(bytes_input):
decoded = tf.io.decode_jpeg(bytes_input, channels=3)
decoded = tf.image.convert_image_dtype(decoded, tf.float32)
resized = tf.image.resize(decoded, size=(32, 32))
rescale = tf.cast(resized / 255.0, tf.float32)
return rescale
@tf.function(input_signature=[tf.TensorSpec([None], tf.string)])
def preprocess_fn(bytes_inputs):
decoded_images = tf.map_fn(
_preprocess, bytes_inputs, dtype=tf.float32, back_prop=False
)
return {
CONCRETE_INPUT: decoded_images
} # User needs to make sure the key matches model's input
m_call = tf.function(model.call).get_concrete_function(
[tf.TensorSpec(shape=[None, 32, 32, 3], dtype=tf.float32, name=CONCRETE_INPUT)]
)
@tf.function(input_signature=[tf.TensorSpec([None], tf.string)])
def serving_fn(bytes_inputs):
images = preprocess_fn(bytes_inputs)
prob = m_call(**images)
return prob
tf.saved_model.save(
model,
model_artifact_dir,
signatures={
"serving_default": serving_fn,
},
)In [ ]:
loaded = tf.saved_model.load(model_artifact_dir)
input_name = list(
loaded.signatures["serving_default"].structured_input_signature[1].keys()
)[0]
print("Serving function input:", input_name)In [ ]:
model = {
"display_name": "custom_job_TF" + TIMESTAMP,
"metadata_schema_uri": "",
"artifact_uri": model_artifact_dir,
"container_spec": {
"image_uri": "gcr.io/cloud-aiplatform/prediction/tf2-cpu.2-1:latest"
},
}
print(MessageToJson(aip.UploadModelRequest(parent=PARENT, model=model).__dict__["_pb"]))In [ ]:
request = clients["model"].upload_model(parent=PARENT, model=model)In [ ]:
result = request.result()
print(MessageToJson(result.__dict__["_pb"]))In [ ]:
# The full unique ID for the model
model_id = result.model
print(model_id)In [ ]:
import base64
import json
import cv2
import numpy as np
import tensorflow as tf
(_, _), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()
test_image_1, test_label_1 = x_test[0], y_test[0]
test_image_2, test_label_2 = x_test[1], y_test[1]
cv2.imwrite("tmp1.jpg", (test_image_1).astype(np.uint8))
cv2.imwrite("tmp2.jpg", (test_image_2).astype(np.uint8))
gcs_input_uri = "gs://" + BUCKET_NAME + "/" + "test.jsonl"
with tf.io.gfile.GFile(gcs_input_uri, "w") as f:
bytes = tf.io.read_file("tmp1.jpg")
b64str = base64.b64encode(bytes.numpy()).decode("utf-8")
f.write(json.dumps({input_name: {"b64": b64str}}) + "\n")
bytes = tf.io.read_file("tmp2.jpg")
b64str = base64.b64encode(bytes.numpy()).decode("utf-8")
f.write(json.dumps({input_name: {"b64": b64str}}) + "\n")
! gcloud storage cat $gcs_input_uriIn [ ]:
batch_prediction_job = aip.BatchPredictionJob(
display_name="custom_job_TF" + TIMESTAMP,
model=model_id,
input_config={
"instances_format": "jsonl",
"gcs_source": {"uris": [gcs_input_uri]},
},
model_parameters=ParseDict(
{"confidenceThreshold": 0.5, "maxPredictions": 2}, Value()
),
output_config={
"predictions_format": "jsonl",
"gcs_destination": {
"output_uri_prefix": "gs://" + f"{BUCKET_NAME}/batch_output/"
},
},
dedicated_resources={
"machine_spec": {"machine_type": "n1-standard-2", "accelerator_type": 0},
"starting_replica_count": 1,
"max_replica_count": 1,
},
)
print(
MessageToJson(
aip.CreateBatchPredictionJobRequest(
parent=PARENT, batch_prediction_job=batch_prediction_job
).__dict__["_pb"]
)
)In [ ]:
request = clients["job"].create_batch_prediction_job(
parent=PARENT, batch_prediction_job=batch_prediction_job
)In [ ]:
print(MessageToJson(request.__dict__["_pb"]))In [ ]:
# The fully qualified ID for the batch job
batch_job_id = request.name
# The short numeric ID for the batch job
batch_job_short_id = batch_job_id.split("/")[-1]
print(batch_job_id)In [ ]:
request = clients["job"].get_batch_prediction_job(name=batch_job_id)In [ ]:
print(MessageToJson(request.__dict__["_pb"]))In [ ]:
def get_latest_predictions(gcs_out_dir):
""" Get the latest prediction subfolder using the timestamp in the subfolder name"""
folders = !gcloud storage ls $gcs_out_dir
latest = ""
for folder in folders:
subfolder = folder.split("/")[-2]
if subfolder.startswith("prediction-"):
if subfolder > latest:
latest = folder[:-1]
return latest
while True:
response = clients["job"].get_batch_prediction_job(name=batch_job_id)
if response.state != aip.JobState.JOB_STATE_SUCCEEDED:
print("The job has not completed:", response.state)
if response.state == aip.JobState.JOB_STATE_FAILED:
break
else:
folder = get_latest_predictions(
response.output_config.gcs_destination.output_uri_prefix
)
! gcloud storage ls $folder/prediction*
! gcloud storage cat $folder/prediction*
break
time.sleep(60)Warning:
Output truncated. This notebook contains too many cells to display efficiently.