UseCases/Vertex_AI_Integration/Vertex_AI_Integration_Python.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "entire-mountain",
   "metadata": {},
   "source": [
    "# Vertex Pipelines - Housing Example\n",
    "## Overview"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "tracked-thermal",
   "metadata": {},
   "source": [
    "Vertex AI is Google's single environment for data scientists to develop and deploy ML models, from experimentation, to deployment, to managing and monitoring models. In this tutorial, we will show how to integrate Vantage Analytics capabilites in Vertex AI ML Pipelines. We'll create two pipelines:\n",
    "1. Training - the first will be a three step pipeline to train and deploy a model; the first step transforms data in Vantage and then exports a file for training, the second step trains a model using scikit-learn, and the final step deploys the model to Vantage using BYOM. \n",
    "\n",
    "    ![Training Pipeline](pipeline1.png)\n",
    "2. Scoring - the second pipeline will use the model created by the the first pipeline to score new data stored in a table on Vantage.\n",
    "\n",
    "    ![Scoring Pipeline](pipeline2.png)\n",
    "\n",
    "Both pipelines are very simple, but the first pipeline could be triggered to retrain a model with new data when the production model has drifted.  The second pipeline could be run on a regular schedule when new data for scoring was available."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "european-silicon",
   "metadata": {},
   "source": [
    "## Prerequisites"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "municipal-perspective",
   "metadata": {},
   "source": [
    "* GCP account - register [here](https://cloud.google.com/gcp)\n",
    "* Kaggle account - register [here](https://www.kaggle.com/account/login?phase=startRegisterTab)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "simplified-mailman",
   "metadata": {},
   "source": [
    "## Setting up Vantage and loading data\n",
    "\n",
    "### Setup the notebook environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "organic-jersey",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "!{sys.executable} -m pip install --upgrade --force-reinstall ipython-sql\n",
    "!{sys.executable} -m pip install teradatasqlalchemy teradataml kaggle ipython-sql kfp"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "decreased-helping",
   "metadata": {},
   "source": [
    "### Setup a Vantage instance\n",
    "Follow the [Run Vantage Express on GCP](https://quickstarts.teradata.com/vantage.express.gcp.html) how-to to get Vantage setup. Make sure to follow the instructions to open the VM up to the Internet.\n",
    "\n",
    "### Create GCS bucket\n",
    "You will need a GCS bucket to store artifacts managed by KubeFlow. \n",
    "\n",
    "Define the bucket name:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "saving-conditions",
   "metadata": {},
   "outputs": [],
   "source": [
    "BUCKET_NAME = \"<your-bucket-name>\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "frequent-liability",
   "metadata": {},
   "source": [
    "If the bucket doesn't exist, go ahead and create it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "lovely-pottery",
   "metadata": {},
   "outputs": [],
   "source": [
    "!gsutil ls -b gs://$BUCKET_NAME || gsutil mb gs://$BUCKET_NAME"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "innovative-lender",
   "metadata": {},
   "source": [
    "### Give permissions to Vertex AI to access your bucket\n",
    "Go to [IAM tab in GCS console](https://console.cloud.google.com/iam-admin/iam) and assign `Storage Admin` role to your default Compute Engine. The principal of the default Compute Engine account looks like this: `<project-id>-compute@developer.gserviceaccount.com`.\n",
    "\n",
    "### Download sample data\n",
    "We'll use the Boston Housing dataset which can be obtained from [Kaggle](https://www.kaggle.com/).\n",
    "Login to your Kaggle account. In the top right corner click on your user icon and select `Account`. Find `API` section and click on `Create New API Token`. This will produce `kaggle.json` file. Open `kaggle.json` and copy the username and key. Substitute the values in the cell and run it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fundamental-alpha",
   "metadata": {},
   "outputs": [],
   "source": [
    "%env KAGGLE_USERNAME=<your-kaggle-username>\n",
    "%env KAGGLE_KEY=<your-kaggle-key>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "protecting-christianity",
   "metadata": {},
   "outputs": [],
   "source": [
    "!kaggle datasets download -f housing.csv vikrishnan/boston-house-prices"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "contemporary-possibility",
   "metadata": {},
   "source": [
    "### Load training data to Vantage\n",
    "Let's setup `DATABASE_URL` environment variable that will point to your instance of Vantage. Make sure that you default to `mldb` database where `BYOM` package is installed, e.g.:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "applied-assignment",
   "metadata": {},
   "outputs": [],
   "source": [
    "DATABASE_URL='teradatasql://dbc:dbc@34.121.78.209/mldb'\n",
    "%env DATABASE_URL=$DATABASE_URL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "private-prefix",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas\n",
    "import os\n",
    "\n",
    "df=pandas.read_fwf('housing.csv', names=['CRIM', 'ZN', 'INDUS', 'CHAS', 'NOX', 'RM', 'AGE', 'DIS', 'RAD', 'TAX', 'PTRATIO', 'B', 'LSTAT', 'MEDV'])\n",
    "df.to_sql('housing', con=DATABASE_URL, index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dutch-growth",
   "metadata": {},
   "source": [
    "For this tutiorial we need a table to store the trained model and another table with some new data that we want to score with our model. Use teradatasql to execute the following SQL on your Vantage instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "sapphire-struggle",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "CREATE SET TABLE demo_models (model_id VARCHAR (30), model BLOB) PRIMARY INDEX (model_id);\n",
    "CREATE SET TABLE test_housing (ID INTEGER, CRIM FLOAT, ZN FLOAT,INDUS FLOAT,CHAS INTEGER,NOX FLOAT,RM FLOAT, \n",
    "    AGE FLOAT,DIS FLOAT, RAD INTEGER,TAX INTEGER,PTRATIO FLOAT,B FLOAT,LSTAT FLOAT) PRIMARY INDEX (CRIM);\n",
    "INSERT INTO test_housing (ID, CRIM, ZN, INDUS, CHAS, NOX, RM, AGE, DIS, RAD, TAX, PTRATIO, B, LSTAT) \n",
    "    VALUES (1,.02,0.0,7.07,0,.46,6.4,78.9,4.9,2,242,17.8,396.9,9.14);"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "congressional-opera",
   "metadata": {},
   "source": [
    "## The first pipeline to train and deploy a model using Kubeflow"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "played-appointment",
   "metadata": {},
   "source": [
    "Now we are ready to create the components in the pipeline.  Vertex AI Pipelines can run pipelines built using the Kubeflow Pipelines SDK or TensorFlow Extended.  We'll be using the Kubeflow Pipelines SDK for this simple example using scikit-learn.\n",
    "\n",
    "In this example we will create the following three components:\n",
    "* `read_data_from_vantage`\n",
    "    * input: ipaddr of the VM hosting Vantage\n",
    "    * output: csv file with the data for training and testing\n",
    "* `train_model`\n",
    "    * input: csv file with data for training and testing\n",
    "    * output: file containing the model\n",
    "    * output: Metric artifact with model performance\n",
    "* `deploy_model`\n",
    "    * input: file containing the model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "classified-brain",
   "metadata": {},
   "source": [
    "First, import the Kubeflow Pipeline component and dsl packages."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "pediatric-pavilion",
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp.v2.dsl as dsl\n",
    "from kfp.v2.dsl import (\n",
    "    component,\n",
    "    Input,\n",
    "    Output,\n",
    "    Dataset,\n",
    "    Model,\n",
    "    Metrics,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "satisfactory-novel",
   "metadata": {},
   "source": [
    "## Create the component that reads data from Vantage"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "floating-antigua",
   "metadata": {},
   "source": [
    "The first component reads data from a Vantage warehouse (see above and make sure you have set up Vantage Express in GCP including opening up a firewall to the VM so you can access Vantage from the Internet.)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "vulnerable-federation",
   "metadata": {},
   "source": [
    "The component connects to Vantage using the connection string passed as an input parameter, reads the rows from the table mldb.housing in Vantage and then outputs the data to an `Output[Dataset]`. The Output is  a temporary file used to pass data between components (see more about passing data between components [here](https://www.kubeflow.org/docs/components/pipelines/sdk-v2/python-function-components/#passing-artifacts-by-file)).\n",
    "\n",
    "The component uses sqlalchemy to talk to Teradata. Each component is run in a separate container on Kubernetes so all import statements need to be done within the component. We have created a base image with teradatasql already installed. When you pass `base_image='python'` the component will use that image to create a container. `packages_to_install` parameter defines what other packages the component needs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "oriental-assignment",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(base_image='python', packages_to_install=['teradatasqlalchemy'])\n",
    "def read_data_from_vantage(\n",
    "    connection_string: str,\n",
    "    output_file: Output[Dataset]\n",
    "):\n",
    "    import sqlalchemy\n",
    "    \n",
    "    file_name = output_file.path\n",
    "    engine = sqlalchemy.create_engine(connection_string)    \n",
    "    \n",
    "    with engine.connect() as con:\n",
    "        rs = con.execute('SELECT * FROM housing')\n",
    "        with open(output_file.path, 'w') as output_file:\n",
    "            output_file.write('CRIM,ZN,INDUS,CHAS,NOX,RM,AGE,DIS,RAD,TAX,PTRATIO,B,LSTAT,MEDV\\n')\n",
    "            for row in rs:\n",
    "                output_file.write(','.join([str(i) for i in row]) + '\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "floating-mechanism",
   "metadata": {},
   "source": [
    "## Create the train model component"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "annual-campbell",
   "metadata": {},
   "source": [
    "Next we'll create a component to train a model with the training data.\n",
    "\n",
    "The input into this component is the file from the previous component. The output is the file with the trained model using joblib.dump and a file with the test data.\n",
    "\n",
    "The component will use scikit-learn and pandas so we need to pass `packages_to_install=['pandas==1.3.5','scikit-learn']` - this will tell Kubeflow to install the packages when the container is created."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "federal-fossil",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(base_image='teradata/python-sklearn2pmml', packages_to_install=['pandas==1.3.5','scikit-learn','sklearn-pandas==1.5.0'])\n",
    "def train_model(\n",
    "    input_file : Input[Dataset],\n",
    "    output_model: Output[Model],\n",
    "    output_metrics: Output[Metrics]\n",
    "):\n",
    "    import pandas as pd\n",
    "    from sklearn.model_selection import train_test_split\n",
    "    from sklearn.ensemble import RandomForestRegressor\n",
    "    from sklearn.preprocessing import StandardScaler\n",
    "    from sklearn import metrics\n",
    "    from sklearn_pandas import DataFrameMapper\n",
    "    import joblib\n",
    "    from sklearn2pmml.pipeline import PMMLPipeline\n",
    "    from sklearn2pmml import sklearn2pmml\n",
    " \n",
    "    df = pd.read_csv(input_file.path)\n",
    "    \n",
    "    train, test = train_test_split(df, test_size = .33)\n",
    "    train = train.apply(pd.to_numeric, errors='ignore')\n",
    "    test = test.apply(pd.to_numeric, errors='ignore')\n",
    "\n",
    "    target = 'MEDV'\n",
    "    features = train.columns.drop(target)\n",
    "\n",
    "    \n",
    "    pipeline = PMMLPipeline([\n",
    "    (\"mapping\", DataFrameMapper([\n",
    "    (['CRIM', 'ZN', 'INDUS', 'CHAS', 'NOX', 'RM', 'AGE', 'DIS', 'RAD', 'TAX', 'PTRATIO', 'B', 'LSTAT'], StandardScaler())\n",
    "    ])),\n",
    "    (\"rfc\", RandomForestRegressor(n_estimators = 100, random_state = 0))\n",
    "    ])\n",
    "    \n",
    "    pipeline.fit(train[features], train[target])\n",
    "    y_pred = pipeline.predict(test[features])\n",
    "\n",
    "    metric_accuracy = metrics.mean_squared_error(y_pred,test[target])\n",
    "    output_metrics.log_metric('accuracy', metric_accuracy)\n",
    "    output_model.metadata['accuracy'] = metric_accuracy\n",
    "        \n",
    "    joblib.dump(pipeline, output_model.path)\n",
    "  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "alternate-california",
   "metadata": {},
   "source": [
    "## Create component to deploy model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "preceding-psychiatry",
   "metadata": {},
   "source": [
    "The last component loads the model and tests it on the test data. The `Output[Metrics]` creates an artifact with the models performance and can be visualize in the Runtime Graph."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "another-tradition",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(base_image='teradata/python-sklearn2pmml')\n",
    "def deploy_model(\n",
    "    connection_string: str,\n",
    "    input_model : Input[Model],\n",
    "):\n",
    "    import sqlalchemy\n",
    "    import teradataml as tdml\n",
    "    import joblib\n",
    "    from sklearn2pmml.pipeline import PMMLPipeline\n",
    "    from sklearn2pmml import sklearn2pmml\n",
    "    \n",
    "    engine = sqlalchemy.create_engine(connection_string)\n",
    "    tdml.create_context(tdsqlengine = engine)\n",
    "    \n",
    "    pipeline = joblib.load(input_model.path)\n",
    "    \n",
    "    sklearn2pmml(pipeline, \"test_local.pmml\", with_repr = True)\n",
    "        \n",
    "    model_id = 'housing_rf'\n",
    "    model_file = 'test_local.pmml'\n",
    "    table_name = 'demo_models'\n",
    "    \n",
    "    tdml.configure.byom_install_location = \"mldb\"\n",
    "    \n",
    "    try:\n",
    "        res = tdml.save_byom(model_id = model_id, model_file = model_file, table_name = table_name)\n",
    "    except Exception as e:\n",
    "        # if our model exists, delete and rewrite\n",
    "        if str(e.args).find('TDML_2200') >= 1:\n",
    "            res = tdml.delete_byom(model_id = model_id, table_name = table_name)\n",
    "            res = tdml.save_byom(model_id = model_id, model_file = model_file, table_name = table_name)\n",
    "            pass\n",
    "        else:\n",
    "            raise\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "differential-mobility",
   "metadata": {},
   "source": [
    "## Create function for executing the pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "baking-verse",
   "metadata": {},
   "source": [
    "Now we'll create a function to execute each component in the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "regulated-figure",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "   name='run-vantage-pipeline',\n",
    "   description='An example pipeline that connects to Vantage.',\n",
    ")\n",
    "def run_vantage_pipeline_vertex(\n",
    "   connection_string: str\n",
    "):\n",
    "    data_file = read_data_from_vantage(connection_string).output\n",
    "    test_model_data = train_model(data_file)\n",
    "    deploy_model(connection_string,test_model_data.outputs['output_model'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "hollow-buddy",
   "metadata": {},
   "source": [
    "Compile the pipeline.  The pipline will be saved in a json file with the name provided as the `package_path`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "moved-isaac",
   "metadata": {},
   "outputs": [],
   "source": [
    "from kfp.v2 import compiler\n",
    "compiler.Compiler().compile(pipeline_func=run_vantage_pipeline_vertex,\n",
    "    package_path='train_housing_pipeline.json')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "danish-clearance",
   "metadata": {},
   "source": [
    "Now use the Vertex AI client to execute the pipeline.  Import the `google.cloud.aiplatform` package. Vertex AI needs a Cloud Storage bucket to for temporary files.  Create a new job using the json file above and pass the ipaddr as the parameter.  Then submit the job.\n",
    "\n",
    "When the job starts a link will appear that will take you to the Runtime Graph.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "solid-ireland",
   "metadata": {},
   "outputs": [],
   "source": [
    "import google.cloud.aiplatform as aip\n",
    "\n",
    "pipeline_root_path = 'gs://' + BUCKET_NAME\n",
    "\n",
    "job = aip.PipelineJob(\n",
    "    display_name=\"housing_training_deploy\",\n",
    "    template_path=\"train_housing_pipeline.json\",\n",
    "    pipeline_root=pipeline_root_path,\n",
    "    parameter_values={\n",
    "        'connection_string': DATABASE_URL\n",
    "    }\n",
    ")\n",
    "\n",
    "job.submit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bigger-plasma",
   "metadata": {},
   "source": [
    "## Inspect model metrics\n",
    "\n",
    "When the pipeline has completed running (each component in the graph should have a green check mark). You can click on each component to see details of the execution and the logs created. If you click on the `output_metrics` artifact, in the `Pipeline run analysis` window the `Node Info` will show the accuracy of the model. Yyou can learn more about other metrics you can pass and visulation using the Metrics artifict [here](https://www.kubeflow.org/docs/components/pipelines/sdk/output-viewer/#introduction).)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "sapphire-hughes",
   "metadata": {},
   "source": [
    "## Test the deployed model\n",
    "Let's test the model we have just deployed by scoring some new data.  We'll use the `teradataml` driver to retrieve the saved model and score the rows in a table with new data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "handled-theorem",
   "metadata": {},
   "outputs": [],
   "source": [
    "import teradataml as tdml\n",
    "import sqlalchemy\n",
    "import os\n",
    "\n",
    "engine = sqlalchemy.create_engine(DATABASE_URL)\n",
    "eng = tdml.create_context(tdsqlengine = engine)\n",
    "\n",
    "#indicate the database that BYOM is using\n",
    "tdml.configure.byom_install_location = \"mldb\"\n",
    "\n",
    "tdf_test = tdml.DataFrame('test_housing')\n",
    "\n",
    "modeldata = tdml.retrieve_byom(\"housing_rf\", table_name=\"demo_models\")\n",
    "\n",
    "predictions = tdml.PMMLPredict(\n",
    "        modeldata = modeldata,\n",
    "        newdata = tdf_test,\n",
    "        accumulate = ['ID']\n",
    "        )\n",
    "predictions.result.to_pandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "powerful-study",
   "metadata": {},
   "source": [
    "## Create a new pipeline to score new data"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "atlantic-favorite",
   "metadata": {},
   "source": [
    "This pipeline will have only one component that uses the teradatasql driver to execute a SQL query that retrieves the model from the `demo_model` table and scores the rows in the `test_housing` table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "formal-packing",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(base_image='teradata/python-sklearn2pmml', packages_to_install=['pandas==1.3.5','scikit-learn'])\n",
    "def score_new_data(\n",
    "    connection_string: str,\n",
    "    model_name: str,\n",
    "    model_table: str,\n",
    "    data_table: str,\n",
    "    prediction_table: str  \n",
    "):\n",
    "    import teradataml as tdml\n",
    "    import sqlalchemy\n",
    "        \n",
    "    engine = sqlalchemy.create_engine(connection_string)\n",
    "    \n",
    "    with engine.connect() as con:\n",
    "        con.execute(f'CREATE TABLE {prediction_table} AS (SELECT * FROM mldb.PMMLPredict ( ON {data_table} ON (SELECT * FROM {model_table} where model_id=\\'{model_name}\\') DIMENSION USING Accumulate (\\'ID\\')) AS td ) WITH DATA')\n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "valuable-apache",
   "metadata": {},
   "source": [
    "The `run_new_data_score` pipeline takes the following parameters:\n",
    "- `model_name`: ID of the model\n",
    "- `model_table`: the name of the table storing the model\n",
    "- `data_table`: the name of the table with new data to score\n",
    "- `prediction_table`: the name of the table to store the scoring results\n",
    "\n",
    "When the pipeline is executed the dashboard will provide fields to enter the values you want to use."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "selected-treasury",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "   name='new-data-pipeline',\n",
    "   description='An example of a component that scores new data with a saved model.',\n",
    ")\n",
    "def run_new_data_score(\n",
    "    connection_string: str,\n",
    "    model_name: str,\n",
    "    model_table: str,\n",
    "    data_table: str,\n",
    "    prediction_table: str\n",
    "):\n",
    "    score_new_data(DATABASE_URL,model_name,model_table,data_table,prediction_table)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bearing-revision",
   "metadata": {},
   "source": [
    "To compile the pipeline run the following code. The pipeline will be saved in `score_new_data_pipeline_sql.json` file. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "round-borough",
   "metadata": {},
   "outputs": [],
   "source": [
    "compiler.Compiler().compile(pipeline_func=run_new_data_score,\n",
    "    package_path='score_new_data_pipeline_sql.json')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "alternate-bearing",
   "metadata": {},
   "source": [
    "We will now execute the pipeline in Vertex AI Pipelines."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "composite-amazon",
   "metadata": {},
   "outputs": [],
   "source": [
    "import google.cloud.aiplatform as aip\n",
    "\n",
    "pipeline_root_path = 'gs://' + BUCKET_NAME\n",
    "\n",
    "job = aip.PipelineJob(\n",
    "    display_name=\"new_data_housing\",\n",
    "    template_path=\"score_new_data_pipeline_sql.json\",\n",
    "    pipeline_root=pipeline_root_path,\n",
    "    parameter_values={\n",
    "        'connection_string': DATABASE_URL,\n",
    "        'model_name': 'housing_rf',\n",
    "        'model_table': 'demo_models',\n",
    "        'data_table': 'test_housing',\n",
    "        'prediction_table': 'housing_predictions'\n",
    "    }\n",
    ")\n",
    "\n",
    "job.submit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "global-premiere",
   "metadata": {},
   "source": [
    "Once the job completes, you can view the batch predictions with:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "compact-tongue",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%sql\n",
    "SELECT * FROM housing_predictions;"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "accessible-sheffield",
   "metadata": {},
   "source": [
    "## Cleanup"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "framed-dividend",
   "metadata": {},
   "source": [
    "To stop incurring charges you need to clean up the following resources:\n",
    "* Delete the Vantage Express VM - go to the list of Compute Engine instances and selecting the instance with Vantage Express and then click on `Delete`.\n",
    "* Delete the storage bucket you configured"
   ]
  }
 ],
 "metadata": {
  "environment": {
   "kernel": "python3",
   "name": "common-cpu.m91",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/base-cpu:m91"
  },
  "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}