xiaohei/model_garden
1
0
Fork 0
mirror of https://github.com/GoogleCloudPlatform/vertex-ai-samples.git synced 2026-09-26 14:42:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add batch and TPU (#262)

Browse Source 
* feat: friday update

* feat: friday update

* feat: hpt notebook

* feat: hpt notebook

* feat: sklearn

* feat: sklearn

* feat: xgb training

* feat: xgb training

* fix: predict on exported BQML

* fix: predict on exported BQML

* feat: add Pytorch notebook

* feat: add Pytorch notebook

* feat: add R notebook

* feat: add R notebook

* fix: spelling

* fix: spelling

* feat: add batch and TPU

* feat: add batch and TPU
This commit is contained in:
Andrew Ferlitsch
2022-02-02 17:45:31 -08:00
committed by GitHub
parent 0f3e257773
commit a73335c0af
3 changed files with 567 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
notebooks/community/ml_ops/stage2/get_started_vertex_distributed_training.ipynb
+229 -1
View File
@@ -90,7 +90,8 @@
"- `MirroredStrategy`: Train on a single VM with multiple GPUs.\n",
"- `MultiWorkerMirroredStrategy`: Train on multiple VMs with automatic setup of replicas.\n",
"- `MultiWorkerMirroredStrategy`: Train on multiple VMs with fine grain control of replicas.\n",
"- `ReductionServer`: Train on multiple VMS and sync updates across VMS with Vertex AI Reduction Server"
"- `ReductionServer`: Train on multiple VMS and sync updates across VMS with `Vertex AI Reduction Server`.\n",
"- `TPUTraining`: Train with multiple Cloud TPUs."
]
},
{
@@ -1546,6 +1547,233 @@
"job.delete()"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "tpu_intro"
},
"source": [
"## Cloud TPU Training\n",
"\n",
"To further speed up trainig, your organization can utilize Google's Cloud Tensor Processing Units (TPU) pods.\n",
"\n",
"Cloud TPU is the custom-designed machine learning ASIC that powers Google products like Translate, Photos, Search, Assistant, and Gmail. Cloud TPU is designed to run cutting-edge machine learning models with AI services on Google Cloud. And its custom high-speed network offers over 100 petaflops of performance in a single pod.\n",
"\n",
"Learn more about [Cloud TPU](https://cloud.google.com/tpu)\n",
"\n",
"*Note*: TPU VM Training is currently an opt-in feature. Your GCP project must first be added to the feature allowlist. Please email your project information(project id/number) to vertex-ai-tpu-vm-training-support@google.com for the allowlist. You will receive an email as soon as your project is ready."
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "docker_write:tpu"
},
"source": [
"### Write Docker file for TPU training\n",
"\n",
"Currently, there is no pre-built Vertex AI Docker image for training with TPUs. No problems, you can make your own, as follows:\n",
"\n",
"1. Create a vanilla Python 3 image (e.g., `python3:8`).\n",
"2. Get and install the TPU library (`libtpu.so`).\n",
"3. Copy in your training package"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "docker_write:tpu"
},
"outputs": [],
"source": [
"%%writefile custom/Dockerfile\n",
"FROM python:3.8\n",
"\n",
"WORKDIR /root\n",
"\n",
"# Copies the trainer code to the docker image.\n",
"COPY trainer /trainer\n",
"\n",
"RUN pip3 install tensorflow-datasets\n",
"\n",
"# Install TPU Tensorflow and dependencies.\n",
"# libtpu.so must be under the '/lib' directory.\n",
"RUN wget https://storage.googleapis.com/cloud-tpu-tpuvm-artifacts/libtpu/20210525/libtpu.so -O /lib/libtpu.so\n",
"RUN chmod 777 /lib/libtpu.so\n",
"\n",
"RUN wget https://storage.googleapis.com/cloud-tpu-tpuvm-artifacts/tensorflow/20210525/tf_nightly-2.6.0-cp38-cp38-linux_x86_64.whl\n",
"RUN pip3 install tf_nightly-2.6.0-cp38-cp38-linux_x86_64.whl\n",
"RUN rm tf_nightly-2.6.0-cp38-cp38-linux_x86_64.whl\n",
"# Sets up the entry point to invoke the trainer.\n",
"ENTRYPOINT [\"python\", \"-m\", \"trainer.task\"]"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "docker_push:tpu"
},
"source": [
"### Build and push the Docker image to the Artifact Registry"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "docker_push:tpu"
},
"outputs": [],
"source": [
"TRAIN_IMAGE = f\"gcr.io/\" + PROJECT_ID + \"/tpu-train:latest\"\n",
"\n",
"os.chdir(\"custom\")\n",
"! docker build --quiet --tag={TRAIN_IMAGE} .\n",
"! docker push {TRAIN_IMAGE}\n",
"os.chdir(\"..\")"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "worker_pool_tpu"
},
"source": [
"### TPU worker specification pool\n",
"\n",
"Next, you create the worker specification pool. For TPUs, you do:\n",
"\n",
"- Create only one worker pool (Primary).\n",
"- Set the machine type to `cloud-tpu`.\n",
"- Set the accelerator type to a `TPU`."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "worker_pool_tpu"
},
"outputs": [],
"source": [
"# Use TPU Accelerators. Temporarily using numeric codes, until types are added to the SDK\n",
"# 6 = TPU_V2\n",
"# 7 = TPU_V3\n",
"TRAIN_TPU, TRAIN_NTPU = (7, 8)\n",
"TRAIN_COMPUTE = \"cloud-tpu\"\n",
"\n",
"\n",
"if not TRAIN_NTPU or TRAIN_NTPU < 2:\n",
" TRAIN_STRATEGY = \"single\"\n",
"else:\n",
" TRAIN_STRATEGY = \"tpu\"\n",
"print(TRAIN_STRATEGY)\n",
"\n",
"EPOCHS = 20\n",
"STEPS = 10000\n",
"\n",
"TRAINER_ARGS = [\n",
" \"--epochs=\" + str(EPOCHS),\n",
" \"--steps=\" + str(STEPS),\n",
" \"--distribute=\" + TRAIN_STRATEGY,\n",
"]\n",
"\n",
"\n",
"WORKER_POOL_SPECS = [\n",
" {\n",
" \"container_spec\": {\n",
" \"args\": TRAINER_ARGS,\n",
" \"image_uri\": TRAIN_IMAGE,\n",
" },\n",
" \"replica_count\": 1,\n",
" \"machine_spec\": {\n",
" \"machine_type\": TRAIN_COMPUTE,\n",
" \"accelerator_type\": TRAIN_TPU,\n",
" \"accelerator_count\": TRAIN_NTPU,\n",
" },\n",
" }\n",
"]\n",
"\n",
"print(WORKER_POOL_SPECS[0])"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "custom_job:worker_pool"
},
"source": [
"### Create CustomJob with worker pool specifications\n",
"\n",
"Next, you create a `CustomJob` for the multi-worker distributed training job:\n",
"\n",
"-`display_name`: The display name for the custom job.\n",
"\n",
"-`worker_pool_specs`: The detailed specifications for each worker pool."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "custom_job:worker_pool"
},
"outputs": [],
"source": [
"DISPLAY_NAME = \"boston_\" + TIMESTAMP\n",
"\n",
"job = aip.CustomJob(display_name=DISPLAY_NAME, worker_pool_specs=WORKER_POOL_SPECS)"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "run_custom_job:multiworker"
},
"source": [
"### Run the CustomJob\n",
"\n",
"Next, you run the custom job."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "run_custom_job:multiworker"
},
"outputs": [],
"source": [
"try:\n",
" job.run(sync=True)\n",
"except Exception as e:\n",
" # may fail in multi-worker to find startup script\n",
" print(e)"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "delete_job"
},
"source": [
"### Delete a custom training job\n",
"\n",
"After a training job is completed, you can delete the training job with the method `delete()`. Prior to completion, a training job can be canceled with the method `cancel()`."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "delete_job"
},
"outputs": [],
"source": [
"job.delete()"
]
},
{
"cell_type": "markdown",
"metadata": {
notebooks/community/ml_ops/stage3/get_started_with_automl_pipeline_components.ipynb
+162 -12
View File
@@ -86,10 +86,14 @@
"- `Vertex AI AutoML`\n",
"- `Google Cloud Pipeline Components`\n",
"- `Vertex AI Dataset, Model and Endpoint` resources\n",
"- `Vertex AI Prediction`\n",
"\n",
"The steps performed include:\n",
"\n",
"- Construct a pipeline for training and deploying a Vertex AI AutoML model.\n",
"- Construct a pipeline for:\n",
" - Training a Vertex AI AutoML trained model.\n",
" - Test the serving binary with a batch prediction job.\n",
" - Deploying a Vertex AI AutoML trained model.\n",
"- Execute a Vertex AI pipeline."
]
},
@@ -125,7 +129,9 @@
" ! pip3 install --upgrade apache-beam[gcp] $USER_FLAG\n",
" ! pip3 install --upgrade pyarrow $USER_FLAG\n",
" ! pip3 install --upgrade cloudml-hypertune $USER_FLAG\n",
" ! pip3 install --upgrade kfp $USER_FLAG"
" ! pip3 install --upgrade kfp $USER_FLAG\n",
" ! pip3 install --upgrade torchvision $USER_FLAG\n",
" ! pip3 install --upgrade rpy2 $USER_FLAG"
]
},
{
@@ -375,7 +381,7 @@
"):\n",
" # Get your GCP project id from gcloud\n",
" shell_output = !gcloud auth list 2>/dev/null\n",
" SERVICE_ACCOUNT = shell_output[2].strip()\n",
" SERVICE_ACCOUNT = shell_output[2].replace(\"*\", \"\").strip()\n",
" print(\"Service Account:\", SERVICE_ACCOUNT)"
]
},
@@ -569,15 +575,25 @@
" - The display name for the dataset is passed into the pipeline.\n",
" - The import file for the dataset is passed into the pipeline.\n",
" - The component returns the dataset resource as `outputs[\"dataset\"]`\n",
"\n",
"\n",
"2. Use the prebuilt component `AutoMLImageTrainingJobRunOp` to train a Vertex AI AutoML Model resource, where:\n",
" - The display name for the dataset is passed into the pipeline.\n",
" - The dataset is the output from the `ImageDatasetCreateOp`.\n",
" - The component returns the model resource as `outputs[\"model\"]`.\n",
"3. Use the prebuilt component `EndpointCreateOp` to create a Vertex AI Endpoint to deploy the trained model to, where:\n",
"\n",
"\n",
"3. Use the prebuild component `ModelBatchPredictOp` to do a test batch prediction, where:\n",
" - The model is the output from the `AutoMLTrainingJobRunOp`.\n",
"\n",
"\n",
"4. Use the prebuilt component `EndpointCreateOp` to create a Vertex AI Endpoint to deploy the trained model to, where:\n",
" - Since the component has no dependencies on other components, by default it would be executed in parallel with the model training.\n",
" - The `after(training_op)` is added to serialize its execution, so its only executed if the training operation completes successfully.\n",
" - The component returns the endpoint resource as `outputs[\"endpoint\"]`.\n",
"4. Use the prebuilt component `ModelDeployOp` to deploy the trained AutoML model to, where:\n",
"\n",
"\n",
"5. Use the prebuilt component `ModelDeployOp` to deploy the trained AutoML model to, where:\n",
" - The display name for the dataset is passed into the pipeline.\n",
" - The model is the output from the `AutoMLTrainingJobRunOp`.\n",
" - The endpoint is the output from the `EndpointCreateOp`\n",
@@ -596,13 +612,19 @@
"from google_cloud_pipeline_components import aiplatform as gcc_aip\n",
"\n",
"PIPELINE_ROOT = \"{}/pipeline_root/automl_icn_training\".format(BUCKET_NAME)\n",
"DEPLOY_COMPUTE = \"n1-standard-4\"\n",
"\n",
"\n",
"@dsl.pipeline(\n",
" name=\"automl-icn-training\", description=\"AutoML image classification training\"\n",
")\n",
"def pipeline(\n",
" import_file: str, display_name: str, project: str = PROJECT_ID, region: str = REGION\n",
" import_file: str,\n",
" batch_files: list,\n",
" display_name: str,\n",
" bucket: str = PIPELINE_ROOT,\n",
" project: str = PROJECT_ID,\n",
" region: str = REGION,\n",
"):\n",
"\n",
" dataset_op = gcc_aip.ImageDatasetCreateOp(\n",
@@ -628,11 +650,25 @@
"\n",
" eval_op = evaluateAutoMLModelOp(model=training_op.outputs[\"model\"], region=region)\n",
"\n",
" batch_op = gcc_aip.ModelBatchPredictOp(\n",
" project=project,\n",
" job_display_name=\"batch_predict_job\",\n",
" model=training_op.outputs[\"model\"],\n",
" gcs_source_uris=batch_files,\n",
" gcs_destination_output_uri_prefix=bucket,\n",
" instances_format=\"jsonl\",\n",
" predictions_format=\"jsonl\",\n",
" model_parameters={},\n",
" machine_type=DEPLOY_COMPUTE,\n",
" starting_replica_count=1,\n",
" max_replica_count=1,\n",
" ).after(eval_op)\n",
"\n",
" endpoint_op = gcc_aip.EndpointCreateOp(\n",
" project=project,\n",
" location=region,\n",
" display_name=display_name,\n",
" ).after(eval_op)\n",
" ).after(batch_op)\n",
"\n",
" deploy_op = gcc_aip.ModelDeployOp(\n",
" model=training_op.outputs[\"model\"],\n",
@@ -642,6 +678,107 @@
" )"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "get_test_items:batch_prediction"
},
"source": [
"### Get test item(s)\n",
"\n",
"Now do a batch prediction to your Vertex model. You will use arbitrary examples out of the dataset as a test items. Don't be concerned that the examples were likely used in training the model -- we just want to demonstrate how to make a prediction."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "get_test_items:automl,icn,csv"
},
"outputs": [],
"source": [
"test_items = !gsutil cat $IMPORT_FILE | head -n2\n",
"if len(str(test_items[0]).split(\",\")) == 3:\n",
" _, test_item_1, test_label_1 = str(test_items[0]).split(\",\")\n",
" _, test_item_2, test_label_2 = str(test_items[1]).split(\",\")\n",
"else:\n",
" test_item_1, test_label_1 = str(test_items[0]).split(\",\")\n",
" test_item_2, test_label_2 = str(test_items[1]).split(\",\")\n",
"\n",
"print(test_item_1, test_label_1)\n",
"print(test_item_2, test_label_2)"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "copy_test_items:batch_prediction"
},
"source": [
"### Copy test item(s)\n",
"\n",
"For the batch prediction, copy the test items over to your Cloud Storage bucket."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "copy_test_items:batch_prediction"
},
"outputs": [],
"source": [
"file_1 = test_item_1.split(\"/\")[-1]\n",
"file_2 = test_item_2.split(\"/\")[-1]\n",
"\n",
"! gsutil cp $test_item_1 $BUCKET_NAME/$file_1\n",
"! gsutil cp $test_item_2 $BUCKET_NAME/$file_2\n",
"\n",
"test_item_1 = BUCKET_NAME + \"/\" + file_1\n",
"test_item_2 = BUCKET_NAME + \"/\" + file_2"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "make_batch_file:automl,image"
},
"source": [
"### Make the batch input file\n",
"\n",
"Now make a batch input file, which you will store in your local Cloud Storage bucket. The batch input file can be either CSV or JSONL. You will use JSONL in this tutorial. For JSONL file, you make one dictionary entry per line for each data item (instance). The dictionary contains the key/value pairs:\n",
"\n",
"- `content`: The Cloud Storage path to the image.\n",
"- `mime_type`: The content type. In our example, it is a `jpeg` file.\n",
"\n",
"For example:\n",
"\n",
" {'content': '[your-bucket]/file1.jpg', 'mime_type': 'jpeg'}"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "make_batch_file:automl,image"
},
"outputs": [],
"source": [
"import json\n",
"\n",
"import tensorflow as tf\n",
"\n",
"gcs_input_uri = BUCKET_NAME + \"/test.jsonl\"\n",
"with tf.io.gfile.GFile(gcs_input_uri, \"w\") as f:\n",
" data = {\"content\": test_item_1, \"mime_type\": \"image/jpeg\"}\n",
" f.write(json.dumps(data) + \"\\n\")\n",
" data = {\"content\": test_item_2, \"mime_type\": \"image/jpeg\"}\n",
" f.write(json.dumps(data) + \"\\n\")\n",
"\n",
"print(gcs_input_uri)\n",
"! gsutil cat $gcs_input_uri"
]
},
{
"cell_type": "markdown",
"metadata": {
@@ -653,6 +790,7 @@
"Next, you compile the pipeline and then exeute it. The pipeline takes the following parameters, which are passed as the dictionary `parameter_values`:\n",
"\n",
"- `import_file`: The Cloud Storage path to the dataset index file.\n",
"- `batch_files`: A list of Cloud Storage paths to the input batch files.\n",
"- `display_name`: The display name for the generated Vertex AI resources.\n",
"- `project`: The project ID.\n",
"- `region`: The region."
@@ -676,6 +814,7 @@
" pipeline_root=PIPELINE_ROOT,\n",
" parameter_values={\n",
" \"import_file\": IMPORT_FILE,\n",
" \"batch_files\": [gcs_input_uri],\n",
" \"display_name\": \"flowers\" + TIMESTAMP,\n",
" \"project\": PROJECT_ID,\n",
" \"region\": REGION,\n",
@@ -751,18 +890,29 @@
"\n",
"print(\"imagedataset-create\")\n",
"artifacts = print_pipeline_output(pipeline, \"imagedataset-create\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"automlimagetrainingjob-run\")\n",
"artifacts = print_pipeline_output(pipeline, \"automlimagetrainingjob-run\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"endpoint-create\")\n",
"artifacts = print_pipeline_output(pipeline, \"endpoint-create\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"model-deploy\")\n",
"artifacts = print_pipeline_output(pipeline, \"model-deploy\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"evaluateautomlmodelop\")\n",
"artifacts = print_pipeline_output(pipeline, \"evaluateautomlmodelop\")"
"artifacts = print_pipeline_output(pipeline, \"evaluateautomlmodelop\")\n",
"print(\"\\n\\n\")\n",
"print(\"model-batch-predict\")\n",
"artifacts = print_pipeline_output(pipeline, \"model-batch-predict\")\n",
"output = !gsutil cat $artifacts\n",
"output = json.loads(output[0])\n",
"print(\"\\n\\n\")\n",
"print(\n",
" output[\"artifacts\"][\"batchpredictionjob\"][\"artifacts\"][0][\"metadata\"][\n",
" \"gcsOutputDirectory\"\n",
" ]\n",
")"
]
},
{
notebooks/community/ml_ops/stage3/get_started_with_custom_training_pipeline_components.ipynb
+176 -13
View File
@@ -86,10 +86,14 @@
"- `Vertex AI Training`\n",
"- `Google Cloud Pipeline Components`\n",
"- `Vertex AI Dataset, Model and Endpoint` resources\n",
"- `Vertex AI Prediction`\n",
"\n",
"The steps performed include:\n",
"\n",
"- Construct a pipeline for training and deploying a Vertex AI custom trained model.\n",
"- Construct a pipeline for:\n",
" - Training a Vertex AI custom trained model.\n",
" - Test the serving binary with a batch prediction job.\n",
" - Deploying a Vertex AI custom trained model.\n",
"- Execute a Vertex AI pipeline."
]
},
@@ -125,7 +129,9 @@
" ! pip3 install --upgrade apache-beam[gcp] $USER_FLAG\n",
" ! pip3 install --upgrade pyarrow $USER_FLAG\n",
" ! pip3 install --upgrade cloudml-hypertune $USER_FLAG\n",
" ! pip3 install --upgrade kfp $USER_FLAG"
" ! pip3 install --upgrade kfp $USER_FLAG\n",
" ! pip3 install --upgrade torchvision $USER_FLAG\n",
" ! pip3 install --upgrade rpy2 $USER_FLAG"
]
},
{
@@ -375,7 +381,7 @@
"):\n",
" # Get your GCP project id from gcloud\n",
" shell_output = !gcloud auth list 2>/dev/null\n",
" SERVICE_ACCOUNT = shell_output[2].strip()\n",
" SERVICE_ACCOUNT = shell_output[2].replace(\"*\", \"\").strip()\n",
" print(\"Service Account:\", SERVICE_ACCOUNT)"
]
},
@@ -768,6 +774,7 @@
"import json\n",
"import logging\n",
"import tqdm\n",
"import hypertune as hpt\n",
"\n",
"def parse_args():\n",
" parser = argparse.ArgumentParser(description=\"TF.Keras Image Classification\")\n",
@@ -794,7 +801,9 @@
" parser.add_argument(\n",
" \"--lr\", dest=\"lr\", default=0.01, type=float, help=\"Learning rate.\"\n",
" )\n",
" parser.add_argument(\"--batch-size\", default=16, type=int, help=\"mini-batch size\")\n",
" parser.add_argument(\n",
" \"--batch-size\", dest=\"batch_size\", default=16, type=int, help=\"mini-batch size\"\n",
" )\n",
" parser.add_argument(\n",
" \"--epochs\", default=10, type=int, help=\"number of training epochs\"\n",
" )\n",
@@ -813,6 +822,14 @@
" help=\"distributed training strategy\",\n",
" )\n",
"\n",
" parser.add_argument(\n",
" \"--tuning\",\n",
" dest=\"tuning\",\n",
" type=bool,\n",
" default=False,\n",
" help=\"hyperparameter tuning\"\n",
" )\n",
"\n",
" args = parser.parse_args()\n",
" return args\n",
"\n",
@@ -938,8 +955,19 @@
"def train_model(model, train_dataset, val_dataset):\n",
" logging.info(\"Start model training\")\n",
" history = model.fit(\n",
" x=train_dataset, epochs=args.epochs, validation_data=val_dataset, steps_per_epoch=args.steps\n",
" x=train_dataset, epochs=args.epochs, steps_per_epoch=args.steps, batch_size=args.batch_size, validation_data=val_dataset\n",
" )\n",
"\n",
" if args.tuning:\n",
" hp_metric = history.history['val_accuracy'][-1]\n",
"\n",
" hpt = hypertune.HyperTune()\n",
" hpt.report_hyperparameter_tuning_metric(\n",
" hyperparameter_metric_tag='accuracy',\n",
" metric_value=hp_metric,\n",
" global_step=args.epochs\n",
" )\n",
"\n",
" return history\n",
"\n",
"num_classes, train_dataset, val_dataset = get_data()\n",
@@ -1021,13 +1049,17 @@
" - The component returns the model resource as `outputs[\"model\"]`.\n",
"\n",
"\n",
"4. Use the prebuilt component `EndpointCreateOp` to create a Vertex AI Endpoint to deploy the trained model to, where:\n",
"4. Use the prebuild component `ModelBatchPredictOp` to do a test batch prediction, where:\n",
" - The model is the output from the `CustomPythonPackageTrainingJobRunOp`.\n",
"\n",
"\n",
"5. Use the prebuilt component `EndpointCreateOp` to create a Vertex AI Endpoint to deploy the trained model to, where:\n",
" - Since the component has no dependencies on other components, by default it would be executed in parallel with the model training.\n",
" - The `after(training_op)` is added to serialize its execution, so its only executed if the training operation completes successfully.\n",
" - The component returns the endpoint resource as `outputs[\"endpoint\"]`.\n",
"\n",
"\n",
"5. Use the prebuilt component `ModelDeployOp` to deploy the trained Vertex AI model to, where:\n",
"6. Use the prebuilt component `ModelDeployOp` to deploy the trained Vertex AI model to, where:\n",
" - The display name for the dataset is passed into the pipeline.\n",
" - The model is the output from the `CustomPythonPackageTrainingJobRunOp`.\n",
" - The endpoint is the output from the `EndpointCreateOp`\n",
@@ -1046,6 +1078,7 @@
"from google_cloud_pipeline_components import aiplatform as gcc_aip\n",
"\n",
"PIPELINE_ROOT = \"{}/pipeline_root/custom_icn_training\".format(BUCKET_NAME)\n",
"DEPLOY_COMPUTE = \"n1-standard-4\"\n",
"\n",
"\n",
"@dsl.pipeline(\n",
@@ -1055,8 +1088,10 @@
"def pipeline(\n",
" import_file: str,\n",
" display_name: str,\n",
" batch_files: list,\n",
" python_package: str,\n",
" python_module: str,\n",
" bucket: str = PIPELINE_ROOT,\n",
" project: str = PROJECT_ID,\n",
" region: str = REGION,\n",
"):\n",
@@ -1088,21 +1123,136 @@
" model_display_name=display_name,\n",
" )\n",
"\n",
" batch_op = gcc_aip.ModelBatchPredictOp(\n",
" project=project,\n",
" job_display_name=\"batch_predict_job\",\n",
" model=training_op.outputs[\"model\"],\n",
" gcs_source_uris=batch_files,\n",
" gcs_destination_output_uri_prefix=bucket,\n",
" instances_format=\"jsonl\",\n",
" predictions_format=\"jsonl\",\n",
" model_parameters={},\n",
" machine_type=DEPLOY_COMPUTE,\n",
" starting_replica_count=1,\n",
" max_replica_count=1,\n",
" )\n",
"\n",
" endpoint_op = gcc_aip.EndpointCreateOp(\n",
" project=project,\n",
" location=region,\n",
" display_name=display_name,\n",
" ).after(training_op)\n",
" ).after(batch_op)\n",
"\n",
" deploy_op = gcc_aip.ModelDeployOp(\n",
" model=training_op.outputs[\"model\"],\n",
" endpoint=endpoint_op.outputs[\"endpoint\"],\n",
" dedicated_resources_min_replica_count=1,\n",
" dedicated_resources_max_replica_count=1,\n",
" dedicated_resources_machine_type=\"n1-standard-4\",\n",
" dedicated_resources_machine_type=DEPLOY_COMPUTE,\n",
" )"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "get_test_items:batch_prediction"
},
"source": [
"### Get test item(s)\n",
"\n",
"Now do a batch prediction to your Vertex model. You will use arbitrary examples out of the dataset as a test items. Don't be concerned that the examples were likely used in training the model -- we just want to demonstrate how to make a prediction."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "get_test_items:automl,icn,csv"
},
"outputs": [],
"source": [
"test_items = !gsutil cat $IMPORT_FILE | head -n2\n",
"if len(str(test_items[0]).split(\",\")) == 3:\n",
" _, test_item_1, test_label_1 = str(test_items[0]).split(\",\")\n",
" _, test_item_2, test_label_2 = str(test_items[1]).split(\",\")\n",
"else:\n",
" test_item_1, test_label_1 = str(test_items[0]).split(\",\")\n",
" test_item_2, test_label_2 = str(test_items[1]).split(\",\")\n",
"\n",
"print(test_item_1, test_label_1)\n",
"print(test_item_2, test_label_2)"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "copy_test_items:batch_prediction"
},
"source": [
"### Copy test item(s)\n",
"\n",
"For the batch prediction, copy the test items over to your Cloud Storage bucket."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "copy_test_items:batch_prediction"
},
"outputs": [],
"source": [
"file_1 = test_item_1.split(\"/\")[-1]\n",
"file_2 = test_item_2.split(\"/\")[-1]\n",
"\n",
"! gsutil cp $test_item_1 $BUCKET_NAME/$file_1\n",
"! gsutil cp $test_item_2 $BUCKET_NAME/$file_2\n",
"\n",
"test_item_1 = BUCKET_NAME + \"/\" + file_1\n",
"test_item_2 = BUCKET_NAME + \"/\" + file_2"
]
},
{
"cell_type": "markdown",
"metadata": {
"id": "make_batch_file:automl,image"
},
"source": [
"### Make the batch input file\n",
"\n",
"Now make a batch input file, which you will store in your local Cloud Storage bucket. The batch input file can be either CSV or JSONL. You will use JSONL in this tutorial. For JSONL file, you make one dictionary entry per line for each data item (instance). The dictionary contains the key/value pairs:\n",
"\n",
"- `content`: The Cloud Storage path to the image.\n",
"- `mime_type`: The content type. In our example, it is a `jpeg` file.\n",
"\n",
"For example:\n",
"\n",
" {'content': '[your-bucket]/file1.jpg', 'mime_type': 'jpeg'}"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {
"id": "make_batch_file:automl,image"
},
"outputs": [],
"source": [
"import json\n",
"\n",
"import tensorflow as tf\n",
"\n",
"gcs_input_uri = BUCKET_NAME + \"/test.jsonl\"\n",
"with tf.io.gfile.GFile(gcs_input_uri, \"w\") as f:\n",
" data = {\"content\": test_item_1, \"mime_type\": \"image/jpeg\"}\n",
" f.write(json.dumps(data) + \"\\n\")\n",
" data = {\"content\": test_item_2, \"mime_type\": \"image/jpeg\"}\n",
" f.write(json.dumps(data) + \"\\n\")\n",
"\n",
"print(gcs_input_uri)\n",
"! gsutil cat $gcs_input_uri"
]
},
{
"cell_type": "markdown",
"metadata": {
@@ -1114,6 +1264,7 @@
"Next, you compile the pipeline and then exeute it. The pipeline takes the following parameters, which are passed as the dictionary `parameter_values`:\n",
"\n",
"- `import_file`: The Cloud Storage path to the dataset index file.\n",
"- `batch_files`: A list of Cloud Storage paths to the input batch files.\n",
"- `display_name`: The display name for the generated Vertex AI resources.\n",
"- `python_package`: The Python package for the custom training job.\n",
"- `python_module`: The Python module in the package to execute.\n",
@@ -1139,6 +1290,7 @@
" pipeline_root=PIPELINE_ROOT,\n",
" parameter_values={\n",
" \"import_file\": IMPORT_FILE,\n",
" \"batch_files\": [gcs_input_uri],\n",
" \"display_name\": \"flowers\" + TIMESTAMP,\n",
" \"python_package\": f\"{BUCKET_NAME}/trainer_flowers.tar.gz\",\n",
" \"python_module\": \"trainer.task\",\n",
@@ -1216,15 +1368,26 @@
"\n",
"print(\"imagedataset-create\")\n",
"artifacts = print_pipeline_output(pipeline, \"imagedataset-create\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"custompythonpackagetrainingjob-run\")\n",
"artifacts = print_pipeline_output(pipeline, \"custompythonpackagetrainingjob-run\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"endpoint-create\")\n",
"artifacts = print_pipeline_output(pipeline, \"endpoint-create\")\n",
"print(\"\\n\")\n",
"print(\"\\n\\n\")\n",
"print(\"model-deploy\")\n",
"artifacts = print_pipeline_output(pipeline, \"model-deploy\")"
"artifacts = print_pipeline_output(pipeline, \"model-deploy\")\n",
"print(\"\\n\\n\")\n",
"print(\"model-batch-predict\")\n",
"artifacts = print_pipeline_output(pipeline, \"model-batch-predict\")\n",
"output = !gsutil cat $artifacts\n",
"output = json.loads(output[0])\n",
"print(\"\\n\\n\")\n",
"print(\n",
" output[\"artifacts\"][\"batchpredictionjob\"][\"artifacts\"][0][\"metadata\"][\n",
" \"gcsOutputDirectory\"\n",
" ]\n",
")"
]
},
{