04-pipeline-deployment.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "d272910e",
   "metadata": {},
   "source": [
    "# 04 - Test and deploy a TFX 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",
    "\n",
    "1. Run the tests locally.\n",
    "2. Run the `TFX` pipeline using `Vertex Pipelines`\n",
    "3. Execute the pipeline deployment `CI/CD` steps using `Cloud Build`."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "beaa2787",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "51e05608",
   "metadata": {},
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9aa72b29",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import kfp\n",
    "import tfx\n",
    "\n",
    "print('TFX:', tfx.__version__)\n",
    "print('KFP:', kfp.__version__)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "24aceb9a",
   "metadata": {},
   "source": [
    "### Setup Google Cloud project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d8d9f81b",
   "metadata": {},
   "outputs": [],
   "source": [
    "PROJECT_ID = '[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",
    "SERVICE_ACCOUNT = '[your-service-account]'\n",
    "\n",
    "if PROJECT_ID == '' or PROJECT_ID is None or PROJECT_ID == '[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_ID = 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",
    "    # Set your bucket name using your GCP project id\n",
    "    BUCKET = PROJECT_ID\n",
    "    # Try to create the bucket if it doesn'exists\n",
    "    ! gsutil mb -l $REGION gs://$BUCKET\n",
    "    print('')\n",
    "    \n",
    "print('Project ID:', PROJECT_ID)\n",
    "print('Region:', REGION)\n",
    "print('Bucket name:', BUCKET)\n",
    "print('Service Account:', SERVICE_ACCOUNT)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d24d18b",
   "metadata": {},
   "source": [
    "### Set configurations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8295cca",
   "metadata": {},
   "outputs": [],
   "source": [
    "BQ_LOCATION = 'US'\n",
    "BQ_DATASET_NAME = 'playground_us' # Change to your BQ dataset name.\n",
    "BQ_TABLE_NAME = 'chicago_taxitrips_prep'\n",
    "\n",
    "VERSION = 'v1'\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_ID}/{CICD_IMAGE_NAME}'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81d049f5",
   "metadata": {},
   "outputs": [],
   "source": [
    "! rm -r src/raw_schema/.ipynb_checkpoints/"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4cbcbbb8",
   "metadata": {},
   "source": [
    "## 1. Run the CI/CD steps locally"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "58845362",
   "metadata": {},
   "source": [
    "### Set pipeline configurations for the local run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "44c48da6",
   "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_ID\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": "fcf65dee",
   "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": "0e4989b9",
   "metadata": {},
   "source": [
    "### Run the unit tests for the data and model pipeline components"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "37324634",
   "metadata": {},
   "outputs": [],
   "source": [
    "! py.test src/tests/datasource_utils_tests.py -s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1a40f106",
   "metadata": {},
   "outputs": [],
   "source": [
    "! py.test src/tests/model_tests.py -s"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3b62aea",
   "metadata": {},
   "source": [
    "### Run the e2e pipeline test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3acb31cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "! py.test src/tests/pipeline_deployment_tests.py::test_e2e_pipeline -s"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1c8758df",
   "metadata": {},
   "source": [
    "## 2. Run the training pipeline using `Vertex Pipelines`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a02ce062",
   "metadata": {},
   "source": [
    "### Set the pipeline configurations for the `Vertex Pipeline` run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01c2b3e1",
   "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_ID\n",
    "os.environ['REGION'] = 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_ID}/{DATASET_DISPLAY_NAME}:{VERSION}'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9e8be723",
   "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": "286ff84e",
   "metadata": {},
   "source": [
    "### Build the training container image\n",
    "\n",
    "This is the `TFX` runtime environment for the training pipeline steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d9686014",
   "metadata": {},
   "outputs": [],
   "source": [
    "!echo $TFX_IMAGE_URI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7f7986c2",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud builds submit --tag $TFX_IMAGE_URI . --timeout=15m --machine-type=e2-highcpu-8"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce2d5c9a",
   "metadata": {},
   "source": [
    "### Compile the `TFX` pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "df29fc7e",
   "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": "75928c08",
   "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": "ef836781",
   "metadata": {},
   "source": [
    "### Submit run to Vertex Pipelines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1d1115bd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from kfp.v2.google.client import AIPlatformClient\n",
    "\n",
    "pipeline_client = AIPlatformClient(\n",
    "    project_id=PROJECT_ID, 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",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7cc4477f",
   "metadata": {},
   "source": [
    "### Extracting pipeline runs metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "464ad3a8",
   "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": "ad380129",
   "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": "e00b075f",
   "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": "867e5ae1",
   "metadata": {},
   "outputs": [],
   "source": [
    "! echo $CICD_IMAGE_URI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "40f497f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "! gcloud builds submit --tag $CICD_IMAGE_URI build/. --timeout=15m --machine-type=e2-highcpu-8"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4f6e2dd7",
   "metadata": {},
   "source": [
    "### Run CI/CD from pipeline deployment using Cloud Build"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "117895d7",
   "metadata": {},
   "outputs": [],
   "source": [
    "REPO_URL = 'https://github.com/ksalama/ucaip-labs.git' # Change to your github repo.\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-0-30'\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_ID}/{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_ID={PROJECT_ID},\\\n",
    "_REGION={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",
    "'''\n",
    "\n",
    "!echo $SUBSTITUTIONS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b54081db",
   "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": "72d9baf5",
   "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
}