cloud_scheduler_trigger.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ZhYmiSEZSIBl"
   },
   "source": [
    "## References\n",
    "\n",
    "* [Cloud Scheduler Official Doc](https://cloud.google.com/scheduler/docs)\n",
    "* [Cloud Scheduler RPC Spec](https://cloud.google.com/scheduler/docs/reference/rpc/google.cloud.scheduler.v1#google.cloud.scheduler.v1.CloudScheduler)\n",
    "* [Cloud Scheduler Python API Docs](https://googleapis.dev/python/cloudscheduler/latest/scheduler_v1/cloud_scheduler.html?highlight=google%20cloud%20scheduler_v1#module-google.cloud.scheduler_v1.services.cloud_scheduler)\n",
    "* [Cloud Scheduler Python Example](https://stackoverflow.com/questions/60681672/how-to-create-a-job-with-google-cloud-scheduler-python-api)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pre-requisites\n",
    "This notebook only shows how to create jobs for Google Cloud Scheduler service. In order to reproduce the same output, the following pre-requisites must be met.\n",
    "- [Cloud Build TFX Notebook](https://github.com/sayakpaul/CI-CD-for-Model-Training/blob/main/cloud_build_tfx.ipynb)\n",
    "  - Build TFX pipeline, docker image which each TFX component will be run, and compile a TFX pipeline job spec.\n",
    "- [Cloud Function Trigger Notebook](https://github.com/sayakpaul/CI-CD-for-Model-Training/blob/main/cloud_function_trigger.ipynb)\n",
    "  - Create a Pub/Sub topic\n",
    "  - Create and deploy Cloud Function to trigger Vertex AI pipeline by refering to the TFX pipeline job spec."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "PqwV9XW-of3W"
   },
   "source": [
    "## Setting up\n",
    "By installing `google-cloud-scheduler` Python package, you can create jobs for Cloud Scheduler programatically."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "IqkPzmaXVNPU"
   },
   "outputs": [],
   "source": [
    "!pip install --upgrade -q google-cloud-scheduler"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "l2RDeydWpULu"
   },
   "source": [
    "### ***Restart runtime (if you are using Colab)***"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "6Ii493lzSIBm"
   },
   "outputs": [],
   "source": [
    "!gcloud init # only need if you are using Colab"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "wAkLWKYRSIBm"
   },
   "outputs": [],
   "source": [
    "# only need if you are using Colab\n",
    "from google.colab import auth\n",
    "auth.authenticate_user()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "id": "EbQKTTkuSIBn"
   },
   "outputs": [],
   "source": [
    "GOOGLE_CLOUD_PROJECT = \"gcp-ml-172005\"\n",
    "GOOGLE_CLOUD_REGION = \"us-central1\"\n",
    "\n",
    "PIPELINE_NAME = \"penguin-vertex-training\"\n",
    "PUBSUB_TOPIC = f\"trigger-{PIPELINE_NAME}\"\n",
    "SCHEDULER_JOB_NAME = \"MLOpsJob\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Pub/Sub Topic via CLI\n",
    "\n",
    "#### Setup environment variable for GCP credentials which has a permission to the Google Cloud Scheduler\n",
    "- You need to get and upload the credentials beforehand. Please refer to this [official document](https://cloud.google.com/run/docs/triggering/using-scheduler#create-service-account).\n",
    "- `gcloud scheduler jobs` command will automatically recognize the environment variable, `GOOGLE_APPLICATION_CREDENTIALS`. Otherwise, you have to specify it explicitly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!export GOOGLE_APPLICATION_CREDENTIALS=\"/home/jupyter/CI-CD-for-Model-Training/gcp-ml-172005-528977a75f85.json\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- The message comsumed by Cloud Function from Pub/Sub should be encoded with `json.dumps`. \n",
    "- The following `gcloud` command will schedule the job `*/3 * * * *` which means every three minutes. You don't want to schedule this often for real world project, but this value is set for a demonstration purpose only."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 172,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "data = '{\"num_epochs\": \"3\", \"learning_rate\": \"1e-2\"}'\n",
    "data = json.dumps(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 173,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "name: projects/gcp-ml-172005/locations/us-central1/jobs/MLOpsJob\n",
      "pubsubTarget:\n",
      "  data: eyJudW1fZXBvY2hzIjogIjMiLCAibGVhcm5pbmdfcmF0ZSI6ICIxZS0yIn0=\n",
      "  topicName: projects/gcp-ml-172005/topics/trigger-penguin-vertex-training\n",
      "retryConfig:\n",
      "  maxBackoffDuration: 3600s\n",
      "  maxDoublings: 16\n",
      "  maxRetryDuration: 0s\n",
      "  minBackoffDuration: 5s\n",
      "schedule: '*/3 * * * *'\n",
      "state: ENABLED\n",
      "timeZone: Etc/UTC\n",
      "userUpdateTime: '2021-08-26T17:41:58Z'\n"
     ]
    }
   ],
   "source": [
    "!gcloud scheduler jobs create pubsub $SCHEDULER_JOB_NAME --schedule \"*/3 * * * *\" --topic $PUBSUB_TOPIC --message-body $data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### See Its behaviour in Vertex AI Pipeline\n",
    "\n",
    "![](https://i.ibb.co/GkHmwTL/Screen-Shot-2021-08-27-at-2-49-00-AM.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "lFt8n26ioh3F"
   },
   "source": [
    "## Create Pub/Sub Topic Programatically\n",
    "\n",
    "Let's see how we can do the same thing programatically in Python."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "from google.cloud import scheduler_v1\n",
    "from google.cloud.scheduler_v1.types.target import PubsubTarget\n",
    "from google.cloud.scheduler_v1.types.job import Job\n",
    "from google.cloud.scheduler_v1.types.cloudscheduler import CreateJobRequest\n",
    "\n",
    "client = scheduler_v1.CloudSchedulerClient.from_service_account_json(\n",
    "    r\"./gcp-ml-172005-528977a75f85.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are three main differences compared to `gcloud` command.\n",
    "- The message should be encoded in `utf-8`. This makes sure the message is encoded in bytes, and `data` parameter in `PubsubTarget` requires the message to be `bytes`.\n",
    "- Pub/Sub topic name should follow the `\"projects/<PROJECT-ID>/topics/<TOPIC-NAME>\"` format.\n",
    "- Scheduler Job name should follow the `\"projects/<PROJECT-ID>/locations/<REGION-ID>/jobs/<JOB-NAME>\"` format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "parent = client.common_location_path(GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_REGION)\n",
    "\n",
    "data = {\"num_epochs\": \"3\", \"learning_rate\": \"1e-2\"}\n",
    "data = json.dumps(data).encode('utf-8')\n",
    "pubsub_target = PubsubTarget(\n",
    "    topic_name=f\"projects/{GOOGLE_CLOUD_PROJECT}/topics/{PUBSUB_TOPIC}\", \n",
    "    data=data)\n",
    "\n",
    "job = Job(name=f\"projects/{GOOGLE_CLOUD_PROJECT}/locations/{GOOGLE_CLOUD_REGION}/jobs/traing_for_model\", \n",
    "          pubsub_target=pubsub_target, \n",
    "          schedule=\"*/3 * * * *\")\n",
    "\n",
    "req = CreateJobRequest(parent=parent, job=job)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "result_job = client.create_job(req)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "name: \"projects/gcp-ml-172005/locations/us-central1/jobs/traing_for_model\"\n",
       "pubsub_target {\n",
       "  topic_name: \"projects/gcp-ml-172005/topics/trigger-penguin-vertex-training\"\n",
       "  data: \"{\\\"num_epochs\\\": \\\"3\\\", \\\"learning_rate\\\": \\\"1e-2\\\"}\"\n",
       "}\n",
       "user_update_time {\n",
       "  seconds: 1630031544\n",
       "}\n",
       "state: ENABLED\n",
       "schedule: \"*/3 * * * *\"\n",
       "time_zone: \"UTC\""
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result_job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "include_colab_link": true,
   "name": "cloud_function_trigger.ipynb",
   "provenance": []
  },
  "environment": {
   "name": "common-cpu.m78",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/base-cpu:m78"
  },
  "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": 4
}