ames-xgboost-build-train-deploy.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Train and deploy on Kubeflow from Notebooks\n",
    "\n",
    "This notebook introduces you to using Kubeflow Fairing to train and deploy a model to Kubeflow on Google Kubernetes Engine (GKE), and Google Cloud ML Engine. This notebook demonstrate how to:\n",
    " \n",
    "* Train an XGBoost model in a local notebook,\n",
    "* Use Kubeflow Fairing to train an XGBoost model remotely on Kubeflow,\n",
    "  * Data is read from a PVC\n",
    "  * The append builder is used to rapidly build a docker image\n",
    "* Use Kubeflow Fairing to deploy a trained model to Kubeflow, and\n",
    "* Call the deployed endpoint for predictions.\n",
    "\n",
    "To learn more about how to run this notebook locally, see the guide to [training and deploying on GCP from a local notebook][gcp-local-notebook].\n",
    "\n",
    "[gcp-local-notebook]: https://kubeflow.org/docs/fairing/gcp-local-notebook/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set up your notebook for training an XGBoost model\n",
    "\n",
    "Import the libraries required to train this model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:root:Adding /home/jovyan/git_jlewi-kubecon-demo/fairing to path\n"
     ]
    }
   ],
   "source": [
    "import demo_util\n",
    "from pathlib import Path\n",
    "import os\n",
    "fairing_code = os.path.join(Path.home(), \"git_jlewi-kubecon-demo\", \"fairing\")\n",
    "demo_util.notebook_setup(fairing_code)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# fairing:include-cell\n",
    "import ames\n",
    "import fire\n",
    "import joblib\n",
    "import logging\n",
    "import nbconvert\n",
    "import os\n",
    "import pathlib\n",
    "import sys\n",
    "from pathlib import Path\n",
    "import pandas as pd\n",
    "import pprint\n",
    "from sklearn.metrics import mean_absolute_error\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.impute import SimpleImputer\n",
    "from xgboost import XGBRegressor\n",
    "from importlib import reload"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Imports not to be included in the built docker image\n",
    "import kfp\n",
    "import kfp.components as comp\n",
    "import kfp.gcp as gcp\n",
    "import kfp.dsl as dsl\n",
    "import kfp.compiler as compiler\n",
    "from kubernetes import client as k8s_client\n",
    "import fairing   \n",
    "from fairing.builders import append\n",
    "from fairing.deployers import job\n",
    "import fairing_util"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define various constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "nfs_path = os.path.join(\"/mnt/kubeflow-gcfs/data/ames_dataset\")\n",
    "model_dir = os.path.join(\"/mnt/kubeflow-gcfs/models\")\n",
    "train_data = \"/mnt/kubeflow-gcfs/data/ames_dataset/train.csv\"\n",
    "model_file = os.path.join(model_dir, \"trained_ames_model.dat\")\n",
    "\n",
    "# Base image is built from the Dockerfile in the repo\n",
    "# Can be the same image as your notebook\n",
    "base_image = \"gcr.io/code-search-demo/kubecon-demo/notebook:v20190518-2d04328-dirty-a8c2a5\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Copy data to nfs\n",
    "demo_util.copy_data_to_nfs(nfs_path, model_dir)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define Train and Predict functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# fairing:include-cell\n",
    "class HousingServe(object):    \n",
    "    def __init__(self, model_file=None):\n",
    "        self.n_estimators = 50\n",
    "        self.learning_rate = 0.1\n",
    "        if not model_file:\n",
    "            print(\"model_file not supplied; checking environment variable\")\n",
    "            model_file = os.getenv(\"MODEL_FILE\")\n",
    "        \n",
    "        self.model_file = model_file\n",
    "        print(\"model_file={0}\".format(self.model_file))\n",
    "        \n",
    "        self.model = None\n",
    "                \n",
    "\n",
    "    def train(self, train_input, model_file):\n",
    "        (train_X, train_y), (test_X, test_y) = ames.read_input(train_input)\n",
    "        model = ames.train_model(train_X,\n",
    "                                 train_y,\n",
    "                                 test_X,\n",
    "                                 test_y,\n",
    "                                 self.n_estimators,\n",
    "                                 self.learning_rate)\n",
    "\n",
    "        ames.eval_model(model, test_X, test_y)\n",
    "        ames.save_model(model, model_file)\n",
    "\n",
    "    def predict(self, X, feature_names):\n",
    "        \"\"\"Predict using the model for given ndarray.\"\"\"\n",
    "        if not self.model:\n",
    "            print(\"Loading model {0}\".format(self.model_file))\n",
    "            self.model = ames.load_model(self.model_file)\n",
    "\n",
    "        # Do any preprocessing\n",
    "        prediction = self.model.predict(data=X)\n",
    "        # Do any postprocessing\n",
    "        return [[prediction.item(0), prediction.item(1)]]\n",
    "    \n",
    "    def create_pr_to_update_model(self, job_spec_file, new_model):\n",
    "        ames.create_pr_to_update_model(job_spec_file, new_model)\n",
    "    \n",
    "    def deploy_model(self, model_file):\n",
    "        ames.deploy_model(model_file)\n",
    "\n",
    "    def validate_model(self, endpoint):\n",
    "        ames.validate_model(endpoint)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train your Model Locally\n",
    "\n",
    "* Train your model locally inside your notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "model_file=/tmp/trained_model.dat\n",
      "[0]\tvalidation_0-rmse:177514\n",
      "Will train until validation_0-rmse hasn't improved in 40 rounds.\n",
      "[1]\tvalidation_0-rmse:161858\n",
      "[2]\tvalidation_0-rmse:147237\n",
      "[3]\tvalidation_0-rmse:134132\n",
      "[4]\tvalidation_0-rmse:122224\n",
      "[5]\tvalidation_0-rmse:111538\n",
      "[6]\tvalidation_0-rmse:102142\n",
      "[7]\tvalidation_0-rmse:93392.3\n",
      "[8]\tvalidation_0-rmse:85824.6\n",
      "[9]\tvalidation_0-rmse:79667.6\n",
      "[10]\tvalidation_0-rmse:73463.4\n",
      "[11]\tvalidation_0-rmse:68059.4\n",
      "[12]\tvalidation_0-rmse:63350.5\n",
      "[13]\tvalidation_0-rmse:59732.1\n",
      "[14]\tvalidation_0-rmse:56260.7\n",
      "[15]\tvalidation_0-rmse:53392.6\n",
      "[16]\tvalidation_0-rmse:50770.8\n",
      "[17]\tvalidation_0-rmse:48107.8\n",
      "[18]\tvalidation_0-rmse:45923.9\n",
      "[19]\tvalidation_0-rmse:44154.2\n",
      "[20]\tvalidation_0-rmse:42488.1\n",
      "[21]\tvalidation_0-rmse:41263.3\n",
      "[22]\tvalidation_0-rmse:40212.8\n",
      "[23]\tvalidation_0-rmse:39089.1\n",
      "[24]\tvalidation_0-rmse:37691.1\n",
      "[25]\tvalidation_0-rmse:36875.2\n",
      "[26]\tvalidation_0-rmse:36276.2\n",
      "[27]\tvalidation_0-rmse:35444.1\n",
      "[28]\tvalidation_0-rmse:34831.5\n",
      "[29]\tvalidation_0-rmse:34205.4\n",
      "[30]\tvalidation_0-rmse:33831.9\n",
      "[31]\tvalidation_0-rmse:33183.6\n",
      "[32]\tvalidation_0-rmse:33019.4\n",
      "[33]\tvalidation_0-rmse:32680\n",
      "[34]\tvalidation_0-rmse:32438.5\n",
      "[35]\tvalidation_0-rmse:32130.4\n",
      "[36]\tvalidation_0-rmse:31644.2\n",
      "[37]\tvalidation_0-rmse:31248.9\n",
      "[38]\tvalidation_0-rmse:31059.8\n",
      "[39]\tvalidation_0-rmse:30862.4\n",
      "[40]\tvalidation_0-rmse:30754\n",
      "[41]\tvalidation_0-rmse:30561.6\n",
      "[42]\tvalidation_0-rmse:30416.9\n",
      "[43]\tvalidation_0-rmse:30156.4\n",
      "[44]\tvalidation_0-rmse:29852.9\n",
      "[45]\tvalidation_0-rmse:29486.6\n",
      "[46]\tvalidation_0-rmse:29158.8\n",
      "[47]\tvalidation_0-rmse:29017\n",
      "[48]\tvalidation_0-rmse:28973.9\n",
      "[49]\tvalidation_0-rmse:28787.7\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:root:Best RMSE on eval: 28787.72 with 50 rounds\n",
      "INFO:root:mean_absolute_error=18173.15\n",
      "INFO:root:Model export success: /tmp/trained_model.dat\n"
     ]
    }
   ],
   "source": [
    "local_model_file = \"/tmp/trained_model.dat\"\n",
    "housing = HousingServe(local_model_file)\n",
    "\n",
    "housing.train(train_data, local_model_file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Predict locally\n",
    "\n",
    "* Run prediction inside the notebook using the newly created notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Loading model /tmp/trained_model.dat\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[[165164.875, 111924.984375]]"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(train_X, train_y), (test_X, test_y) = ames.read_input(\"ames_dataset/train.csv\")\n",
    "\n",
    "housing.predict(test_X, None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Use Fairing to Launch a K8s Job to train your model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Set up Kubeflow Fairing for training and predictions\n",
    "\n",
    "Import the `fairing` library and configure the environment that your training or prediction job will run in."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Setting up google container repositories (GCR) for storing output containers\n",
    "# You can use any docker container registry istead of GCR\n",
    "GCP_PROJECT = fairing.cloud.gcp.guess_project_name()\n",
    "DOCKER_REGISTRY = 'gcr.io/{}/fairing-job'.format(GCP_PROJECT)\n",
    "PY_VERSION = \".\".join([str(x) for x in sys.version_info[0:3]])\n",
    "BASE_IMAGE = 'python:{}'.format(PY_VERSION)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Use fairing to build the docker image\n",
    "\n",
    "* This uses the append builder to rapidly build docker images"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:fairing.builders.append.append:Building image...\n",
      "INFO:root:Creating docker context: /tmp/fairing.context.tar.gz\n",
      "INFO:root:Adding files to context: [PosixPath('ames-xgboost-build-train-deploy.py'), 'ames.py', 'deployment/update_model_job.yaml', 'update_model.py']\n",
      "INFO:root:Context: /tmp/fairing.context.tar.gz, Adding /home/jovyan/git_jlewi-kubecon-demo/fairing/fairing/__init__.py at /app/fairing/__init__.py\n",
      "INFO:root:Context: /tmp/fairing.context.tar.gz, Adding /home/jovyan/git_jlewi-kubecon-demo/fairing/fairing/runtime_config.py at /app/fairing/runtime_config.py\n",
      "INFO:root:Context: /tmp/fairing.context.tar.gz, Adding ames-xgboost-build-train-deploy.py at /app/ames-xgboost-build-train-deploy.py\n",
      "INFO:root:Context: /tmp/fairing.context.tar.gz, Adding ames.py at /app/ames.py\n",
      "INFO:root:Context: /tmp/fairing.context.tar.gz, Adding deployment/update_model_job.yaml at /app/deployment/update_model_job.yaml\n",
      "INFO:root:Context: /tmp/fairing.context.tar.gz, Adding update_model.py at /app/update_model.py\n",
      "INFO:root:Loading Docker credentials for repository 'gcr.io/code-search-demo/kubecon-demo/notebook:v20190518-2d04328-dirty-a8c2a5'\n",
      "INFO:root:Invoking 'docker-credential-gcloud' to obtain Docker credentials.\n",
      "INFO:root:Successfully obtained Docker credentials.\n",
      "WARNING:fairing.builders.append.append:Image successfully built in 0.9568250589072704s.\n",
      "WARNING:fairing.builders.append.append:Pushing image gcr.io/code-search-demo/fairing-job/fairing-job:77105FE...\n",
      "INFO:root:Loading Docker credentials for repository 'gcr.io/code-search-demo/fairing-job/fairing-job:77105FE'\n",
      "INFO:root:Invoking 'docker-credential-gcloud' to obtain Docker credentials.\n",
      "INFO:root:Successfully obtained Docker credentials.\n",
      "WARNING:fairing.builders.append.append:Uploading gcr.io/code-search-demo/fairing-job/fairing-job:77105FE\n",
      "INFO:root:Tag points to the right manifest, skipping push.\n",
      "INFO:root:Finished upload of: gcr.io/code-search-demo/fairing-job/fairing-job:77105FE\n",
      "WARNING:fairing.builders.append.append:Pushed image gcr.io/code-search-demo/fairing-job/fairing-job:77105FE in 0.705476351082325s.\n"
     ]
    }
   ],
   "source": [
    "preprocessor = fairing_util.ConvertNotebookPreprocessorWithFire(\"HousingServe\")\n",
    "\n",
    "if not preprocessor.input_files:\n",
    "    preprocessor.input_files = set()\n",
    "input_files=[\"ames.py\", \"deployment/update_model_job.yaml\", \"update_model.py\"]\n",
    "preprocessor.input_files =  set([os.path.normpath(f) for f in input_files])\n",
    "preprocessor.preprocess()\n",
    "builder = append.append.AppendBuilder(registry=DOCKER_REGISTRY,\n",
    "                                      base_image=base_image, preprocessor=preprocessor)\n",
    "builder.build()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Launch the K8s Job\n",
    "\n",
    "* Use pod mutators to attach a PVC and credentials to the pod"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING:fairing.deployers.job.job:Training job fairing-job-c695b launched.\n",
      "WARNING:fairing.kubernetes.manager:Waiting for fairing-job-c695b-bhvdw to start...\n",
      "WARNING:fairing.kubernetes.manager:Waiting for fairing-job-c695b-bhvdw to start...\n",
      "WARNING:fairing.kubernetes.manager:Waiting for fairing-job-c695b-bhvdw to start...\n",
      "INFO:fairing.kubernetes.manager:Pod started running True\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "model_file not supplied; checking environment variable\n",
      "model_file=None\n",
      "[0]\tvalidation_0-rmse:177514\n",
      "Will train until validation_0-rmse hasn't improved in 40 rounds.\n",
      "[1]\tvalidation_0-rmse:161858\n",
      "[2]\tvalidation_0-rmse:147237\n",
      "[3]\tvalidation_0-rmse:134132\n",
      "[4]\tvalidation_0-rmse:122224\n",
      "[5]\tvalidation_0-rmse:111538\n",
      "[6]\tvalidation_0-rmse:102142\n",
      "[7]\tvalidation_0-rmse:93392.3\n",
      "[8]\tvalidation_0-rmse:85824.6\n",
      "[9]\tvalidation_0-rmse:79667.6\n",
      "[10]\tvalidation_0-rmse:73463.4\n",
      "[11]\tvalidation_0-rmse:68059.4\n",
      "[12]\tvalidation_0-rmse:63350.5\n",
      "[13]\tvalidation_0-rmse:59732.1\n",
      "[14]\tvalidation_0-rmse:56260.7\n",
      "[15]\tvalidation_0-rmse:53392.6\n",
      "[16]\tvalidation_0-rmse:50770.8\n",
      "[17]\tvalidation_0-rmse:48107.8\n",
      "[18]\tvalidation_0-rmse:45923.9\n",
      "[19]\tvalidation_0-rmse:44154.2\n",
      "[20]\tvalidation_0-rmse:42488.1\n",
      "[21]\tvalidation_0-rmse:41263.3\n",
      "[22]\tvalidation_0-rmse:40212.8\n",
      "[23]\tvalidation_0-rmse:39089.1\n",
      "[24]\tvalidation_0-rmse:37691.1\n",
      "[25]\tvalidation_0-rmse:36875.2\n",
      "[26]\tvalidation_0-rmse:36276.2\n",
      "[27]\tvalidation_0-rmse:35444.1\n",
      "[28]\tvalidation_0-rmse:34831.5\n",
      "[29]\tvalidation_0-rmse:34205.4\n",
      "[30]\tvalidation_0-rmse:33831.9\n",
      "[31]\tvalidation_0-rmse:33183.6\n",
      "[32]\tvalidation_0-rmse:33019.4\n",
      "[33]\tvalidation_0-rmse:32680\n",
      "[34]\tvalidation_0-rmse:32438.5\n",
      "[35]\tvalidation_0-rmse:32130.4\n",
      "[36]\tvalidation_0-rmse:31644.2\n",
      "[37]\tvalidation_0-rmse:31248.9\n",
      "[38]\tvalidation_0-rmse:31059.8\n",
      "[39]\tvalidation_0-rmse:30862.4\n",
      "[40]\tvalidation_0-rmse:30754\n",
      "[41]\tvalidation_0-rmse:30561.6\n",
      "[42]\tvalidation_0-rmse:30416.9\n",
      "[43]\tvalidation_0-rmse:30156.4\n",
      "[44]\tvalidation_0-rmse:29852.9\n",
      "[45]\tvalidation_0-rmse:29486.6\n",
      "[46]\tvalidation_0-rmse:29158.8\n",
      "[47]\tvalidation_0-rmse:29017\n",
      "[48]\tvalidation_0-rmse:28973.9\n",
      "[49]\tvalidation_0-rmse:28787.7\n",
      "Best RMSE on eval: 28787.72 with 50 rounds\n",
      "mean_absolute_error=18173.15\n",
      "Model export success: /mnt/kubeflow-gcfs/models/trained_ames_model.dat\n"
     ]
    }
   ],
   "source": [
    "pod_spec = builder.generate_pod_spec()\n",
    "pvc_mutator = fairing_util.add_pvc_mutator(\"kubeflow-gcfs\", \"/mnt/kubeflow-gcfs\")\n",
    "train_deployer = job.job.Job(namespace=\"kubeflow\", \n",
    "                             cleanup=False,\n",
    "                             pod_spec_mutators=[\n",
    "                             fairing.cloud.gcp.add_gcp_credentials_if_exists, pvc_mutator])\n",
    "\n",
    "# Add command line arguments\n",
    "pod_spec.containers[0].command.extend([\"train\", train_data, model_file])\n",
    "result = train_deployer.deploy(pod_spec)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apiVersion: v1\r\n",
      "items:\r\n",
      "- apiVersion: batch/v1\r\n",
      "  kind: Job\r\n",
      "  metadata:\r\n",
      "    creationTimestamp: \"2019-05-21T02:24:50Z\"\r\n",
      "    generateName: fairing-job-\r\n",
      "    labels:\r\n",
      "      fairing-deployer: job\r\n",
      "      fairing-id: 9c53eafe-7b6f-11e9-bcfd-0a580a000143\r\n",
      "    name: fairing-job-c695b\r\n",
      "    namespace: kubeflow\r\n",
      "    resourceVersion: \"14533956\"\r\n",
      "    selfLink: /apis/batch/v1/namespaces/kubeflow/jobs/fairing-job-c695b\r\n",
      "    uid: 9c5fd4b8-7b6f-11e9-8964-42010a8e00ff\r\n",
      "  spec:\r\n",
      "    backoffLimit: 6\r\n",
      "    completions: 1\r\n",
      "    parallelism: 1\r\n",
      "    selector:\r\n",
      "      matchLabels:\r\n",
      "        controller-uid: 9c5fd4b8-7b6f-11e9-8964-42010a8e00ff\r\n",
      "    template:\r\n",
      "      metadata:\r\n",
      "        creationTimestamp: null\r\n",
      "        labels:\r\n",
      "          controller-uid: 9c5fd4b8-7b6f-11e9-8964-42010a8e00ff\r\n",
      "          fairing-deployer: job\r\n",
      "          fairing-id: 9c53eafe-7b6f-11e9-bcfd-0a580a000143\r\n",
      "          job-name: fairing-job-c695b\r\n",
      "        name: fairing-deployer\r\n",
      "      spec:\r\n",
      "        containers:\r\n",
      "        - command:\r\n",
      "          - python\r\n",
      "          - /app/ames-xgboost-build-train-deploy.py\r\n",
      "          - train\r\n",
      "          - /mnt/kubeflow-gcfs/data/ames_dataset/train.csv\r\n",
      "          - /mnt/kubeflow-gcfs/models/trained_ames_model.dat\r\n",
      "          env:\r\n",
      "          - name: FAIRING_RUNTIME\r\n",
      "            value: \"1\"\r\n",
      "          - name: GOOGLE_APPLICATION_CREDENTIALS\r\n",
      "            value: /etc/secrets/user-gcp-sa.json\r\n",
      "          image: gcr.io/code-search-demo/fairing-job/fairing-job:77105FE\r\n",
      "          imagePullPolicy: IfNotPresent\r\n",
      "          name: model\r\n",
      "          resources: {}\r\n",
      "          securityContext:\r\n",
      "            runAsUser: 0\r\n",
      "          terminationMessagePath: /dev/termination-log\r\n",
      "          terminationMessagePolicy: File\r\n",
      "          volumeMounts:\r\n",
      "          - mountPath: /etc/secrets\r\n",
      "            name: user-gcp-sa\r\n",
      "            readOnly: true\r\n",
      "          - mountPath: /mnt/kubeflow-gcfs\r\n",
      "            name: kubeflow-gcfs\r\n",
      "          workingDir: /app\r\n",
      "        dnsPolicy: ClusterFirst\r\n",
      "        restartPolicy: Never\r\n",
      "        schedulerName: default-scheduler\r\n",
      "        securityContext: {}\r\n",
      "        terminationGracePeriodSeconds: 30\r\n",
      "        volumes:\r\n",
      "        - name: user-gcp-sa\r\n",
      "          secret:\r\n",
      "            defaultMode: 420\r\n",
      "            secretName: user-gcp-sa\r\n",
      "        - name: kubeflow-gcfs\r\n",
      "          persistentVolumeClaim:\r\n",
      "            claimName: kubeflow-gcfs\r\n",
      "  status:\r\n",
      "    completionTime: \"2019-05-21T02:24:58Z\"\r\n",
      "    conditions:\r\n",
      "    - lastProbeTime: \"2019-05-21T02:24:58Z\"\r\n",
      "      lastTransitionTime: \"2019-05-21T02:24:58Z\"\r\n",
      "      status: \"True\"\r\n",
      "      type: Complete\r\n",
      "    startTime: \"2019-05-21T02:24:50Z\"\r\n",
      "    succeeded: 1\r\n",
      "kind: List\r\n",
      "metadata:\r\n",
      "  resourceVersion: \"\"\r\n",
      "  selfLink: \"\"\r\n"
     ]
    }
   ],
   "source": [
    "!kubectl get jobs -l fairing-id={train_deployer.job_id} -o yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Deploy the trained model to Kubeflow for predictions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:root:Cluster endpoint: http://fairing-service-hncnr.kubeflow.svc.cluster.local\n"
     ]
    }
   ],
   "source": [
    "from fairing.deployers import serving\n",
    "import fairing_util\n",
    "pod_spec = builder.generate_pod_spec()\n",
    "pvc_mutator = fairing_util.add_pvc_mutator(\"kubeflow-gcfs\", \"/mnt/kubeflow-gcfs\")\n",
    "\n",
    "module_name = os.path.splitext(preprocessor.executable.name)[0]\n",
    "deployer = serving.serving.Serving(module_name + \".HousingServe\",\n",
    "                                   service_type=\"ClusterIP\",\n",
    "                                   labels={\"app\": \"ames\"})\n",
    "    \n",
    "pvc_mutator(None, pod_spec, deployer.namespace)\n",
    "pod_spec.containers[0].env.append({\"name\": \"MODEL_FILE\", \"value\": model_file})\n",
    "url = deployer.deploy(pod_spec)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apiVersion: extensions/v1beta1\r\n",
      "kind: Deployment\r\n",
      "metadata:\r\n",
      "  annotations:\r\n",
      "    deployment.kubernetes.io/revision: \"1\"\r\n",
      "  creationTimestamp: \"2019-05-21T02:25:03Z\"\r\n",
      "  generateName: fairing-deployer-\r\n",
      "  generation: 1\r\n",
      "  labels:\r\n",
      "    app: ames\r\n",
      "    fairing-deployer: serving\r\n",
      "    fairing-id: a3cc6e5a-7b6f-11e9-bcfd-0a580a000143\r\n",
      "  name: fairing-deployer-rkwgf\r\n",
      "  namespace: kubeflow\r\n",
      "  resourceVersion: \"14534016\"\r\n",
      "  selfLink: /apis/extensions/v1beta1/namespaces/kubeflow/deployments/fairing-deployer-rkwgf\r\n",
      "  uid: a3ce3fea-7b6f-11e9-8964-42010a8e00ff\r\n",
      "spec:\r\n",
      "  progressDeadlineSeconds: 600\r\n",
      "  replicas: 1\r\n",
      "  revisionHistoryLimit: 10\r\n",
      "  selector:\r\n",
      "    matchLabels:\r\n",
      "      app: ames\r\n",
      "      fairing-deployer: serving\r\n",
      "      fairing-id: a3cc6e5a-7b6f-11e9-bcfd-0a580a000143\r\n",
      "  strategy:\r\n",
      "    rollingUpdate:\r\n",
      "      maxSurge: 25%\r\n",
      "      maxUnavailable: 25%\r\n",
      "    type: RollingUpdate\r\n",
      "  template:\r\n",
      "    metadata:\r\n",
      "      creationTimestamp: null\r\n",
      "      labels:\r\n",
      "        app: ames\r\n",
      "        fairing-deployer: serving\r\n",
      "        fairing-id: a3cc6e5a-7b6f-11e9-bcfd-0a580a000143\r\n",
      "      name: fairing-deployer\r\n",
      "    spec:\r\n",
      "      containers:\r\n",
      "      - command:\r\n",
      "        - seldon-core-microservice\r\n",
      "        - ames-xgboost-build-train-deploy.HousingServe\r\n",
      "        - REST\r\n",
      "        - --service-type=MODEL\r\n",
      "        - --persistence=0\r\n",
      "        env:\r\n",
      "        - name: FAIRING_RUNTIME\r\n",
      "          value: \"1\"\r\n",
      "        - name: MODEL_FILE\r\n",
      "          value: /mnt/kubeflow-gcfs/models/trained_ames_model.dat\r\n",
      "        image: gcr.io/code-search-demo/fairing-job/fairing-job:77105FE\r\n",
      "        imagePullPolicy: IfNotPresent\r\n",
      "        name: model\r\n",
      "        resources: {}\r\n",
      "        securityContext:\r\n",
      "          runAsUser: 0\r\n",
      "        terminationMessagePath: /dev/termination-log\r\n",
      "        terminationMessagePolicy: File\r\n",
      "        volumeMounts:\r\n",
      "        - mountPath: /mnt/kubeflow-gcfs\r\n",
      "          name: kubeflow-gcfs\r\n",
      "        workingDir: /app\r\n",
      "      dnsPolicy: ClusterFirst\r\n",
      "      restartPolicy: Always\r\n",
      "      schedulerName: default-scheduler\r\n",
      "      securityContext: {}\r\n",
      "      terminationGracePeriodSeconds: 30\r\n",
      "      volumes:\r\n",
      "      - name: kubeflow-gcfs\r\n",
      "        persistentVolumeClaim:\r\n",
      "          claimName: kubeflow-gcfs\r\n",
      "status:\r\n",
      "  availableReplicas: 1\r\n",
      "  conditions:\r\n",
      "  - lastTransitionTime: \"2019-05-21T02:25:08Z\"\r\n",
      "    lastUpdateTime: \"2019-05-21T02:25:08Z\"\r\n",
      "    message: Deployment has minimum availability.\r\n",
      "    reason: MinimumReplicasAvailable\r\n",
      "    status: \"True\"\r\n",
      "    type: Available\r\n",
      "  - lastTransitionTime: \"2019-05-21T02:25:03Z\"\r\n",
      "    lastUpdateTime: \"2019-05-21T02:25:08Z\"\r\n",
      "    message: ReplicaSet \"fairing-deployer-rkwgf-578748c55\" has successfully progressed.\r\n",
      "    reason: NewReplicaSetAvailable\r\n",
      "    status: \"True\"\r\n",
      "    type: Progressing\r\n",
      "  observedGeneration: 1\r\n",
      "  readyReplicas: 1\r\n",
      "  replicas: 1\r\n",
      "  updatedReplicas: 1\r\n"
     ]
    }
   ],
   "source": [
    "!kubectl get deploy -o yaml {deployer.deployment.metadata.name}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Call the prediction endpoint\n",
    "\n",
    "Create a test dataset, then call the endpoint on Kubeflow for predictions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "(train_X, train_y), (test_X, test_y) = ames.read_input(\"ames_dataset/train.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(b'{\"data\":{\"names\":[\"t:0\",\"t:1\"],\"tensor\":{\"shape\":[1,2],\"values\":[165164.875,'\n",
      " b'111924.984375]}},\"meta\":{}}\\n')\n"
     ]
    }
   ],
   "source": [
    "full_url = url + \":5000/predict\"\n",
    "result = fairing_util.predict_nparray(full_url, test_X)\n",
    "pprint.pprint(result.content)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Clean up the prediction endpoint\n",
    "\n",
    "Delete the prediction endpoint created by this notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !kubectl delete service -l app=ames\n",
    "# !kubectl delete deploy -l app=ames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build a simple 1 step pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "EXPERIMENT_NAME = 'Ames'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define the pipeline\n",
    "Pipeline function has to be decorated with the `@dsl.pipeline` decorator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "   name='Training pipeline',\n",
    "   description='A pipeline that trains an xgboost model for the Ames dataset.'\n",
    ")\n",
    "def train_pipeline(\n",
    "   train_data=\"gs://code-search-demo_ames/data/ames_dataset/train.csv\",\n",
    "   model_file=\"gs://code-search-demo_ames/output/hello-world1.txt\",\n",
    "):      \n",
    "    command=[\"python\", preprocessor.executable.name, \"train\", train_data, model_file]\n",
    "    train_op = dsl.ContainerOp(\n",
    "            name=\"train\", \n",
    "            image=builder.image_tag,        \n",
    "            command=command,\n",
    "            ).apply(\n",
    "                gcp.use_gcp_secret('user-gcp-sa'),\n",
    "            )\n",
    "    train_op.container.working_dir = \"/app\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Compile the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_func = train_pipeline\n",
    "pipeline_filename = pipeline_func.__name__ + '.pipeline.zip'\n",
    "compiler.Compiler().compile(pipeline_func, pipeline_filename)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Submit the pipeline for execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "Experiment link <a href=\"/pipeline/#/experiments/details/b96e70b2-e551-442d-9e04-ac35551cd07c\" target=\"_blank\" >here</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "Run link <a href=\"/pipeline/#/runs/details/b8e5652a-7b6f-11e9-8964-42010a8e00ff\" target=\"_blank\" >here</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#Specify pipeline argument values\n",
    "arguments = {\"train_data\": \"gs://code-search-demo_ames/data/ames_dataset/train.csv\",\n",
    "             \"model_file\": \"gs://code-search-demo_ames/output/hello-world1.txt\"}\n",
    "\n",
    "# Get or create an experiment and submit a pipeline run\n",
    "client = kfp.Client()\n",
    "experiment = client.create_experiment(EXPERIMENT_NAME)\n",
    "\n",
    "#Submit a pipeline run\n",
    "run_name = pipeline_func.__name__ + ' run'\n",
    "run_result = client.run_pipeline(experiment.id, run_name, pipeline_filename, arguments)\n",
    "\n",
    "#vvvvvvvvv This link leads to the run information page. (Note: There is a bug in JupyterLab that modifies the URL and makes the link stop working)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define a pipeline for CI/CD\n",
    "\n",
    "* Define a pipeline that trains the model\n",
    "* Then deploy the model and verify its working\n",
    "* If the model is good we create a PR updating the model in the deployment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "Experiment link <a href=\"/pipeline/#/experiments/details/0eb82fbc-94a7-4224-a425-893f4d7dd216\" target=\"_blank\" >here</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "Run link <a href=\"/pipeline/#/runs/details/b902baf8-7b6f-11e9-8964-42010a8e00ff\" target=\"_blank\" >here</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "CICD_EXPERIMENT_NAME = 'Ames CICD'\n",
    "@dsl.pipeline(\n",
    "   name='Ames CICD pipeline',\n",
    "   description='A pipeline that trains an xgboost model for the Ames dataset and updates it.'\n",
    ")\n",
    "def cicd_pipeline(\n",
    "   train_data=\"gs://code-search-demo_ames/data/ames_dataset/train.csv\",\n",
    "   model_file=\"gs://code-search-demo_ames/output/default.txt\",\n",
    "):  \n",
    "    command=[\"python\", preprocessor.executable.name, \"train\", train_data, model_file]\n",
    "    train_op = dsl.ContainerOp(\n",
    "            name=\"train\", \n",
    "            image=builder.image_tag,        \n",
    "            command=command,\n",
    "            ).apply(\n",
    "                gcp.use_gcp_secret('user-gcp-sa'),\n",
    "            )\n",
    "    train_op.container.working_dir = \"/app\"\n",
    "    \n",
    "    command=[\"python3\", preprocessor.executable.name, \"deploy-model\", \n",
    "             model_file]\n",
    "    deploy_op = dsl.ContainerOp(\n",
    "            name=\"deploy-model\", \n",
    "            image=builder.image_tag,        \n",
    "            command=command,\n",
    "            )\n",
    "    deploy_op.container.working_dir = \"/app\"\n",
    "    deploy_op.after(train_op)\n",
    "\n",
    "    command=[\"python3\", preprocessor.executable.name, \"validate-model\", \n",
    "             model_file]    \n",
    "    validate_op = dsl.ContainerOp(\n",
    "            name=\"validate-model\", \n",
    "            image=builder.image_tag,        \n",
    "            command=command,\n",
    "            )\n",
    "    validate_op.container.working_dir = \"/app\"\n",
    "    validate_op.after(deploy_op)\n",
    "\n",
    "    command=[\"python3\", preprocessor.executable.name, \"create-pr-to-update-model\", \n",
    "             \"deployment/update_model_job.yaml\", model_file]    \n",
    "    pr_op = dsl.ContainerOp(\n",
    "            name=\"create-pr\", \n",
    "            image=builder.image_tag,        \n",
    "            command=command,\n",
    "            )\n",
    "    pr_op.container.working_dir = \"/app\"  \n",
    "    pr_op.after(validate_op)\n",
    "\n",
    "pipeline_func = cicd_pipeline\n",
    "pipeline_filename = pipeline_func.__name__ + '.pipeline.zip'\n",
    "compiler.Compiler().compile(pipeline_func, pipeline_filename)\n",
    "\n",
    "import datetime\n",
    "gcs_model_file = \"gs://code-search-demo_ames/models/\" + datetime.datetime.now().strftime(\"%y%m%d_%H%M%S\")\n",
    "\n",
    "#Specify pipeline argument values\n",
    "arguments = {\"train_data\": \"gs://code-search-demo_ames/data/ames_dataset/train.csv\",\n",
    "             \"model_file\": gcs_model_file}\n",
    "\n",
    "# Get or create an experiment and submit a pipeline run\n",
    "client = kfp.Client()\n",
    "experiment = client.create_experiment(CICD_EXPERIMENT_NAME)\n",
    "\n",
    "#Submit a pipeline run\n",
    "run_name = pipeline_func.__name__ + ' run'\n",
    "run_result = client.run_pipeline(experiment.id, run_name, pipeline_filename, arguments)\n"
   ]
  }
 ],
 "metadata": {
  "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}