Train and deploy end-to-end sklearn text classifier (#122)

* added sklearn-example

* nb formatting

* added readme

* added endpoint to notebook

* nb formatting

* added endpoint prediction example

* nb formatting

* fixed request

* minor woring fixes in docstrings

* added codeowner and included PR feedback

Co-authored-by: Maximilian Engelhardt <maximilian.engelhardt@ing.com>

community-content/CODEOWNERS

@@ -1,3 +1,4 @@
 *  @vertex-ai-samples-contributors  @GoogleCloudPlatform/cloudml-samples-owners
 /tf_agents_bandits_movie_recommendation_with_kfp_and_vertex_sdk	@yinghsienwu
 /pytorch_text_classification_using_vertex_sdk_and_gcloud		@RajeshThallam @ultrons
+/sklearn_text_classification_from_script_using_vertex_sdk       @maxhardt

community-content/sklearn_text_classification_from_script_using_vertex_sdk/README.md

@@ -0,0 +1,28 @@
+# Train and deploy a scikit-learn model with Vertex AI
+
+This repository shows how to train and deploy a text classifier using scikit-learn and Vertex AI.
+
+The main used Vertex AI features are:
+- Vertex AI Custom Training
+- Vertex AI Model
+- Vertex AI Endpoint
+
+Further used GCP services are:
+- Google Cloud Logging
+- Google Cloud Storage
+
+## Repository
+
+    ├── README.md
+    ├── create_job.ipynb    # <-- creates the training job and deploys the model
+    ├── requirements.txt    # <-- requirements for deploying the job
+    └── task.py             # <-- contains the training application
+
+## Training job overview
+
+The training job performs the following steps:
+
+1. Downloads the `NewsAggregator` dataset from the UCI Machine Learning Repository
+2. Trains and evaluates a classifier using scikit-learn
+3. Exports model and evaluation artifacts to GCS
+4. Deploys the model as a `Vertex AI Endpoint`

community-content/sklearn_text_classification_from_script_using_vertex_sdk/create_job.ipynb

@@ -0,0 +1,290 @@
+{
+  "cells": [
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "72b875d67303"
+      },
+      "source": [
+        "# Create and run a custom Vertex AI Training Job from a local script"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "398b976f501e"
+      },
+      "source": [
+        "## Install Vertex AI Python Client"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "2162adc1d8fe"
+      },
+      "outputs": [],
+      "source": [
+        "!pip install -r requirements.txt --upgrade"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "d4bf4ab70e65"
+      },
+      "source": [
+        "## GCP authentication"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "41084de2e96a"
+      },
+      "outputs": [],
+      "source": [
+        "import os\n",
+        "\n",
+        "os.environ[\n",
+        "    \"GOOGLE_APPLICATION_CREDENTIALS\"\n",
+        "] = \"\"  # TODO: path to credentials .json file"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "1ce7efb99a95"
+      },
+      "source": [
+        "## Create the custom Vertex AI Training Job\n",
+        "\n",
+        "1. Define the custom job parameters\n",
+        "2. Submit the job to create a `Vertex AI Model`"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "c04b9efb5eb3"
+      },
+      "outputs": [],
+      "source": [
+        "# Import the Vertex AI SDK (Python Client)\n",
+        "from google.cloud import aiplatform"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "b56f5fdcefd6"
+      },
+      "outputs": [],
+      "source": [
+        "# Project meta data\n",
+        "PROJECT_ID = \"\"  # TODO\n",
+        "REGION = \"\"  # TODO e.g. europe\n",
+        "ZONE = \"\"  # TODO e.g. west4\n",
+        "LOCATION = f\"{REGION}-{ZONE}\""
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "4af8cfb1e1a1"
+      },
+      "outputs": [],
+      "source": [
+        "aiplatform.init(project=PROJECT_ID, location=LOCATION)"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "e23d0161b489"
+      },
+      "outputs": [],
+      "source": [
+        "# Variables for specifying the job\n",
+        "DISPLAY_NAME = (\n",
+        "    \"news-classifier-training\"  # TODO: How the job is displayed on Vertex AI GUI\n",
+        ")\n",
+        "SCRIPT_PATH = \"./task.py\"  # Path to local training script\n",
+        "STAGING_BUCKET = (\n",
+        "    \"\"  # TODO GCS URI where meta data and artifacts are stored for this job\n",
+        ")\n",
+        "MODEL_TRAINING_IMAGE = f\"{REGION}-docker.pkg.dev/vertex-ai/training/scikit-learn-cpu.0-23:latest\"  # Pre-built training image\n",
+        "REQUIREMENTS = [\"wget\"]  # Additional requirements not already part of the base image\n",
+        "# !Required if the Training Pipeline produces a managed Vertex AI Model!\n",
+        "MODEL_SERVING_IMAGE = f\"{REGION}-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.0-23:latest\"  # Pre-built serving image"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "61565ec3e6de"
+      },
+      "outputs": [],
+      "source": [
+        "# Job definition\n",
+        "custom_training_job = aiplatform.CustomTrainingJob(\n",
+        "    project=PROJECT_ID,\n",
+        "    location=LOCATION,\n",
+        "    display_name=DISPLAY_NAME,\n",
+        "    script_path=SCRIPT_PATH,\n",
+        "    staging_bucket=STAGING_BUCKET,\n",
+        "    container_uri=MODEL_TRAINING_IMAGE,\n",
+        "    requirements=REQUIREMENTS,\n",
+        "    model_serving_container_image_uri=MODEL_SERVING_IMAGE,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "8c8d4bd78688"
+      },
+      "outputs": [],
+      "source": [
+        "# Variables for running the job\n",
+        "MACHINE_TYPE = \"n1-standard-4\"  # Standard VM with 4 CPUs\n",
+        "# !Required if the Training Pipeline produces a managed Vertex AI Model!\n",
+        "MODEL_DISPLAY_NAME = (\n",
+        "    \"news-classifier-model\"  # TODO: Name for the resulting managed Vertex AI Model.\n",
+        ")\n",
+        "# Note that a single job may produce multiple models (e.g. one per run).\n",
+        "# The url to download the training data from.\n",
+        "DATASET_URL = \"https://archive.ics.uci.edu/ml/machine-learning-databases/00359/NewsAggregatorDataset.zip\""
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "13d31a7d0bb5"
+      },
+      "outputs": [],
+      "source": [
+        "# Run the job\n",
+        "model = custom_training_job.run(\n",
+        "    machine_type=MACHINE_TYPE,\n",
+        "    model_display_name=MODEL_DISPLAY_NAME,\n",
+        "    args=[f\"--dataset_url={DATASET_URL}\", f\"--project_id={PROJECT_ID}\"],\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "8026ac119722"
+      },
+      "outputs": [],
+      "source": [
+        "MODEL_RESOURCE_NAME = model.resource_name"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "2d985fca37f3"
+      },
+      "source": [
+        "## Deploy model to Vertex AI Endpoint\n",
+        "\n",
+        "1. Retrieve the registered `Vertex AI Model`\n",
+        "2. Deploy the model to a new `Vertex AI Endpoint`\n",
+        "3. Get some test predictions from the endpoint"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "58c08631f94c"
+      },
+      "outputs": [],
+      "source": [
+        "ENDPOINT_DISPLAY_NAME = \"news-classifier-endpoint\"  # TODO\n",
+        "MACHINE_TYPE_SERVING = \"n1-standard-2\""
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "b54867240880"
+      },
+      "outputs": [],
+      "source": [
+        "endpoint = aiplatform.Endpoint.create(\n",
+        "    display_name=ENDPOINT_DISPLAY_NAME,\n",
+        "    location=LOCATION,\n",
+        "    project=PROJECT_ID,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "a76e7c721b88"
+      },
+      "outputs": [],
+      "source": [
+        "model = aiplatform.Model(model_name=MODEL_RESOURCE_NAME)"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "d06714302f81"
+      },
+      "outputs": [],
+      "source": [
+        "model.deploy(\n",
+        "    endpoint=endpoint,\n",
+        "    deployed_model_display_name=MODEL_DISPLAY_NAME,\n",
+        "    machine_type=MACHINE_TYPE,\n",
+        "    traffic_percentage=100,\n",
+        "    min_replica_count=1,\n",
+        "    max_replica_count=1,\n",
+        "    accelerator_type=None,\n",
+        "    accelerator_count=None,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "5b7eca31d80e"
+      },
+      "outputs": [],
+      "source": [
+        "endpoint.predict(instances={\"instances\": [\"A news headline to be classified\"]})"
+      ]
+    }
+  ],
+  "metadata": {
+    "colab": {
+      "name": "create_job.ipynb",
+      "toc_visible": true
+    },
+    "kernelspec": {
+      "display_name": "Python 3",
+      "name": "python3"
+    }
+  },
+  "nbformat": 4,
+  "nbformat_minor": 0
+}

