00_pipelines_intro_kfp.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "copyright"
   },
   "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": "title:generic"
   },
   "source": [
    "# Vertex AI Pipelines: Pipelines introduction for KFP\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "overview:pipelines,intro"
   },
   "source": [
    "## Overview\n",
    "\n",
    "This notebook provides an introduction to using [Vertex AI Pipelines](https://cloud.google.com/vertex-ai/docs/pipelines) with [the Kubeflow Pipelines (KFP) SDK](https://www.kubeflow.org/docs/components/pipelines/).\n",
    "\n",
    "Learn more about [Vertex AI Pipelines](https://cloud.google.com/vertex-ai/docs/pipelines/introduction)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "objective:pipelines,intro"
   },
   "source": [
    "### Objective\n",
    "\n",
    "In this tutorial, you learn how to use the KFP SDK to build pipelines that generate evaluation metrics.\n",
    "\n",
    "This tutorial uses the following Google Cloud ML services:\n",
    "\n",
    "- `Vertex AI Pipelines`\n",
    "\n",
    "The steps performed include:\n",
    "\n",
    "- Define and compile a `Vertex AI` pipeline.\n",
    "- Specify which service account to use for a pipeline run."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "costs:functions,scheduler"
   },
   "source": [
    "### Costs\n",
    "\n",
    "This tutorial uses billable components of Google Cloud:\n",
    "\n",
    "* Vertex AI\n",
    "* Cloud Storage\n",
    "\n",
    "Learn about [Vertex AI pricing](https://cloud.google.com/vertex-ai/pricing),\n",
    "[Cloud Storage pricing](https://cloud.google.com/storage/pricing),\n",
    "and use the [Pricing\n",
    "Calculator](https://cloud.google.com/products/calculator/)\n",
    "to generate a cost estimate based on your projected usage."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "773901ca47fd"
   },
   "source": [
    "### Install additional packages\n",
    "\n",
    "Install the following packages required to execute this notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "id": "b7c7ce6bbf03",
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[33m  WARNING: The scripts dsl-compile, dsl-compile-v2 and kfp are installed in '/home/jupyter/.local/bin' which is not on PATH.\n",
      "  Consider adding this directory to PATH or, if you prefer to suppress this warning, use --no-warn-script-location.\u001b[0m\u001b[33m\n",
      "\u001b[0m"
     ]
    }
   ],
   "source": [
    "!pip install --upgrade --no-warn-conflicts --user -q \\\n",
    "    google-cloud-pipeline-components \\\n",
    "    kfp==1.8.18 \\\n",
    "    tensorflow \\\n",
    "    google-cloud-aiplatform==1.21.0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "d07214a67580"
   },
   "source": [
    "After you install the additional packages, you need to restart the notebook kernel so it can find the packages."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "id": "18c113700b6f"
   },
   "outputs": [],
   "source": [
    "# Automatically restart kernel after installs\n",
    "import os\n",
    "\n",
    "if not os.getenv(\"IS_TESTING\"):\n",
    "\n",
    "    import IPython\n",
    "\n",
    "    app = IPython.Application.instance()\n",
    "    app.kernel.do_shutdown(True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "a47846030fef"
   },
   "source": [
    "## Before you begin"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "project_id"
   },
   "source": [
    "#### Set your project ID"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "id": "3c8049930470",
    "tags": []
   },
   "outputs": [],
   "source": [
    "GCP_PROJECTS = !gcloud config get-value project\n",
    "PROJECT_ID = GCP_PROJECTS[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "a54f9d7c1876"
   },
   "source": [
    "#### Region\n",
    "\n",
    "You can also change the `REGION` variable used by Vertex AI. Learn more about [Vertex AI regions](https://cloud.google.com/vertex-ai/docs/general/locations)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "id": "3aaadaaf9b30"
   },
   "outputs": [],
   "source": [
    "REGION = \"us-central1\"  # @param {type: \"string\"}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "bucket:custom"
   },
   "source": [
    "### Create a Cloud Storage bucket\n",
    "\n",
    "Create a storage bucket to store intermediate artifacts such as datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "id": "bucket"
   },
   "outputs": [],
   "source": [
    "BUCKET_URI = f\"gs://{PROJECT_ID}-mlops\"  # @param {type:\"string\"}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "create_bucket"
   },
   "source": [
    "Run the following cell to create your Cloud Storage bucket."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "Oz8J0vmSlugt"
   },
   "outputs": [],
   "source": [
    "! gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URI"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "setup_vars"
   },
   "source": [
    "### Set up variables\n",
    "\n",
    "Next, set up some variables used throughout the tutorial.\n",
    "### Import libraries and define constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "id": "import_aip:mbsdk"
   },
   "outputs": [],
   "source": [
    "from typing import NamedTuple\n",
    "\n",
    "import google.cloud.aiplatform as aip\n",
    "from kfp import dsl\n",
    "from kfp.v2 import compiler\n",
    "from kfp.v2.dsl import component, Input, Output, Artifact"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "pipeline_constants"
   },
   "source": [
    "#### Vertex AI Pipelines constants\n",
    "\n",
    "Setup up the following constants for Vertex AI Pipelines:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "id": "ONHtQDz32Ou5"
   },
   "outputs": [],
   "source": [
    "PIPELINE_ROOT = \"{}/pipeline_root/intro\".format(BUCKET_URI)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "init_aip:mbsdk"
   },
   "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": 12,
   "metadata": {
    "id": "mbpw7oyM2Ou5"
   },
   "outputs": [],
   "source": [
    "aip.init(project=PROJECT_ID, staging_bucket=BUCKET_URI)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "define_component:hello_world"
   },
   "source": [
    "### Define Python function-based pipeline components\n",
    "\n",
    "In this tutorial, you define a simple pipeline that has three steps, where each step is defined as a component.\n",
    "\n",
    "#### Define hello_world component\n",
    "\n",
    "First, define a component based on a very simple Python function. It takes a string input parameter and returns that value as output.\n",
    "\n",
    "Note the use of the `@component` decorator, which compiles the function to a KFP component when evaluated.  For example purposes, this example specifies a base image to use for the component (`python:3.9`), and a component YAML file, `hw.yaml`. The compiled component specification is written to this file.  (The default base image is `python:3.7`, which would of course work just fine too)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "id": "GjJhJUID2Ou6"
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/conda/lib/python3.7/site-packages/ipykernel_launcher.py:1: DeprecationWarning: output_component_file parameter is deprecated and will eventually be removed. Please use `Compiler().compile()` to compile a component instead.\n",
      "  \"\"\"Entry point for launching an IPython kernel.\n",
      "/opt/conda/lib/python3.7/site-packages/ipykernel_launcher.py:2: DeprecationWarning: output_component_file parameter is deprecated and will eventually be removed. Please use `Compiler().compile()` to compile a component instead.\n",
      "  \n"
     ]
    }
   ],
   "source": [
    "@component(output_component_file=\"hw.yaml\", base_image=\"python:3.9\")\n",
    "def hello_world(text: str) -> str:\n",
    "    print(text)\n",
    "    return text"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "SWcIXuxR2Ou6"
   },
   "source": [
    "As you'll see below, compilation of this component creates a [task factory function](https://www.kubeflow.org/docs/components/pipelines/sdk/python-function-components/)—called `hello_world`— that you can use in defining a pipeline step.\n",
    "\n",
    "While not shown here, if you want to share this component definition, or use it in another context, you could also load it from its yaml file like this:\n",
    "`hello_world_op = components.load_component_from_file('./hw.yaml')`.\n",
    "You can also use the `load_component_from_url` method, if your component yaml file is stored online. (For GitHub URLs, load the 'raw' file.)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "define_component:two_outputs",
    "tags": []
   },
   "source": [
    "#### Define three_outputs component\n",
    "\n",
    "The first component below, `three_outputs`, demonstrates installing a package -- in this case the `google-cloud-storage` package. Alternatively, you can specify a base image that includes the necessary installations.\n",
    "\n",
    "Alternatively, you can specify a base image that includes the necessary installations.\n",
    "\n",
    "This component additionally showcases how to produce different inputs / outputs as part of a component.\n",
    "\n",
    "##### Input/Output parameters\n",
    "Input and Ouput parameters are declared when you use a str, int, float, bool, dict or list type annotation. The data passed to parameters typed with dict or list may only container JSON-serializable Python primitives. Union types are not permitted. When producing multiple outputs, a NamedTuple should be used. In this case `text` it's an input parameter.\n",
    "\n",
    "\n",
    "##### Input/Output artifacts\n",
    "Input artifacts are defined when you use an Input/Output[<ArtifactClass>] annotation. In the following cell `file_output` it's an Output artifact.\n",
    "\n",
    "At component runtime, input artifacts are copied to the local filesystem by the executing backend. This abstracts away the need for the component author to know where artifacts are stored in remote storage and allows component authors to only interact with the local filesystem when implementing a component that uses an artifact. All artifacts implement a .path method, which can be used to access the local path where the artifact file has been copied. All the artifacts also expose an .uri method to get the relative GCS uri.\n",
    "    \n",
    "There are also other specialised Artifact types accepted as part of a pipeline, such as Model, Dataset, Metrics, HTML.\n",
    "    \n",
    "The `three_outputs` component returns:\n",
    "- two named outputs, so outputs that can be returned as output of the python function.\n",
    "- one File Artifact output, so a file produced by the component which gets tracked as part of the pipeline lineage. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "id": "U4Yv33su2Ou6"
   },
   "outputs": [],
   "source": [
    "@component(packages_to_install=[\"google-cloud-storage\"])\n",
    "def three_outputs(\n",
    "    text: str,\n",
    "    file_output: Output[Artifact]\n",
    ") -> NamedTuple(\n",
    "    \"Outputs\",\n",
    "    [\n",
    "        (\"output_one\", str),  # Return parameters\n",
    "        (\"output_two\", str),\n",
    "    ],\n",
    "):\n",
    "    # the import is not actually used for this simple example, but the import\n",
    "    # is successful, as it was included in the `packages_to_install` list.\n",
    "    from google.cloud import storage\n",
    "    \n",
    "    o1 = f\"output one from text: {text}\"\n",
    "    o2 = f\"output two from text: {text}\"\n",
    "    print(\"output one: {}; output_two: {}\".format(o1, o2))\n",
    "    \n",
    "    with open(file_output.path, 'w') as f: \n",
    "        f.write(\"third output\")\n",
    "    \n",
    "    return (o1, o2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "define_component:consumer"
   },
   "source": [
    "#### Define the consumer component\n",
    "\n",
    "The third component, `consumer`, takes three string inputs and the Artifact produced earlier (this time as Input) and prints them out."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "id": "bu8XvOj82Ou6"
   },
   "outputs": [],
   "source": [
    "@component\n",
    "def consumer(text1: str, text2: str, text3: str, file: Input[Artifact]):\n",
    "    with open(file.path) as f:\n",
    "        print(f.read())\n",
    "    print(f\"text1: {text1}; text2: {text2}; text3: {text3}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "define_pipeline:intro"
   },
   "source": [
    "### Define a pipeline that uses the components\n",
    "\n",
    "Next, define a pipeline that uses these three components.\n",
    "\n",
    "By evaluating the component definitions above, you've created task factory functions that are used in the pipeline definition to create the pipeline steps.\n",
    "\n",
    "The pipeline takes an input parameter, and passes that parameter as an argument to the first two pipeline steps (`hw_task` and `three_outputs_task`).\n",
    "\n",
    "Then, the third pipeline step (`consumer_task`) consumes the outputs of the first and second steps.  Because the `hello_world` component definition just returns one unnamed output, you refer to it as `hw_task.output`.  The `three_outputs` task returns two named outputs and the Artifact Output, which you access as `three_outputs_task.outputs[\"<output_name>\"]`.\n",
    "\n",
    "*Note:* In the `@dsl.pipeline` decorator, you're defining the `PIPELINE_ROOT` Cloud Storage path to use.  If you had not included that info here, it would be required to specify it when creating the pipeline run, as you'll see below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "id": "CV5dRAeJ2Ou7"
   },
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "    name=\"intro-pipeline-unique\",\n",
    "    description=\"A simple intro pipeline\",\n",
    "    pipeline_root=PIPELINE_ROOT,\n",
    ")\n",
    "def pipeline(text: str = \"hi there\"):\n",
    "    hw_task = hello_world(text=text)\n",
    "    three_outputs_task = three_outputs(text=text)\n",
    "    consumer_task = consumer(\n",
    "        file=three_outputs_task.outputs[\"file_output\"],\n",
    "        text1=hw_task.output,\n",
    "        text2=three_outputs_task.outputs[\"output_one\"],\n",
    "        text3=three_outputs_task.outputs[\"output_two\"],\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "compile_pipeline"
   },
   "source": [
    "## Compile the pipeline\n",
    "\n",
    "Next, compile the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "id": "VP_JJ9Oe2Ou7"
   },
   "outputs": [],
   "source": [
    "from kfp.v2 import compiler  # noqa: F811\n",
    "\n",
    "compiler.Compiler().compile(pipeline_func=pipeline, package_path=\"intro_pipeline.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "run_pipeline:intro"
   },
   "source": [
    "## Run the pipeline\n",
    "\n",
    "Next, run the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "sjxaBix_2Ou7"
   },
   "outputs": [],
   "source": [
    "job = aip.PipelineJob(\n",
    "    display_name=\"intro_pipeline_job\",\n",
    "    template_path=\"intro_pipeline.json\",\n",
    "    pipeline_root=PIPELINE_ROOT,\n",
    ")\n",
    "\n",
    "job.submit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "view_pipeline_run:intro"
   },
   "source": [
    "Click on the generated link to see your run in the Cloud Console.\n",
    "\n",
    "<!-- It should look something like this as it is running:\n",
    "\n",
    "<a href=\"https://storage.googleapis.com/amy-jo/images/mp/automl_tabular_classif.png\" target=\"_blank\"><img src=\"https://storage.googleapis.com/amy-jo/images/mp/automl_tabular_classif.png\" width=\"40%\"/></a> -->\n",
    "\n",
    "In the UI, many of the pipeline DAG nodes will expand or collapse when you click on them. Here is a partially-expanded view of the DAG (click image to see larger version).\n",
    "\n",
    "<a href=\"https://storage.googleapis.com/amy-jo/images/mp/intro_pipeline.png\" target=\"_blank\"><img src=\"https://storage.googleapis.com/amy-jo/images/mp/intro_pipeline.png\" width=\"60%\"/></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Experiment with a different parameter\n",
    "You can now try to provide a different input parameter to the pipeline and see how this changes the print statements"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating PipelineJob\n",
      "PipelineJob created. Resource name: projects/370018035372/locations/us-central1/pipelineJobs/intro-pipeline-unique-20230123150643\n",
      "To use this PipelineJob in another session:\n",
      "pipeline_job = aiplatform.PipelineJob.get('projects/370018035372/locations/us-central1/pipelineJobs/intro-pipeline-unique-20230123150643')\n",
      "View Pipeline Job:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/intro-pipeline-unique-20230123150643?project=370018035372\n"
     ]
    }
   ],
   "source": [
    "job = aip.PipelineJob(\n",
    "    display_name=\"intro_pipeline_job2\",\n",
    "    template_path=\"intro_pipeline.json\",\n",
    "    pipeline_root=PIPELINE_ROOT,\n",
    "#     Your text goes here\n",
    "    parameter_values={\"text\": \"\"}\n",
    ")\n",
    "\n",
    "job.submit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "name": "pipelines_intro_kfp.ipynb",
   "toc_visible": true
  },
  "environment": {
   "kernel": "python3",
   "name": "common-cpu.m102",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/base-cpu:m102"
  },
  "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.13 (default, Oct 26 2022, 14:34:33) \n[Clang 14.0.0 (clang-1400.0.29.102)]"
  },
  "vscode": {
   "interpreter": {
    "hash": "82b4ab82dc5f425ab3c6d72cb2a5f96ebe8c6e2dcd94cf41a2992da868b63006"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}