notebooks/community/pipelines/google_cloud_pipeline_components_bqml_pipeline_demand_forecasting.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ur8xi4C7S06n"
      },
      "outputs": [],
      "source": [
        "# Copyright 2022 Google LLC\n",
        "#\n",
        "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
        "# you may not use this file except in compliance with the License.\n",
        "# You may obtain a copy of the License at\n",
        "#\n",
        "#     https://www.apache.org/licenses/LICENSE-2.0\n",
        "#\n",
        "# Unless required by applicable law or agreed to in writing, software\n",
        "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
        "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
        "# See the License for the specific language governing permissions and\n",
        "# limitations under the License."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "JAPoU8Sm5E6e"
      },
      "source": [
        "<table align=\"left\">\n",
        "\n",
        "  <td>\n",
        "    <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/community/pipelines/google_cloud_pipeline_components_bqml_pipeline_demand_forecasting.ipynb\">\n",
        "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Colab logo\"> Run in Colab\n",
        "    </a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a href=\"https://github.com/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/community/pipelines/google_cloud_pipeline_components_bqml_pipeline_demand_forecasting.ipynb\">\n",
        "      <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\">\n",
        "      View on GitHub\n",
        "    </a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/community/pipelines/google_cloud_pipeline_components_bqml_pipeline_demand_forecasting.ipynb\">\n",
        "      <img src=\"https://lh3.googleusercontent.com/UiNooY4LUgW_oTvpsNhPpQzsstV5W8F7rYgxgGBD85cWJoLmrOzhVs_ksK_vgx40SHs7jCqkTkCk=e14-rj-sc0xffffff-h130-w32\" alt=\"Vertex AI logo\">\n",
        "      Open in Vertex AI Workbench\n",
        "    </a>\n",
        "  </td>                                                                                               \n",
        "</table>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "239ba71252d3"
      },
      "source": [
        "## Overview\n",
        "\n",
        "This notebook shows how to use `Vertex AI Pipelines` and `BigQuery ML pipeline components` to train and evaluate a demand forecasting model."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "25c28706c23e"
      },
      "source": [
        "### Objective\n",
        "\n",
        "In this tutorial, you learn how to train and evaluate a BigQuery ML model using Vertex AI Pipelines and BigQuery ML pipeline components. \n",
        "\n",
        "This tutorial uses the following Google Cloud ML services and resources:\n",
        "\n",
        "- `Vertex AI Pipelines`\n",
        "- `BigQuery ML pipeline components`\n",
        "\n",
        "The steps performed include:\n",
        "\n",
        "- Define a custom evaluation component\n",
        "- Define a pipeline:\n",
        "  - Get BigQuery training data\n",
        "  - Train a BigQuery Arima Plus model\n",
        "  - Evaluate the BigQuery Arima Plus model\n",
        "  - Plot the evaluations\n",
        "  - Check the model performance\n",
        "  - Generate the ARIMA Plus forecasts\n",
        "  - Generate the ARIMA PLUS forecast explainations\n",
        "- Compile the pipeline.\n",
        "- Execute the pipeline."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "586acfa9b502"
      },
      "source": [
        "### Dataset\n",
        "\n",
        "The dataset is a modified version of the dataset in [Build and visualize demand forecast predictions using Datastream, Dataflow, BigQuery ML, and Looker\n",
        "](https://cloud.google.com/architecture/build-visualize-demand-forecast-prediction-datastream-dataflow-bigqueryml-looker) solution architecture\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tvgnzT1CKxrO"
      },
      "source": [
        "### Costs \n",
        "\n",
        "This tutorial uses billable components of Google Cloud:\n",
        "\n",
        "* Vertex AI\n",
        "* Cloud Storage\n",
        "* BigQuery"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ze4-nDLfK4pw"
      },
      "source": [
        "### Set up your local development environment\n",
        "\n",
        "**If you are using Colab or Vertex AI Workbench Notebooks**, your environment already meets\n",
        "all the requirements to run this notebook. You can skip this step."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "gCuSR8GkAgzl"
      },
      "source": [
        "**Otherwise**, make sure your environment meets this notebook's requirements.\n",
        "You need the following:\n",
        "\n",
        "* The Google Cloud SDK\n",
        "* Git\n",
        "* Python 3\n",
        "* virtualenv\n",
        "* Jupyter notebook running in a virtual environment with Python 3\n",
        "\n",
        "The Google Cloud guide to [Setting up a Python development\n",
        "environment](https://cloud.google.com/python/setup) and the [Jupyter\n",
        "installation guide](https://jupyter.org/install) provide detailed instructions\n",
        "for meeting these requirements. The following steps provide a condensed set of\n",
        "instructions:\n",
        "\n",
        "1. [Install and initialize the Cloud SDK.](https://cloud.google.com/sdk/docs/)\n",
        "\n",
        "1. [Install Python 3.](https://cloud.google.com/python/setup#installing_python)\n",
        "\n",
        "1. [Install\n",
        "   virtualenv](https://cloud.google.com/python/setup#installing_and_using_virtualenv)\n",
        "   and create a virtual environment that uses Python 3. Activate the virtual environment.\n",
        "\n",
        "1. To install Jupyter, run `pip3 install jupyter` on the\n",
        "command-line in a terminal shell.\n",
        "\n",
        "1. To launch Jupyter, run `jupyter notebook` on the command-line in a terminal shell.\n",
        "\n",
        "1. Open this notebook in the Jupyter Notebook Dashboard."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "i7EUnXsZhAGF"
      },
      "source": [
        "### Install additional packages\n",
        "\n",
        "Install the following packages required to execute this notebook."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "2b4ef9b72d43"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "\n",
        "# The Vertex AI Workbench Notebook product has specific requirements\n",
        "IS_WORKBENCH_NOTEBOOK = os.getenv(\"DL_ANACONDA_HOME\")\n",
        "IS_USER_MANAGED_WORKBENCH_NOTEBOOK = os.path.exists(\n",
        "    \"/opt/deeplearning/metadata/env_version\"\n",
        ")\n",
        "\n",
        "# Vertex AI Notebook requires dependencies to be installed with '--user'\n",
        "USER_FLAG = \"\"\n",
        "if IS_WORKBENCH_NOTEBOOK:\n",
        "    USER_FLAG = \"--user\"\n",
        "    \n",
        "! pip3 install --upgrade \"kfp\" \\\n",
        "                         \"google-cloud-aiplatform\" \\\n",
        "                         \"google_cloud_pipeline_components\" \\\n",
        "                         \"google-cloud-bigquery\" {USER_FLAG} -q\n",
        "\n",
        "! pip3 install --upgrade \"tensorflow<2.8.0\" {USER_FLAG} -q"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hhq5zEbGg0XX"
      },
      "source": [
        "### Restart the kernel\n",
        "\n",
        "After you install the additional packages, you need to restart the notebook kernel so it can find the packages."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "EzrelQZ22IZj"
      },
      "outputs": [],
      "source": [
        "# Automatically restart kernel after installs\n",
        "import os\n",
        "\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    # Automatically restart kernel after installs\n",
        "    import IPython\n",
        "\n",
        "    app = IPython.Application.instance()\n",
        "    app.kernel.do_shutdown(True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lWEdiXsJg0XY"
      },
      "source": [
        "## Before you begin"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BF1j6f9HApxa"
      },
      "source": [
        "### Set up your Google Cloud project\n",
        "\n",
        "**The following steps are required, regardless of your notebook environment.**\n",
        "\n",
        "1. [Select or create a Google Cloud project](https://console.cloud.google.com/cloud-resource-manager). When you first create an account, you get a $300 free credit towards your compute/storage costs.\n",
        "\n",
        "1. [Make sure that billing is enabled for your project](https://cloud.google.com/billing/docs/how-to/modify-project).\n",
        "\n",
        "1. [Enable the API](https://console.cloud.google.com/flows/enableapi?apiid=cloudresourcemanager.googleapis.com,aiplatform.googleapis.com,notebooks.googleapis.com).\n",
        "\n",
        "1. If you are running this notebook locally, you will need to install the [Cloud SDK](https://cloud.google.com/sdk).\n",
        "\n",
        "1. Enter your project ID in the cell below. Then run the cell to make sure the\n",
        "Cloud SDK uses the right project for all the commands in this notebook.\n",
        "\n",
        "**Note**: Jupyter runs lines prefixed with `!` as shell commands, and it interpolates Python variables prefixed with `$` into these commands."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WReHDGG5g0XY"
      },
      "source": [
        "#### Set your project ID\n",
        "\n",
        "**If you don't know your project ID**, you may be able to get your project ID using `gcloud`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "oM1iC_MfAts1"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "\n",
        "PROJECT_ID = \"\"\n",
        "\n",
        "# Get your Google Cloud project ID from gcloud\n",
        "if not os.getenv(\"IS_TESTING\"):\n",
        "    shell_output = !gcloud config list --format 'value(core.project)' 2>/dev/null\n",
        "    PROJECT_ID = shell_output[0]\n",
        "    print(\"Project ID: \", PROJECT_ID)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "qJYoRfYng0XZ"
      },
      "source": [
        "Otherwise, set your project ID here."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "riG_qUokg0XZ"
      },
      "outputs": [],
      "source": [
        "if PROJECT_ID == \"\" or PROJECT_ID is None:\n",
        "    PROJECT_ID = \"[your-project]\"  # @param {type:\"string\"}"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "set_gcloud_project_id"
      },
      "outputs": [],
      "source": [
        "! gcloud config set project $PROJECT_ID"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "region"
      },
      "source": [
        "#### Region\n",
        "\n",
        "You can also change the `REGION` variable, which is used for operations\n",
        "throughout the rest of this notebook.  Below are regions supported for Vertex AI. We recommend that you choose the region closest to you.\n",
        "\n",
        "- Americas: `us-central1`\n",
        "- Europe: `europe-west4`\n",
        "- Asia Pacific: `asia-east1`\n",
        "\n",
        "You may not use a multi-regional bucket for training with Vertex AI. Not all regions provide support for all Vertex AI services.\n",
        "\n",
        "Learn more about [Vertex AI regions](https://cloud.google.com/vertex-ai/docs/general/locations)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "R4aDZuADOWzi"
      },
      "outputs": [],
      "source": [
        "REGION = \"[your-region]\"  # @param {type: \"string\"}\n",
        "\n",
        "if REGION == \"[your-region]\":\n",
        "    REGION = \"us-central1\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "06571eb4063b"
      },
      "source": [
        "#### UUID\n",
        "If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a uuid for each instance session, and append it onto the name of resources you create in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "697568e92bd6"
      },
      "outputs": [],
      "source": [
        "import random\n",
        "import string\n",
        "\n",
        "\n",
        "# Generate a uuid of a specifed length(default=8)\n",
        "def generate_uuid(length: int = 8) -> str:\n",
        "    return \"\".join(random.choices(string.ascii_lowercase + string.digits, k=length))\n",
        "\n",
        "\n",
        "UUID = generate_uuid()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dr--iN2kAylZ"
      },
      "source": [
        "### Authenticate your Google Cloud account\n",
        "\n",
        "**If you are using Vertex AI Workbench Notebooks**, your environment is already\n",
        "authenticated. Skip this step."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "sBCra4QMA2wR"
      },
      "source": [
        "**If you are using Colab**, run the cell below and follow the instructions\n",
        "when prompted to authenticate your account via oAuth.\n",
        "\n",
        "**Otherwise**, follow these steps:\n",
        "\n",
        "1. In the Cloud Console, go to the [**Create service account key**\n",
        "   page](https://console.cloud.google.com/apis/credentials/serviceaccountkey).\n",
        "\n",
        "2. Click **Create service account**.\n",
        "\n",
        "3. In the **Service account name** field, enter a name, and\n",
        "   click **Create**.\n",
        "\n",
        "4. In the **Grant this service account access to project** section, click the **Role** drop-down list. Type and select the following roles into the filter box:\n",
        "\n",
        "  - BigQuery Data Editor\n",
        "  - BigQuery Job User\n",
        "  - Service Account User\n",
        "  - Storage Object Admin\n",
        "  - Storgae Admin\n",
        "  - Vertex AI Administrator\n",
        "\n",
        "5. Click *Create*. A JSON file that contains your key downloads to your\n",
        "local environment.\n",
        "\n",
        "6. Enter the path to your service account key as the\n",
        "`GOOGLE_APPLICATION_CREDENTIALS` variable in the cell below and run the cell."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "PyQmSRbKA8r-"
      },
      "outputs": [],
      "source": [
        "# If you are running this notebook in Colab, run this cell and follow the\n",
        "# instructions to authenticate your GCP account. This provides access to your\n",
        "# Cloud Storage bucket and lets you submit training jobs and prediction\n",
        "# requests.\n",
        "\n",
        "import os\n",
        "import sys\n",
        "\n",
        "# If on Vertex AI Workbench, then don't execute this code\n",
        "IS_COLAB = \"google.colab\" in sys.modules\n",
        "if not os.path.exists(\"/opt/deeplearning/metadata/env_version\") and not os.getenv(\n",
        "    \"DL_ANACONDA_HOME\"\n",
        "):\n",
        "    if \"google.colab\" in sys.modules:\n",
        "        from google.colab import auth as google_auth\n",
        "\n",
        "        google_auth.authenticate_user()\n",
        "\n",
        "    # If you are running this notebook locally, replace the string below with the\n",
        "    # path to your service account key and run this cell to authenticate your GCP\n",
        "    # account.\n",
        "    elif not os.getenv(\"IS_TESTING\"):\n",
        "        %env GOOGLE_APPLICATION_CREDENTIALS ''"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zgPO1eR3CYjk"
      },
      "source": [
        "### Create a Cloud Storage bucket\n",
        "\n",
        "**The following steps are required, regardless of your notebook environment.**\n",
        "\n",
        "Set the name of your Cloud Storage bucket below. It must be unique across all\n",
        "Cloud Storage buckets."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "MzGDU7TWdts_"
      },
      "outputs": [],
      "source": [
        "BUCKET_NAME = \"[your-bucket-name]\"  # @param {type:\"string\"}\n",
        "BUCKET_URI = f\"gs://{BUCKET_NAME}\""
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cf221059d072"
      },
      "outputs": [],
      "source": [
        "if BUCKET_NAME == \"\" or BUCKET_NAME is None or BUCKET_NAME == \"[your-bucket-name]\":\n",
        "    BUCKET_NAME = PROJECT_ID + \"-aip-\" + UUID\n",
        "    BUCKET_URI = f\"gs://{BUCKET_NAME}\""
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-EcIXiGsCePi"
      },
      "source": [
        "**Only if your bucket doesn't already exist**: Run the following cell to create your Cloud Storage bucket."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "NIq7R4HZCfIc"
      },
      "outputs": [],
      "source": [
        "! gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URI"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ucvCsknMCims"
      },
      "source": [
        "Finally, validate access to your Cloud Storage bucket by examining its contents:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vhOb7YnwClBb"
      },
      "outputs": [],
      "source": [
        "! gsutil ls -al $BUCKET_URI"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "set_service_account"
      },
      "source": [
        "### Get your service account \n",
        "\n",
        "If you do not want to use your project's Compute Engine service account, set `SERVICE_ACCOUNT` to another service account ID."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "vqUH5pk0OWzl"
      },
      "outputs": [],
      "source": [
        "SERVICE_ACCOUNT = \"[your-service-account]\"  # @param {type:\"string\"}"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "autoset_service_account"
      },
      "outputs": [],
      "source": [
        "if (\n",
        "    SERVICE_ACCOUNT == \"\"\n",
        "    or SERVICE_ACCOUNT is None\n",
        "    or SERVICE_ACCOUNT == \"[your-service-account]\"\n",
        "):\n",
        "    # Get your service account from gcloud\n",
        "    if not IS_COLAB:\n",
        "        shell_output = !gcloud auth list 2>/dev/null\n",
        "        SERVICE_ACCOUNT = shell_output[2].replace(\"*\", \"\").strip()\n",
        "\n",
        "    else:  # IS_COLAB:\n",
        "        shell_output = ! gcloud projects describe  $PROJECT_ID\n",
        "        project_number = shell_output[-1].split(\":\")[1].strip().replace(\"'\", \"\")\n",
        "        SERVICE_ACCOUNT = f\"{project_number}-compute@developer.gserviceaccount.com\"\n",
        "\n",
        "    print(\"Service Account:\", SERVICE_ACCOUNT)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "set_service_account:pipelines"
      },
      "source": [
        "### Set service account access \n",
        "\n",
        "Run the following commands to grant your service account access to the bucket that you created in the previous step. You only need to run this step once per service account."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "nGfP_UQEOWzm"
      },
      "outputs": [],
      "source": [
        "! gsutil iam ch serviceAccount:{SERVICE_ACCOUNT}:roles/storage.objectCreator {BUCKET_URI}\n",
        "\n",
        "! gsutil iam ch serviceAccount:{SERVICE_ACCOUNT}:roles/storage.objectViewer {BUCKET_URI}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "setup_vars"
      },
      "source": [
        "### Import libraries and Set up variables\n",
        "\n",
        "Next, import libraries and set up some variables used throughout the tutorial.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "pRUOFELefqf1"
      },
      "outputs": [],
      "source": [
        "from pathlib import Path as path\n",
        "from typing import NamedTuple\n",
        "# General\n",
        "from urllib.parse import urlparse\n",
        "\n",
        "import google.cloud.aiplatform as vertex_ai\n",
        "# Check components\n",
        "import tensorflow as tf\n",
        "# Simulate operations\n",
        "from google.cloud import bigquery\n",
        "# ML pipeline\n",
        "from google_cloud_pipeline_components.v1.bigquery import (\n",
        "    BigqueryCreateModelJobOp, BigqueryEvaluateModelJobOp,\n",
        "    BigqueryExplainForecastModelJobOp, BigqueryForecastModelJobOp,\n",
        "    BigqueryMLArimaEvaluateJobOp, BigqueryQueryJobOp)\n",
        "from kfp.v2 import compiler, dsl\n",
        "from kfp.v2.dsl import HTML, Artifact, Condition, Input, Output, component"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "init_aip:mbsdk,all"
      },
      "source": [
        "### Initialize Vertex AI SDK for Python\n",
        "\n",
        "Initialize the Vertex AI SDK for Python for your project and corresponding bucket."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Egd-0fYCOWzm"
      },
      "outputs": [],
      "source": [
        "vertex_ai.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "L9iMCmfFBn6e"
      },
      "source": [
        "### Initialize BigQuery SDK for Python\n",
        "\n",
        "Initialize the BigQuery SDK for Python for your project."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "4Md0UdedBn6f"
      },
      "outputs": [],
      "source": [
        "bq_client = bigquery.Client(project=PROJECT_ID, location=REGION)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FutXJbmAa1NC"
      },
      "source": [
        "### Create local directories\n",
        "\n",
        "Next, you create some local directories that you use in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "3Kms5lCRn7N-"
      },
      "outputs": [],
      "source": [
        "DATA_PATH = \"data\"\n",
        "KFP_COMPONENTS_PATH = \"components\"\n",
        "PIPELINES_PATH = \"pipelines\"\n",
        "\n",
        "! mkdir -m 777 -p {DATA_PATH}\n",
        "! mkdir -m 777 -p {KFP_COMPONENTS_PATH}\n",
        "! mkdir -m 777 -p {PIPELINES_PATH}"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "WRRE4t_bdzj_"
      },
      "source": [
        "### Prepare the training data\n",
        "\n",
        "Next, you will make a copy of the CSV training data into your Cloud Storage bucket and then create a BigQuery dataset table for the training data."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8u7VsRuLaaVC"
      },
      "outputs": [],
      "source": [
        "PUBLIC_DATA_URI = \"gs://cloud-samples-data/vertex-ai/pipeline-deployment/datasets/oracle_retail/orders.csv\"\n",
        "RAW_DATA_URI = f\"{BUCKET_URI}/{DATA_PATH}/raw/orders.csv\"\n",
        "\n",
        "! gsutil -m cp -R $PUBLIC_DATA_URI $RAW_DATA_URI"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1lgNhMlvgbHX"
      },
      "source": [
        "#### Quick peek at the CSV data"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "mdcdUwNwgbHX"
      },
      "outputs": [],
      "source": [
        "! gsutil cat {RAW_DATA_URI} | head"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "AgUlIYgufhnT"
      },
      "outputs": [],
      "source": [
        "LOCATION = REGION.split('-')[0]\n",
        "\n",
        "! bq mk --location={LOCATION} --dataset {PROJECT_ID}:fast_fresh \n",
        "\n",
        "! bq load \\\n",
        "  --location={LOCATION} \\\n",
        "  --source_format=CSV \\\n",
        "  --skip_leading_rows=1\\\n",
        "  fast_fresh.orders_{UUID} \\\n",
        "  {RAW_DATA_URI} \\\n",
        "  time_of_sale:DATETIME,order_id:INTEGER,product_name:STRING,price:NUMERIC,quantity:NUMERIC,payment_method:STRING,store_id:INTEGER,user_id:INTEGER"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ZrgOD30o7HcL"
      },
      "source": [
        "## BigQuery ML Training Formalization\n",
        "\n",
        "In the next cells, you build the components and pipeline to train and evaluate the BQML demand forecasting model."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "A9qoVV5YedRi"
      },
      "source": [
        "### Set up pipeline variables\n",
        "\n",
        "Set up some variables specific for the pipeline.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "xYJhVJf5eguR"
      },
      "outputs": [],
      "source": [
        "# BQML pipeline job configuation\n",
        "PIPELINE_NAME = \"bqml-forecast-pipeline\"\n",
        "PIPELINE_ROOT = urlparse(BUCKET_URI)._replace(path=\"pipeline_root\").geturl()\n",
        "PIPELINE_PACKAGE = str(path(PIPELINES_PATH) / f\"{PIPELINE_NAME}.json\")\n",
        "\n",
        "# BQML pipeline conponent configuration\n",
        "BQ_DATASET = \"fast_fresh\"\n",
        "BQ_ORDERS_TABLE_PREFIX = \"orders\"\n",
        "BQ_TRAINING_TABLE_PREFIX = \"orders_training\"\n",
        "BQ_MODEL_TABLE_PREFIX = \"orders_forecast_arima\"\n",
        "BQ_EVALUATE_TS_TABLE_PREFIX = \"orders_arima_time_series_evaluate\"\n",
        "BQ_EVALUATE_MODEL_TABLE_PREFIX = \"orders_arima_model_evaluate\"\n",
        "BQ_FORECAST_TABLE_PREFIX = \"orders_arima_forecast\"\n",
        "BQ_EXPLAIN_FORECAST_TABLE_PREFIX = \"orders_arima_explain_forecast\"\n",
        "BQ_ORDERS_TABLE = f\"{BQ_ORDERS_TABLE_PREFIX}_{UUID}\"\n",
        "BQ_TRAINING_TABLE = f\"{BQ_TRAINING_TABLE_PREFIX}_{UUID}\"\n",
        "BQ_MODEL_TABLE = f\"{BQ_MODEL_TABLE_PREFIX}_{UUID}\"\n",
        "BQ_EVALUATE_TS_TABLE = f\"{BQ_EVALUATE_TS_TABLE_PREFIX}_{UUID}\"\n",
        "BQ_EVALUATE_MODEL_TABLE = f\"{BQ_EVALUATE_MODEL_TABLE_PREFIX}_{UUID}\"\n",
        "BQ_FORECAST_TABLE = f\"{BQ_FORECAST_TABLE_PREFIX}_{UUID}\"\n",
        "BQ_EXPLAIN_FORECAST_TABLE = f\"{BQ_EXPLAIN_FORECAST_TABLE_PREFIX}_{UUID}\"\n",
        "\n",
        "BQ_TRAIN_CONFIGURATION = {\n",
        "    \"destinationTable\": {\n",
        "        \"projectId\": PROJECT_ID,\n",
        "        \"datasetId\": BQ_DATASET,\n",
        "        \"tableId\": BQ_TRAINING_TABLE,\n",
        "    },\n",
        "    \"writeDisposition\": \"WRITE_TRUNCATE\",\n",
        "}\n",
        "\n",
        "BQ_EVALUATE_TS_CONFIGURATION = {\n",
        "    \"destinationTable\": {\n",
        "        \"projectId\": PROJECT_ID,\n",
        "        \"datasetId\": BQ_DATASET,\n",
        "        \"tableId\": BQ_EVALUATE_TS_TABLE,\n",
        "    },\n",
        "    \"writeDisposition\": \"WRITE_TRUNCATE\",\n",
        "}\n",
        "BQ_EVALUATE_MODEL_CONFIGURATION = {\n",
        "    \"destinationTable\": {\n",
        "        \"projectId\": PROJECT_ID,\n",
        "        \"datasetId\": BQ_DATASET,\n",
        "        \"tableId\": BQ_EVALUATE_MODEL_TABLE,\n",
        "    },\n",
        "    \"writeDisposition\": \"WRITE_TRUNCATE\",\n",
        "}\n",
        "BQ_FORECAST_CONFIGURATION = {\n",
        "    \"destinationTable\": {\n",
        "        \"projectId\": PROJECT_ID,\n",
        "        \"datasetId\": BQ_DATASET,\n",
        "        \"tableId\": BQ_FORECAST_TABLE,\n",
        "    },\n",
        "    \"writeDisposition\": \"WRITE_TRUNCATE\",\n",
        "}\n",
        "BQ_EXPLAIN_FORECAST_CONFIGURATION = {\n",
        "    \"destinationTable\": {\n",
        "        \"projectId\": PROJECT_ID,\n",
        "        \"datasetId\": BQ_DATASET,\n",
        "        \"tableId\": BQ_EXPLAIN_FORECAST_TABLE,\n",
        "    },\n",
        "    \"writeDisposition\": \"WRITE_TRUNCATE\",\n",
        "}\n",
        "PERF_THRESHOLD = 3000"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "zb3047PEgbHY"
      },
      "source": [
        "### Create location for storing compoment definitions\n",
        "\n",
        "Next, you create a Storage Cloud location for storing the YAML component definitions for the custom components you create in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ayXuNmkwKD3T"
      },
      "outputs": [],
      "source": [
        "! mkdir -m 777 -p {KFP_COMPONENTS_PATH}/custom_components"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hZHOP2TzBn6k"
      },
      "source": [
        "### Create a custom component to read model evaluation metrics\n",
        "\n",
        "Build a custom component to consume model evaluation metrics for visualizations in the Vertex AI Pipelines UI using Kubeflow SDK visualization APIs. Indeed, Vertex AI allows you to render that HTML in an output page which is easily accessible from the Google Cloud console."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Iybwx_Z4Bn6k"
      },
      "outputs": [],
      "source": [
        "@component(\n",
        "    base_image=\"python:3.8-slim\",\n",
        "    packages_to_install=[\"jinja2\", \"pandas\", \"matplotlib\"],\n",
        "    output_component_file=f\"{KFP_COMPONENTS_PATH}/custom_components/build_bq_evaluate_metrics.yaml\",\n",
        ")\n",
        "def get_model_evaluation_metrics(\n",
        "    metrics_in: Input[Artifact], metrics_out: Output[HTML]\n",
        ") -> NamedTuple(\"Outputs\", [(\"avg_mean_absolute_error\", float)]):\n",
        "    \"\"\"\n",
        "    Get the average mean absolute error from the metrics\n",
        "    Args:\n",
        "        metrics_in: metrics artifact\n",
        "        metrics_out: metrics artifact\n",
        "    Returns:\n",
        "        avg_mean_absolute_error: average mean absolute error\n",
        "    \"\"\"\n",
        "\n",
        "    import pandas as pd\n",
        "\n",
        "    # Helpers\n",
        "    def prettyfier(styler):\n",
        "        \"\"\"\n",
        "        Helper function to prettify the metrics table.\n",
        "        Args:\n",
        "            styler: Styler object\n",
        "        Returns:\n",
        "            Styler object\n",
        "        \"\"\"\n",
        "        caption = {\n",
        "            \"selector\": \"caption\",\n",
        "            \"props\": [\n",
        "                (\"caption-side\", \"top\"),\n",
        "                (\"font-size\", \"150%\"),\n",
        "                (\"font-weight\", \"bold\"),\n",
        "                (\"font-family\", \"arial\"),\n",
        "            ],\n",
        "        }\n",
        "        headers = {\n",
        "            \"selector\": \"th\",\n",
        "            \"props\": [(\"color\", \"black\"), (\"font-family\", \"arial\")],\n",
        "        }\n",
        "        rows = {\n",
        "            \"selector\": \"td\",\n",
        "            \"props\": [(\"text-align\", \"center\"), (\"font-family\", \"arial\")],\n",
        "        }\n",
        "        styler.set_table_styles([caption, headers, rows])\n",
        "        styler.set_caption(\"Forecasting accuracy report <br><br>\")\n",
        "        styler.hide(axis=\"index\")\n",
        "        styler.format(precision=2)\n",
        "        styler.background_gradient(cmap=\"Blues\")\n",
        "        return styler\n",
        "\n",
        "    def get_column_names(header):\n",
        "        \"\"\"\n",
        "        Helper function to get the column names from the metrics table.\n",
        "        Args:\n",
        "            header: header\n",
        "        Returns:\n",
        "            column_names: column names\n",
        "        \"\"\"\n",
        "        header_clean = header.replace(\"_\", \" \")\n",
        "        header_abbrev = \"\".join([h[0].upper() for h in header_clean.split()])\n",
        "        header_prettied = f\"{header_clean} ({header_abbrev})\"\n",
        "        return header_prettied\n",
        "\n",
        "    # Extract rows and schema from metrics artifact\n",
        "    rows = metrics_in.metadata[\"rows\"]\n",
        "    schema = metrics_in.metadata[\"schema\"]\n",
        "\n",
        "    # Convert into a tabular format\n",
        "    columns = [metrics[\"name\"] for metrics in schema[\"fields\"] if \"name\" in metrics]\n",
        "    records = []\n",
        "    for row in rows:\n",
        "        records.append([dl[\"v\"] for dl in row[\"f\"]])\n",
        "    metrics = (\n",
        "        pd.DataFrame.from_records(records, columns=columns, index=\"product_name\")\n",
        "        .astype(float)\n",
        "        .round(3)\n",
        "    )\n",
        "    metrics = metrics.reset_index()\n",
        "\n",
        "    # Create the HTML artifact for the metrics\n",
        "    pretty_columns = list(\n",
        "        map(\n",
        "            lambda h: get_column_names(h)\n",
        "            if h != columns[0]\n",
        "            else h.replace(\"_\", \" \").capitalize(),\n",
        "            columns,\n",
        "        )\n",
        "    )\n",
        "    pretty_metrics = metrics.copy()\n",
        "    pretty_metrics.columns = pretty_columns\n",
        "    html_metrics = pretty_metrics.style.pipe(prettyfier).to_html()\n",
        "    with open(metrics_out.path, \"w\") as f:\n",
        "        f.write(html_metrics)\n",
        "\n",
        "    # Create metrics dictionary for the model\n",
        "    avg_mean_absolute_error = round(float(metrics.mean_absolute_error.mean()), 0)\n",
        "    component_outputs = NamedTuple(\"Outputs\", [(\"avg_mean_absolute_error\", float)])\n",
        "\n",
        "    return component_outputs(avg_mean_absolute_error)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "pcSL1FHk69KT"
      },
      "source": [
        "### Build the BigQuery ML training pipeline\n",
        "\n",
        "Define your workflow using Kubeflow Pipelines DSL package. \n",
        "\n",
        "Below you have the steps of the pipeline workflow:\n",
        "\n",
        "1. Get BigQuery training data\n",
        "2. Train a BigQuery Arima Plus model\n",
        "3. Evaluate the BigQuery Arima Plus model\n",
        "4. Plot the evaluations\n",
        "5. Check the model performance\n",
        "6. Generate the ARIMA Plus forecasts\n",
        "7. Generate the ARIMA PLUS forecast explainations"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "AlFXqsPIAk0l"
      },
      "outputs": [],
      "source": [
        "@dsl.pipeline(\n",
        "    name=PIPELINE_NAME,\n",
        "    description=\"A batch pipeline to train ARIMA PLUS using BQML\",\n",
        ")\n",
        "def pipeline(\n",
        "    bq_dataset: str = BQ_DATASET,\n",
        "    bq_orders_table: str = BQ_ORDERS_TABLE,\n",
        "    bq_training_table: str = BQ_TRAINING_TABLE,\n",
        "    bq_train_configuration: dict = BQ_TRAIN_CONFIGURATION,\n",
        "    bq_model_table: str = BQ_MODEL_TABLE,\n",
        "    bq_evaluate_time_series_configuration: dict = BQ_EVALUATE_TS_CONFIGURATION,\n",
        "    bq_evaluate_model_configuration: dict = BQ_EVALUATE_MODEL_CONFIGURATION,\n",
        "    performance_threshold: float = PERF_THRESHOLD,\n",
        "    bq_forecast_configuration: dict = BQ_FORECAST_CONFIGURATION,\n",
        "    bq_explain_forecast_configuration: dict = BQ_EXPLAIN_FORECAST_CONFIGURATION,\n",
        "    project: str = PROJECT_ID,\n",
        "    location: str = LOCATION,\n",
        "):\n",
        "\n",
        "    # Create the training dataset\n",
        "    create_training_dataset_op = BigqueryQueryJobOp(\n",
        "        query=f\"\"\"\n",
        "        -- create the training table\n",
        "        WITH \n",
        "        -- get 90% percentile for time series splitting\n",
        "        get_split AS (\n",
        "          SELECT APPROX_QUANTILES(DATETIME_TRUNC(time_of_sale, HOUR), 100)[OFFSET(90)] as split\n",
        "          FROM `{project}.{bq_dataset}.{bq_orders_table}`\n",
        "        ),\n",
        "        -- get train table\n",
        "        get_train AS (\n",
        "          SELECT\n",
        "            DATETIME_TRUNC(time_of_sale, HOUR) as hourly_timestamp,\n",
        "            product_name,\n",
        "            SUM(quantity) AS total_sold,\n",
        "            FROM `{project}.{bq_dataset}.{bq_orders_table}`\n",
        "        GROUP BY hourly_timestamp, product_name\n",
        "        )\n",
        "        SELECT\n",
        "          *,\n",
        "          CASE WHEN hourly_timestamp < (SELECT split FROM get_split) THEN 'TRAIN' ELSE 'TEST' END AS split\n",
        "        FROM get_train\n",
        "        ORDER BY hourly_timestamp\n",
        "        \"\"\",\n",
        "        job_configuration_query=bq_train_configuration,\n",
        "        project=project,\n",
        "        location=location,\n",
        "    ).set_display_name(\"get train data\")\n",
        "\n",
        "    # Run an ARIMA PLUS experiment\n",
        "    bq_arima_model_exp_op = (\n",
        "        BigqueryCreateModelJobOp(\n",
        "            query=f\"\"\"\n",
        "        -- create model table\n",
        "        CREATE OR REPLACE MODEL `{project}.{bq_dataset}.{bq_model_table}`\n",
        "        OPTIONS(\n",
        "        MODEL_TYPE = \\'ARIMA_PLUS\\',\n",
        "        TIME_SERIES_TIMESTAMP_COL = \\'hourly_timestamp\\',\n",
        "        TIME_SERIES_DATA_COL = \\'total_sold\\',\n",
        "        TIME_SERIES_ID_COL = [\\'product_name\\']\n",
        "        ) AS\n",
        "        SELECT\n",
        "          hourly_timestamp,\n",
        "          product_name,\n",
        "          total_sold\n",
        "        FROM `{project}.{bq_dataset}.{bq_training_table}`\n",
        "        WHERE split='TRAIN';\n",
        "        \"\"\",\n",
        "            project=project,\n",
        "            location=location,\n",
        "        )\n",
        "        .set_display_name(\"run arima+ model experiment\")\n",
        "        .after(create_training_dataset_op)\n",
        "    )\n",
        "\n",
        "    # Evaluate ARIMA PLUS time series\n",
        "    _ = (\n",
        "        BigqueryMLArimaEvaluateJobOp(\n",
        "            project=project,\n",
        "            location=location,\n",
        "            model=bq_arima_model_exp_op.outputs[\"model\"],\n",
        "            show_all_candidate_models=False,\n",
        "            job_configuration_query=bq_evaluate_time_series_configuration,\n",
        "        )\n",
        "        .set_display_name(\"evaluate arima plus time series\")\n",
        "        .after(bq_arima_model_exp_op)\n",
        "    )\n",
        "\n",
        "    # Evaluate ARIMA Plus model\n",
        "    bq_arima_evaluate_model_op = (\n",
        "        BigqueryEvaluateModelJobOp(\n",
        "            project=project,\n",
        "            location=location,\n",
        "            model=bq_arima_model_exp_op.outputs[\"model\"],\n",
        "            query_statement=f\"\"\"SELECT * FROM `{project}.{bq_dataset}.{bq_training_table}` WHERE split='TEST'\"\"\",\n",
        "            job_configuration_query=bq_evaluate_model_configuration,\n",
        "        )\n",
        "        .set_display_name(\"evaluate arima plus model\")\n",
        "        .after(bq_arima_model_exp_op)\n",
        "    )\n",
        "\n",
        "    # Plot model metrics\n",
        "    get_evaluation_model_metrics_op = (\n",
        "        get_model_evaluation_metrics(\n",
        "            bq_arima_evaluate_model_op.outputs[\"evaluation_metrics\"]\n",
        "        )\n",
        "        .after(bq_arima_evaluate_model_op)\n",
        "        .set_display_name(\"plot evaluation metrics\")\n",
        "    )\n",
        "\n",
        "    # Check the model performance. If ARIMA_PLUS average MAE metric is below to a minimal threshold\n",
        "    with Condition(\n",
        "        get_evaluation_model_metrics_op.outputs[\"avg_mean_absolute_error\"]\n",
        "        < PERF_THRESHOLD,\n",
        "        name=\"avg. mae good\",\n",
        "    ):\n",
        "        # Train the ARIMA PLUS model\n",
        "        bq_arima_model_op = (\n",
        "            BigqueryCreateModelJobOp(\n",
        "                query=f\"\"\"\n",
        "        -- create model table\n",
        "        CREATE OR REPLACE MODEL `{project}.{bq_dataset}.{bq_model_table}`\n",
        "        OPTIONS(\n",
        "        MODEL_TYPE = \\'ARIMA_PLUS\\',\n",
        "        TIME_SERIES_TIMESTAMP_COL = \\'hourly_timestamp\\',\n",
        "        TIME_SERIES_DATA_COL = \\'total_sold\\',\n",
        "        TIME_SERIES_ID_COL = [\\'product_name\\'],\n",
        "        MODEL_REGISTRY = \\'vertex_ai\\',\n",
        "        VERTEX_AI_MODEL_ID = \\'order_demand_forecasting\\',\n",
        "        VERTEX_AI_MODEL_VERSION_ALIASES = [\\'staging\\']\n",
        "        ) AS\n",
        "        SELECT\n",
        "          DATETIME_TRUNC(time_of_sale, HOUR) as hourly_timestamp,\n",
        "          product_name,\n",
        "          SUM(quantity) AS total_sold,\n",
        "          FROM `{project}.{bq_dataset}.{bq_orders_table}`\n",
        "        GROUP BY hourly_timestamp, product_name;\n",
        "        \"\"\",\n",
        "                project=project,\n",
        "                location=location,\n",
        "            )\n",
        "            .set_display_name(\"train arima+ model\")\n",
        "            .after(get_evaluation_model_metrics_op)\n",
        "        )\n",
        "\n",
        "        # Generate the ARIMA PLUS forecasts\n",
        "        bq_arima_forecast_op = (\n",
        "            BigqueryForecastModelJobOp(\n",
        "                project=project,\n",
        "                location=location,\n",
        "                model=bq_arima_model_op.outputs[\"model\"],\n",
        "                horizon=1,  # 1 hour\n",
        "                confidence_level=0.9,\n",
        "                job_configuration_query=bq_forecast_configuration,\n",
        "            )\n",
        "            .set_display_name(\"generate hourly forecasts\")\n",
        "            .after(get_evaluation_model_metrics_op)\n",
        "        )\n",
        "\n",
        "        # Generate the ARIMA PLUS forecast explainations\n",
        "        _ = (\n",
        "            BigqueryExplainForecastModelJobOp(\n",
        "                project=project,\n",
        "                location=location,\n",
        "                model=bq_arima_model_op.outputs[\"model\"],\n",
        "                horizon=1,  # 1 hour\n",
        "                confidence_level=0.9,\n",
        "                job_configuration_query=bq_explain_forecast_configuration,\n",
        "            )\n",
        "            .set_display_name(\"explain hourly forecasts\")\n",
        "            .after(bq_arima_forecast_op)\n",
        "        )"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nghLONQX7JNg"
      },
      "source": [
        "### Compile the pipeline into a JSON file\n",
        "\n",
        "Next, you compile the pipeline, which produces a JSON specification for your pipeline."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8l6IR7OoADJV"
      },
      "outputs": [],
      "source": [
        "compiler.Compiler().compile(pipeline_func=pipeline, package_path=PIPELINE_PACKAGE)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Gtlwu0Xo1WcT"
      },
      "source": [
        "### Execute your pipeline\n",
        "\n",
        "Next, we execute the pipeline. It takes the following parameters which we set as default:\n",
        "\n",
        "- `bq_dataset`: The BigQuery dataset to train on.\n",
        "- `bq_orders_table` : The BigQuery table of raw data.\n",
        "- `bq_training_table` : The BigQuery table of training data after preprocessing.\n",
        "- `bq_train_configuration` : The job configuration of training component\n",
        "- `bq_model_table`: The BigQuery table of trained model.\n",
        "- `bq_evaluate_time_series_configuration` : The job configuration of the evaluation of ARIMA time series\n",
        "- `bq_evaluate_model_configuration` : The job configuration of the evaluation of ARIMA model\n",
        "- `performance_threshold` : The value of average MAE threshold \n",
        "- `bq_forecast_configuration` : The job configuration of forecast component\n",
        "- `bq_explain_forecast_configuration` : The job configuration of the evaluation of forecast component\n",
        "- `project`: The project ID\n",
        "- `location`: The location\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6rJ0NRuWwfnp"
      },
      "outputs": [],
      "source": [
        "bqml_pipeline = vertex_ai.PipelineJob(\n",
        "    display_name=f\"{PIPELINE_NAME}-job\",\n",
        "    template_path=PIPELINE_PACKAGE,\n",
        "    pipeline_root=PIPELINE_ROOT,\n",
        "    enable_caching=False,\n",
        ")\n",
        "\n",
        "bqml_pipeline.run()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dKgPkm4cgbHd"
      },
      "source": [
        "### View BigQuery ML training pipeline results\n",
        "\n",
        "Finally, you view the artifact outputs of each task in the pipeline."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "apGt59bCgbHd"
      },
      "outputs": [],
      "source": [
        "PROJECT_NUMBER = bqml_pipeline.gca_resource.name.split(\"/\")[1]\n",
        "print(\"PROJECT NUMBER: \", PROJECT_NUMBER)\n",
        "print(\"\\n\\n\")\n",
        "\n",
        "\n",
        "def print_pipeline_output(job, output_task_name):\n",
        "    JOB_ID = job.name\n",
        "    print(JOB_ID)\n",
        "    for _ in range(len(job.gca_resource.job_detail.task_details)):\n",
        "        TASK_ID = job.gca_resource.job_detail.task_details[_].task_id\n",
        "        EXECUTE_OUTPUT = (\n",
        "            PIPELINE_ROOT\n",
        "            + \"/\"\n",
        "            + PROJECT_NUMBER\n",
        "            + \"/\"\n",
        "            + JOB_ID\n",
        "            + \"/\"\n",
        "            + output_task_name\n",
        "            + \"_\"\n",
        "            + str(TASK_ID)\n",
        "            + \"/executor_output.json\"\n",
        "        )\n",
        "        GCP_RESOURCES = (\n",
        "            PIPELINE_ROOT\n",
        "            + \"/\"\n",
        "            + PROJECT_NUMBER\n",
        "            + \"/\"\n",
        "            + JOB_ID\n",
        "            + \"/\"\n",
        "            + output_task_name\n",
        "            + \"_\"\n",
        "            + str(TASK_ID)\n",
        "            + \"/gcp_resources\"\n",
        "        )\n",
        "        EVAL_METRICS = (\n",
        "            PIPELINE_ROOT\n",
        "            + \"/\"\n",
        "            + PROJECT_NUMBER\n",
        "            + \"/\"\n",
        "            + JOB_ID\n",
        "            + \"/\"\n",
        "            + output_task_name\n",
        "            + \"_\"\n",
        "            + str(TASK_ID)\n",
        "            + \"/evaluation_metrics\"\n",
        "        )\n",
        "        if tf.io.gfile.exists(EXECUTE_OUTPUT):\n",
        "            ! gsutil cat $EXECUTE_OUTPUT\n",
        "            return EXECUTE_OUTPUT\n",
        "        elif tf.io.gfile.exists(GCP_RESOURCES):\n",
        "            ! gsutil cat $GCP_RESOURCES\n",
        "            return GCP_RESOURCES\n",
        "        elif tf.io.gfile.exists(EVAL_METRICS):\n",
        "            ! gsutil cat $EVAL_METRICS\n",
        "            return EVAL_METRICS\n",
        "\n",
        "    return None\n",
        "\n",
        "\n",
        "print(\"bigquery-create-model-job\")\n",
        "artifacts = print_pipeline_output(bqml_pipeline, \"bigquery-create-model-job\")\n",
        "print(\"\\n\\n\")\n",
        "print(\"bigquery-ml-arima-evaluate-job\")\n",
        "artifacts = print_pipeline_output(bqml_pipeline, \"bigquery-ml-arima-evaluate-job\")\n",
        "print(\"\\n\\n\")\n",
        "print(\"bigquery-evaluate-model-job\")\n",
        "artifacts = print_pipeline_output(bqml_pipeline, \"bigquery-evaluate-model-job\")\n",
        "print(\"\\n\\n\")\n",
        "print(\"bigquery-forecast-model-job\")\n",
        "artifacts = print_pipeline_output(bqml_pipeline, \"bigquery-forecast-model-job\")\n",
        "print(\"\\n\\n\")\n",
        "print(\"bigquery-explain-forecast-model-job\")\n",
        "artifacts = print_pipeline_output(bqml_pipeline, \"bigquery-explain-forecast-model-job\")\n",
        "print(\"\\n\\n\")"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "TpV-iwP9qw9c"
      },
      "source": [
        "## Cleaning up\n",
        "\n",
        "To clean up all Google Cloud resources used in this project, you can [delete the Google Cloud\n",
        "project](https://cloud.google.com/resource-manager/docs/creating-managing-projects#shutting_down_projects) you used for the tutorial.\n",
        "\n",
        "Otherwise, you can delete the individual resources you created in this tutorial."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "P0Ks1UZpoRXS"
      },
      "outputs": [],
      "source": [
        "# delete pipeline\n",
        "vertex_ai_pipeline_jobs = vertex_ai.PipelineJob.list(\n",
        "    filter=f'pipeline_name=\"{PIPELINE_NAME}\"'\n",
        ")\n",
        "for pipeline_job in vertex_ai_pipeline_jobs:\n",
        "    pipeline_job.delete()\n",
        "\n",
        "# delete model\n",
        "DELETE_MODEL_SQL = f\"DROP MODEL {BQ_DATASET}.{BQ_MODEL_TABLE}\"\n",
        "try:\n",
        "    delete_model_query_job = bq_client.query(DELETE_MODEL_SQL)\n",
        "    delete_model_query_result = delete_model_query_job.result()\n",
        "except Exception as e:\n",
        "    print(e)\n",
        "\n",
        "# delete dataset\n",
        "try:\n",
        "    delete_detaset_query_result = bq_client.delete_dataset(\n",
        "        BQ_DATASET, delete_contents=True, not_found_ok=True\n",
        "    )\n",
        "    print(delete_detaset_query_result)\n",
        "except Exception as e:\n",
        "    print(e)\n",
        "\n",
        "# delete bucket\n",
        "delete_bucket = True\n",
        "if os.getenv(\"IS_TESTING\") or delete_bucket:\n",
        "    ! gsutil -m rm -r $BUCKET_URI\n",
        "\n",
        "\n",
        "# Remove local resorces\n",
        "! rm -rf {KFP_COMPONENTS_PATH}\n",
        "! rm -rf {PIPELINES_PATH}\n",
        "! rm -rf {DATA_PATH}"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "google_cloud_pipeline_components_bqml_pipeline_demand_forecasting.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}