community-content/sklearn_text_classification_from_script_using_vertex_sdk/requirements.txt

@@ -0,0 +1,2 @@
+google-cloud-aiplatform
+ipykernel

community-content/sklearn_text_classification_from_script_using_vertex_sdk/task.py

@@ -0,0 +1,167 @@
+import argparse
+import logging
+import os
+import pickle
+import zipfile
+from typing import List, Tuple
+
+import pandas as pd
+import wget
+from google.cloud import storage
+from google.cloud.logging import Client as LogClient
+from sklearn.feature_extraction.text import CountVectorizer, TfidfTransformer
+from sklearn.model_selection import train_test_split
+from sklearn.naive_bayes import MultinomialNB
+from sklearn.pipeline import Pipeline
+
+
+def download_dataset_from_url(url: str) -> pd.DataFrame:
+    """Downloads and unzips the dataset from `url` and reads it with pandas.
+
+    Args:
+        url (str, optional): URL to the dataset.
+    """
+
+    zip_filepath = wget.download(url, out=".")
+
+    with zipfile.ZipFile(zip_filepath, "r") as zf:
+        zf.extract(path=".", member="newsCorpora.csv")
+
+    COLUMN_NAMES = ["id", "title", "url", "publisher",
+                    "category", "story", "hostname", "timestamp"]
+
+    return pd.read_csv(
+        "newsCorpora.csv", delimiter="\t", names=COLUMN_NAMES, index_col=0
+    )
+
+
+def get_train_test_data(dataframe: pd.DataFrame, test_size: float = 0.2
+) -> Tuple[List, List, List, List]:
+    """Splits the news dataset into train and test features and labels.
+
+    Args:
+        news (pd.DataFrame): The dataset as pandas DataFrame.
+        test_size (float): The size in percent of the test data.
+
+    Returns:
+        Tuple[List, List, List, List]: Tuple with train and test data
+    """
+
+    train, test = train_test_split(dataframe, test_size=test_size)
+
+    x_train, y_train = train["title"].values, train["category"].values
+    x_test, y_test = test["title"].values, test["category"].values
+
+    return x_train, y_train, x_test, y_test
+
+
+def export_model_to_gcs(fitted_pipeline: Pipeline, gcs_uri: str) -> str:
+    """Exports trained pipeline to GCS
+
+    Parameters:
+            fitted_pipeline (sklearn.pipelines.Pipeline): the Pipeline object
+                with data already fitted (trained pipeline object).
+            gcs_uri (str): GCS path to store the trained pipeline
+                i.e gs://example_bucket/training-job.
+    Returns:
+            export_path (str): Model GCS location
+    """
+
+    artifact_filename = 'model.pkl'
+
+    # Save model artifact to local filesystem (doesn't persist)
+    local_path = artifact_filename
+    with open(local_path, 'wb') as model_file:
+        pickle.dump(fitted_pipeline, model_file)
+
+    # Upload model artifact to Cloud Storage
+    storage_path = os.path.join(gcs_uri, artifact_filename)
+    blob = storage.blob.Blob.from_string(storage_path, client=storage.Client())
+    blob.upload_from_filename(local_path)
+
+
+def export_evaluation_report_to_gcs(report: str, gcs_uri: str) -> None:
+    """
+    Exports training job report to GCS
+
+    Parameters:
+        report (str): Full report in text to sent to GCS
+        gcs_uri (str): GCS path to store the report
+            i.e gs://example_bucket/training-job
+    """
+
+    artifact_filename = 'report.txt'
+
+    # Upload model artifact to Cloud Storage
+    storage_path = os.path.join(gcs_uri, artifact_filename)
+    blob = storage.blob.Blob.from_string(storage_path, client=storage.Client())
+    blob.upload_from_string(report)
+
+
+def train_and_score(X_train: List, y_train: List, X_test: List, y_test: List
+) -> Tuple[Pipeline, float]:
+    """Trains and cross-validates a text classifier pipeline.
+
+    Args:
+        X_train (List): Train features as list of strings.
+        y_train (List): Train labels as list of strings.
+        X_test (List): Test labels as list of strings.
+        y_test (List): Test labels as list of strings.
+
+    Returns:
+        Tuple[Pipeline, float]: Fitted pipeline and mean accuracy.
+    """
+
+    pipeline = Pipeline([
+                ("vectorizer", CountVectorizer()),
+                ("tfidf", TfidfTransformer()),
+                ("naivebayes", MultinomialNB()),
+        ])
+
+    pipeline.fit(X_train, y_train)
+    score = pipeline.score(X_test, y_test)
+
+    return pipeline, score
+
+
+# Define all the command line arguments your model can accept for training
+if __name__ == "__main__":
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument(
+        "--dataset_url",
+        help="Download url for the training data.",
+        type=str
+    )
+
+    parser.add_argument(
+        "--project_id",
+        help="GCP project id for cloud logging.",
+        type=str
+    )
+
+    args = parser.parse_args()
+    arguments = args.__dict__
+
+    # set up the GCP logger
+    client = LogClient(project=arguments["project_id"])
+    client.setup_logging(log_level=logging.INFO)
+    logging.info("Starting custom training job.")
+
+    # download the data from url
+    logging.info("Downloading training data from: {}".format(arguments["dataset_url"]))
+    dataframe = download_dataset_from_url(arguments["dataset_url"])
+    train_test_data = get_train_test_data(dataframe)
+
+    # train and cross validate
+    logging.info("Training started ...")
+    model, score = train_and_score(*train_test_data)
+    logging.info(f"Training completed with model score: {score}")
+
+    # export model to gcs
+    _gcs_uri = os.environ["AIP_MODEL_DIR"]
+    logging.info("Exporting model artifacts ...")
+    export_model_to_gcs(model, _gcs_uri)
+    export_evaluation_report_to_gcs(str(score), _gcs_uri)
+    logging.info(f"Exported model artifacts to GCS bucket: {_gcs_uri}")