05-continuous-training.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "5043bfb9",
   "metadata": {},
   "source": [
    "# 05 - Continuous Training\n",
    "\n",
    "After testing, compiling, and uploading the pipeline definition to Cloud Storage, the pipeline is executed with respect to a trigger. We use [Cloud Functions](https://cloud.google.com/functions) and [Cloud Pub/Sub](https://cloud.google.com/pubsub) as a triggering mechanism. The triggering can be scheduled using [Cloud Scheduler](https://cloud.google.com/scheduler). The trigger source sends a message to a Cloud Pub/Sub topic that the Cloud Function listens to, and then it submits the pipeline to AI Platform Managed Pipelines to be executed.\n",
    "\n",
    "This notebook covers the following steps:\n",
    "1. Create the Cloud Pub/Sub topic.\n",
    "2. Deploy the Cloud Function \n",
    "3. Test triggering a pipeline.\n",
    "4. Extracting pipeline run metadata."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6f2f9013",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a0e71c08",
   "metadata": {},
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "37cefa26",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import os\n",
    "import logging\n",
    "import tensorflow as tf\n",
    "import tfx\n",
    "import IPython \n",
    "\n",
    "logging.getLogger().setLevel(logging.INFO)\n",
    "\n",
    "print(\"Tensorflow Version:\", tfx.__version__)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ab763d6d",
   "metadata": {},
   "source": [
    "### Setup Google Cloud project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5260d069",
   "metadata": {},
   "outputs": [],
   "source": [
    "PROJECT = '[your-project-id]' # Change to your project id.\n",
    "REGION = 'us-central1' # Change to your region.\n",
    "BUCKET =  '[your-bucket-name]' # Change to your bucket name.\n",
    "\n",
    "if PROJECT == \"\" or PROJECT is None or PROJECT == \"[your-project-id]\":\n",
    "    # Get your GCP project id from gcloud\n",
    "    shell_output = !gcloud config list --format 'value(core.project)' 2>/dev/null\n",
    "    PROJECT = shell_output[0]\n",
    "    \n",
    "if BUCKET == \"\" or BUCKET is None or BUCKET == \"[your-bucket-name]\":\n",
    "    # Get your bucket name to GCP projet id\n",
    "    BUCKET = PROJECT\n",
    "\n",
    "print(\"Project ID:\", PROJECT)\n",
    "print(\"Region:\", REGION)\n",
    "print(\"Bucket name:\", BUCKET)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85dd5e16",
   "metadata": {},
   "source": [
    "### Set configurations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "171a800f",
   "metadata": {},
   "outputs": [],
   "source": [
    "VERSION = 'v01'\n",
    "DATASET_DISPLAY_NAME = 'chicago-taxi-tips'\n",
    "MODEL_DISPLAY_NAME = f'{DATASET_DISPLAY_NAME}-classifier-{VERSION}'\n",
    "PIPELINE_NAME = f'{MODEL_DISPLAY_NAME}-train-pipeline'\n",
    "\n",
    "PIPELINES_STORE = f'gs://{BUCKET}/{DATASET_DISPLAY_NAME}/compiled_pipelines/'\n",
    "GCS_PIPELINE_FILE_LOCATION = os.path.join(PIPELINES_STORE, f'{PIPELINE_NAME}.json')\n",
    "PUBSUB_TOPIC = f'trigger-{PIPELINE_NAME}'\n",
    "CLOUD_FUNCTION_NAME = f'trigger-{PIPELINE_NAME}-fn'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "016df25c",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gsutil ls {GCS_PIPELINE_FILE_LOCATION}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "76d82223",
   "metadata": {},
   "source": [
    "## 1. Create a Pub/Sub topic"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0c1032c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud pubsub topics create {PUBSUB_TOPIC}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "08de54ef",
   "metadata": {},
   "source": [
    "## 2. Deploy the Cloud Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8597ad8d",
   "metadata": {},
   "outputs": [],
   "source": [
    "ENV_VARS=f\"\"\"\\\n",
    "PROJECT={PROJECT},\\\n",
    "REGION={REGION},\\\n",
    "GCS_PIPELINE_FILE_LOCATION={GCS_PIPELINE_FILE_LOCATION}\n",
    "\"\"\"\n",
    "\n",
    "!echo {ENV_VARS}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01a3d62a",
   "metadata": {},
   "outputs": [],
   "source": [
    "!rm -r src/pipeline_triggering/.ipynb_checkpoints"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b5acdb73",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud functions deploy {CLOUD_FUNCTION_NAME} \\\n",
    "    --region={REGION} \\\n",
    "    --trigger-topic={PUBSUB_TOPIC} \\\n",
    "    --runtime=python37 \\\n",
    "    --source=src/pipeline_triggering\\\n",
    "    --entry-point=trigger_pipeline\\\n",
    "    --stage-bucket={BUCKET}\\\n",
    "    --update-env-vars={ENV_VARS}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a5c41af",
   "metadata": {},
   "outputs": [],
   "source": [
    "cloud_fn_url = f\"https://console.cloud.google.com/functions/details/{REGION}/{CLOUD_FUNCTION_NAME}\"\n",
    "html = f'See the Cloud Function details <a href=\"{cloud_fn_url}\" target=\"_blank\">here</a>.'\n",
    "IPython.display.display(IPython.display.HTML(html))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ebbe047a",
   "metadata": {},
   "source": [
    "## 3. Trigger the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0c30028d",
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.cloud import pubsub\n",
    "\n",
    "publish_client = pubsub.PublisherClient()\n",
    "topic = f'projects/{PROJECT}/topics/{PUBSUB_TOPIC}'\n",
    "data = {\n",
    "    'num_epochs': 7,\n",
    "    'learning_rate': 0.0015,\n",
    "    'batch_size': 512,\n",
    "    'hidden_units': '256,126'\n",
    "}\n",
    "message = json.dumps(data)\n",
    "\n",
    "_ = publish_client.publish(topic, message.encode())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ba049fe",
   "metadata": {},
   "source": [
    "Wait for a few seconds for the pipeline run to be submitted, then you can see the run in the Cloud Console"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0dc29797",
   "metadata": {},
   "outputs": [],
   "source": [
    "from kfp.v2.google.client import AIPlatformClient\n",
    "\n",
    "pipeline_client = AIPlatformClient(\n",
    "    project_id=PROJECT, region=REGION)\n",
    " \n",
    "job_display_name = pipeline_client.list_jobs()['pipelineJobs'][0]['displayName']\n",
    "job_url = f\"https://console.cloud.google.com/vertex-ai/locations/{REGION}/pipelines/runs/{job_display_name}\"\n",
    "html = f'See the Pipeline job <a href=\"{job_url}\" target=\"_blank\">here</a>.'\n",
    "IPython.display.display(IPython.display.HTML(html))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4b3ff42",
   "metadata": {},
   "source": [
    "## 4. Extracting pipeline runs metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b13c1b19",
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.cloud import aiplatform as vertex_ai\n",
    "\n",
    "pipeline_df = vertex_ai.get_pipeline_df(PIPELINE_NAME)\n",
    "pipeline_df = pipeline_df[pipeline_df.pipeline_name == PIPELINE_NAME]\n",
    "pipeline_df.T"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9254cbc3",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "environment": {
   "name": "common-cpu.m73",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/base-cpu:m73"
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}