04-pipeline-deployment.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "109707f2",
   "metadata": {},
   "source": [
    "# 04 - Test and Deploy Training Pipeline to Vertex Pipelines\n",
    "\n",
    "The purpose of this notebook is to test, deploy, and run the `TFX` pipeline on `Vertex Pipelines`. The notebook covers the following tasks:\n",
    "1. Run the tests locally.\n",
    "2. Run the pipeline using `Vertex Pipelines`\n",
    "3. Execute the pipeline deployment `CI/CD` steps using `Cloud Build`."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f1a51af1",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "133e7d80",
   "metadata": {},
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "f3bb184e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Tensorflow Version: 1.8.0\n",
      "KFP Version: 1.8.12\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import kfp\n",
    "import tfx.v1 as tfx\n",
    "\n",
    "print(\"Tensorflow Version:\", tfx.__version__)\n",
    "print(\"KFP Version:\", kfp.__version__)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a7bdded",
   "metadata": {},
   "source": [
    "### Setup Google Cloud project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b4b22be",
   "metadata": {},
   "outputs": [],
   "source": [
    "PROJECT = 'pbalm-cxb-aa' # Change to your project id.\n",
    "REGION = 'europe-west4' # Change to your region.\n",
    "BUCKET =  PROJECT + '-eu' # Change to your bucket name.\n",
    "SERVICE_ACCOUNT = \"\"\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 SERVICE_ACCOUNT == \"\" or SERVICE_ACCOUNT is None or SERVICE_ACCOUNT == \"[your-service-account]\":\n",
    "    # Get your GCP project id from gcloud\n",
    "    shell_output = !gcloud config list --format 'value(core.account)' 2>/dev/null\n",
    "    SERVICE_ACCOUNT = shell_output[0]\n",
    "    \n",
    "if BUCKET == \"\" or BUCKET is None or BUCKET == \"[your-bucket-name]\":\n",
    "    # Get your bucket name to GCP project id\n",
    "    BUCKET = PROJECT\n",
    "    # Try to create the bucket if it doesn't exists\n",
    "    ! gsutil mb -l $REGION gs://$BUCKET\n",
    "    print(\"\")\n",
    "    \n",
    "print(\"Project ID:\", PROJECT)\n",
    "print(\"Region:\", REGION)\n",
    "print(\"Bucket name:\", BUCKET)\n",
    "print(\"Service Account:\", SERVICE_ACCOUNT)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "660a9dc9",
   "metadata": {},
   "source": [
    "### Set configurations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a75e169",
   "metadata": {},
   "outputs": [],
   "source": [
    "BQ_LOCATION = 'EU'\n",
    "BQ_DATASET_NAME = 'vertex_eu' # Change to your BQ dataset name.\n",
    "BQ_TABLE_NAME = 'chicago_taxitrips_prep'\n",
    "\n",
    "VERSION = 'v02'\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",
    "CICD_IMAGE_NAME = 'cicd:latest'\n",
    "CICD_IMAGE_URI = f\"gcr.io/{PROJECT}/{CICD_IMAGE_NAME}\"\n",
    "\n",
    "DATAFLOW_REGION = 'europe-west1'\n",
    "DATAFLOW_SERVICE_ACCOUNT = 'sa-notebooks@mlops1-notebooks.iam.gserviceaccount.com'\n",
    "DATAFLOW_SUBNETWORK = 'https://www.googleapis.com/compute/v1/projects/mlops1-network-host/regions/europe-west1/subnetworks/prod-notebooks-ew1'\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "06be3555",
   "metadata": {},
   "outputs": [],
   "source": [
    "!rm -r src/raw_schema/.ipynb_checkpoints/"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44678373",
   "metadata": {},
   "source": [
    "## 1. Run the CICD steps locally"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a68204ed",
   "metadata": {},
   "source": [
    "### Set pipeline configurations for the local run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "05a1d20d",
   "metadata": {},
   "outputs": [],
   "source": [
    "os.environ[\"DATASET_DISPLAY_NAME\"] = DATASET_DISPLAY_NAME\n",
    "os.environ[\"MODEL_DISPLAY_NAME\"] =  MODEL_DISPLAY_NAME\n",
    "os.environ[\"PIPELINE_NAME\"] = PIPELINE_NAME\n",
    "os.environ[\"PROJECT\"] = PROJECT\n",
    "os.environ[\"REGION\"] = REGION\n",
    "os.environ[\"BQ_LOCATION\"] = BQ_LOCATION\n",
    "os.environ[\"BQ_DATASET_NAME\"] = BQ_DATASET_NAME\n",
    "os.environ[\"BQ_TABLE_NAME\"] = BQ_TABLE_NAME\n",
    "os.environ[\"GCS_LOCATION\"] = f\"gs://{BUCKET}/{DATASET_DISPLAY_NAME}/e2e_tests\"\n",
    "os.environ[\"TRAIN_LIMIT\"] = \"1000\"\n",
    "os.environ[\"TEST_LIMIT\"] = \"100\"\n",
    "os.environ[\"UPLOAD_MODEL\"] = \"0\"\n",
    "os.environ[\"ACCURACY_THRESHOLD\"] = \"0.1\"\n",
    "os.environ[\"BEAM_RUNNER\"] = \"DirectRunner\"\n",
    "os.environ[\"TRAINING_RUNNER\"] = \"local\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d353e3d6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from src.tfx_pipelines import config\n",
    "import importlib\n",
    "importlib.reload(config)\n",
    "\n",
    "for key, value in config.__dict__.items():\n",
    "    if key.isupper(): print(f'{key}: {value}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "72d51c12",
   "metadata": {},
   "source": [
    "### Run unit tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9be84a8e",
   "metadata": {},
   "outputs": [],
   "source": [
    "!py.test src/tests/datasource_utils_tests.py -s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4358f955",
   "metadata": {},
   "outputs": [],
   "source": [
    "!py.test src/tests/model_tests.py -s"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5fa00fd5",
   "metadata": {},
   "source": [
    "### Run e2e pipeline test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cb9aad70",
   "metadata": {},
   "outputs": [],
   "source": [
    "!py.test src/tests/pipeline_deployment_tests.py::test_e2e_pipeline -s"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d5704bcb",
   "metadata": {},
   "source": [
    "## 2. Run the training pipeline using Vertex Pipelines\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c6d7db74",
   "metadata": {},
   "source": [
    "### Set the pipeline configurations for the Vertex AI run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9e2fe69b",
   "metadata": {},
   "outputs": [],
   "source": [
    "os.environ[\"DATASET_DISPLAY_NAME\"] = DATASET_DISPLAY_NAME\n",
    "os.environ[\"MODEL_DISPLAY_NAME\"] = MODEL_DISPLAY_NAME\n",
    "os.environ[\"PIPELINE_NAME\"] = PIPELINE_NAME\n",
    "os.environ[\"PROJECT\"] = PROJECT\n",
    "os.environ[\"REGION\"] = DATAFLOW_REGION\n",
    "os.environ[\"GCS_LOCATION\"] = f\"gs://{BUCKET}/{DATASET_DISPLAY_NAME}\"\n",
    "os.environ[\"TRAIN_LIMIT\"] = \"85000\"\n",
    "os.environ[\"TEST_LIMIT\"] = \"15000\"\n",
    "os.environ[\"BEAM_RUNNER\"] = \"DataflowRunner\"\n",
    "os.environ[\"TRAINING_RUNNER\"] = \"vertex\"\n",
    "os.environ[\"TFX_IMAGE_URI\"] = f\"gcr.io/{PROJECT}/{DATASET_DISPLAY_NAME}:{VERSION}\"\n",
    "os.environ[\"ENABLE_CACHE\"] = \"1\"\n",
    "os.environ[\"SUBNETWORK\"] = DATAFLOW_SUBNETWORK\n",
    "os.environ[\"SERVICE_ACCOUNT\"] = DATAFLOW_SERVICE_ACCOUNT"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d83ef31a",
   "metadata": {},
   "outputs": [],
   "source": [
    "from src.tfx_pipelines import config\n",
    "import importlib\n",
    "importlib.reload(config)\n",
    "\n",
    "for key, value in config.__dict__.items():\n",
    "    if key.isupper(): print(f'{key}: {value}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d5f3164f",
   "metadata": {},
   "source": [
    "### Build the ML container image\n",
    "\n",
    "This is the `TFX` runtime environment for the training pipeline steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0a0e729b",
   "metadata": {},
   "outputs": [],
   "source": [
    "!echo $TFX_IMAGE_URI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3087da4e",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud builds submit --tag $TFX_IMAGE_URI . --timeout=15m --machine-type=e2-highcpu-8"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "155568ca",
   "metadata": {},
   "source": [
    "### Compile pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6c1d5ce3",
   "metadata": {},
   "outputs": [],
   "source": [
    "from src.tfx_pipelines import runner\n",
    "\n",
    "pipeline_definition_file = f'{config.PIPELINE_NAME}.json'\n",
    "pipeline_definition = runner.compile_training_pipeline(pipeline_definition_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b5e506ad-e26d-4fd6-b235-e52302061ce8",
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.cloud import aiplatform as vertex_ai\n",
    "vertex_ai.init(project=PROJECT, location=REGION)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "855d0d6b-60cf-4bc4-8598-9a0f45acef14",
   "metadata": {},
   "outputs": [],
   "source": [
    "PIPELINES_STORE = f\"gs://{BUCKET}/{DATASET_DISPLAY_NAME}/compiled_pipelines/\"\n",
    "!gsutil cp {pipeline_definition_file} {PIPELINES_STORE}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0bcb943e",
   "metadata": {},
   "source": [
    "### Submit run to Vertex Pipelines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62c6b0ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "from kfp.v2.google.client import AIPlatformClient\n",
    "\n",
    "pipeline_client = AIPlatformClient(\n",
    "    project_id=PROJECT, region=REGION)\n",
    "                 \n",
    "job = pipeline_client.create_run_from_job_spec(\n",
    "    job_spec_path=pipeline_definition_file,\n",
    "    parameter_values={\n",
    "        'learning_rate': 0.003,\n",
    "        'batch_size': 512,\n",
    "        'hidden_units': '128,128',\n",
    "        'num_epochs': 30,\n",
    "    },\n",
    "    service_account=DATAFLOW_SERVICE_ACCOUNT\n",
    ")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "888be1fd",
   "metadata": {},
   "source": [
    "### Extracting pipeline runs metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "37ae4aa9",
   "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": "markdown",
   "id": "4b454fe9",
   "metadata": {},
   "source": [
    "## 3. Execute the pipeline deployment CI/CD steps in Cloud Build\n",
    "\n",
    "The CI/CD routine is defined in the [pipeline-deployment.yaml](pipeline-deployment.yaml) file, and consists of the following steps:\n",
    "1. Clone the repository to the build environment.\n",
    "2. Run unit tests.\n",
    "3. Run a local e2e test of the pipeline.\n",
    "4. Build the ML container image for pipeline steps.\n",
    "5. Compile the pipeline.\n",
    "6. Upload the pipeline to Cloud Storage."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "29688d4d",
   "metadata": {},
   "source": [
    "### Build CI/CD container Image for Cloud Build\n",
    "\n",
    "This is the runtime environment where the steps of testing and deploying the pipeline will be executed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4759b85b",
   "metadata": {},
   "outputs": [],
   "source": [
    "!echo $CICD_IMAGE_URI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4fc09c3e",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud builds submit --tag $CICD_IMAGE_URI build/. --timeout=15m --machine-type=e2-highcpu-8"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2fc9b2af",
   "metadata": {},
   "source": [
    "### Run CI/CD from pipeline deployment using Cloud Build"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "00b55593",
   "metadata": {},
   "outputs": [],
   "source": [
    "#REPO_URL = \"https://github.com/GoogleCloudPlatform/mlops-with-vertex-ai.git\" # Change to your github repo.\n",
    "REPO_URL=\"https://github.com/javiergp/mlops-with-vertex-ai.git\"\n",
    "\n",
    "BRANCH = \"main\"\n",
    "\n",
    "GCS_LOCATION = f\"gs://{BUCKET}/{DATASET_DISPLAY_NAME}/\"\n",
    "TEST_GCS_LOCATION = f\"gs://{BUCKET}/{DATASET_DISPLAY_NAME}/e2e_tests\"\n",
    "CI_TRAIN_LIMIT = 1000\n",
    "CI_TEST_LIMIT = 100\n",
    "CI_UPLOAD_MODEL = 0\n",
    "CI_ACCURACY_THRESHOLD = 0.1\n",
    "BEAM_RUNNER = \"DataflowRunner\"\n",
    "TRAINING_RUNNER = \"vertex\"\n",
    "VERSION = 'tfx-1.2'\n",
    "PIPELINE_NAME = f'{MODEL_DISPLAY_NAME}-train-pipeline'\n",
    "PIPELINES_STORE = os.path.join(GCS_LOCATION, \"compiled_pipelines\")\n",
    "\n",
    "TFX_IMAGE_URI = f\"gcr.io/{PROJECT}/{DATASET_DISPLAY_NAME}:{VERSION}\"\n",
    "\n",
    "SUBSTITUTIONS=f\"\"\"\\\n",
    "_REPO_URL='{REPO_URL}',\\\n",
    "_BRANCH={BRANCH},\\\n",
    "_CICD_IMAGE_URI={CICD_IMAGE_URI},\\\n",
    "_PROJECT={PROJECT},\\\n",
    "_REGION={DATAFLOW_REGION},\\\n",
    "_GCS_LOCATION={GCS_LOCATION},\\\n",
    "_TEST_GCS_LOCATION={TEST_GCS_LOCATION},\\\n",
    "_BQ_LOCATION={BQ_LOCATION},\\\n",
    "_BQ_DATASET_NAME={BQ_DATASET_NAME},\\\n",
    "_BQ_TABLE_NAME={BQ_TABLE_NAME},\\\n",
    "_DATASET_DISPLAY_NAME={DATASET_DISPLAY_NAME},\\\n",
    "_MODEL_DISPLAY_NAME={MODEL_DISPLAY_NAME},\\\n",
    "_CI_TRAIN_LIMIT={CI_TRAIN_LIMIT},\\\n",
    "_CI_TEST_LIMIT={CI_TEST_LIMIT},\\\n",
    "_CI_UPLOAD_MODEL={CI_UPLOAD_MODEL},\\\n",
    "_CI_ACCURACY_THRESHOLD={CI_ACCURACY_THRESHOLD},\\\n",
    "_BEAM_RUNNER={BEAM_RUNNER},\\\n",
    "_TRAINING_RUNNER={TRAINING_RUNNER},\\\n",
    "_TFX_IMAGE_URI={TFX_IMAGE_URI},\\\n",
    "_PIPELINE_NAME={PIPELINE_NAME},\\\n",
    "_PIPELINES_STORE={PIPELINES_STORE},\\\n",
    "_SUBNETWORK={DATAFLOW_SUBNETWORK},\\\n",
    "_GCS_BUCKET={BUCKET}/cloudbuild,\\\n",
    "_SERVICE_ACCOUNT={DATAFLOW_SERVICE_ACCOUNT}\\\n",
    "\"\"\"\n",
    "!echo $SUBSTITUTIONS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c79ea6ba-0d59-439b-9117-e0fbffa1ca21",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud builds submit --no-source --timeout=60m --config build/pipeline-deployment.yaml --substitutions {SUBSTITUTIONS} --machine-type=e2-highcpu-8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "16028a3c",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6c4ab15c-006d-45d9-9540-59c6c48131bc",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "environment": {
   "kernel": "python3",
   "name": "tf2-gpu.2-8.m93",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-gpu.2-8:m93"
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}