start-version-vertex-ai-pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b7667d1b",
   "metadata": {},
   "outputs": [],
   "source": [
    "! pip install google-cloud-aiplatform==1.16.1 kfp==1.8.20"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "184acc8f-29fd-4a24-921a-c751f8b3d4a8",
   "metadata": {},
   "outputs": [],
   "source": [
    "from kfp.v2.dsl import Artifact, Dataset, Input, Metrics, Model, Output, component, pipeline\n",
    "from kfp.v2 import compiler\n",
    "\n",
    "# Component to get data from BigQuery (the database where the data is hosted)\n",
    "@component(\n",
    "    base_image=\"python:3.9-slim\", \n",
    "    packages_to_install=[\n",
    "        \"google-cloud-bigquery==2.34.4\",\n",
    "        \"pandas==2.0.1\",\n",
    "        \"db-dtypes==1.1.1\"\n",
    "    ]\n",
    ")\n",
    "def get_source_data(\n",
    "    source_data: Output[Dataset],\n",
    "    project_id_host_project: str = \"pj-dgbe-vtk-workshop-2023-host\",\n",
    "    dataset_name: str = \"source_data\",\n",
    "    table_name: str = \"housing\"\n",
    "):\n",
    "    \n",
    "    from google.cloud import bigquery\n",
    "    bigquery_client = bigquery.Client()\n",
    "    \n",
    "    dataset_ref = bigquery.DatasetReference(project_id_host_project, dataset_name)\n",
    "    table_ref = dataset_ref.table(table_name)\n",
    "    table = bigquery_client.get_table(table_ref)\n",
    "    \n",
    "    src_data = bigquery_client.list_rows(table).to_dataframe()\n",
    "    \n",
    "    src_data.to_csv(\n",
    "        source_data.path, \n",
    "        index=False,\n",
    "        header=True\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b0c4975d-8372-41fd-a642-c1b41b747718",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Component to process data and split it into test and train sets\n",
    "@component(\n",
    "    base_image=\"python:3.9-slim\",\n",
    "    packages_to_install=[\n",
    "        \"pandas==2.0.1\",\n",
    "        \"scikit-learn==1.2.2\"\n",
    "    ]\n",
    ")\n",
    "def process_source_data(\n",
    "    source_data: Input[Dataset],\n",
    "    features_train: Output[Dataset],\n",
    "    target_train: Output[Dataset],\n",
    "    features_test: Output[Dataset],\n",
    "    target_test: Output[Dataset]\n",
    "):\n",
    "    import pandas as pd\n",
    "    from sklearn.model_selection import train_test_split\n",
    "    \n",
    "    src_data = pd.read_csv(source_data.path)\n",
    "    \n",
    "    ##############################################################\n",
    "    ## TODO: try to use feature 'furnishingstatus' as a feature ##\n",
    "    ## hint: this is a string and must be converted to a number ##\n",
    "    ## before it can be used                                    ##\n",
    "    ##############################################################\n",
    "    features = src_data.drop('price', axis=1).drop('furnishingstatus', axis=1)\n",
    "    \n",
    "    target = src_data['price']\n",
    "    \n",
    "    ftrs_train, ftrs_test, trgt_train, trgt_test = train_test_split(features,target)\n",
    "    \n",
    "    ftrs_train.to_csv(\n",
    "        features_train.path,\n",
    "        index=False,\n",
    "        header=True\n",
    "    )\n",
    "    \n",
    "    trgt_train.to_csv(\n",
    "        target_train.path,\n",
    "        index=False,\n",
    "        header=True\n",
    "    )\n",
    "    \n",
    "    ftrs_test.to_csv(\n",
    "        features_test.path,\n",
    "        index=False,\n",
    "        header=True\n",
    "    )\n",
    "        \n",
    "    trgt_test.to_csv(\n",
    "        target_test.path,\n",
    "        index=False,\n",
    "        header=True\n",
    "    )\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8d7b926e-223d-446d-9443-6f1d599a5162",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(\n",
    "    base_image=\"python:3.9-slim\",\n",
    "    packages_to_install=[\n",
    "        \"pandas==2.0.1\", \n",
    "        \"scikit-learn==1.2.2\"\n",
    "    ]\n",
    ")\n",
    "def train_model(\n",
    "    features_train: Input[Dataset],\n",
    "    target_train: Input[Dataset],\n",
    "    features_test: Input[Dataset],\n",
    "    target_test: Input[Dataset],\n",
    "    model: Output[Model], \n",
    "    metrics: Output[Metrics]\n",
    "):\n",
    "    \n",
    "    import pandas as pd\n",
    "    import pickle\n",
    "    from sklearn.tree import DecisionTreeRegressor\n",
    "    \n",
    "    ftrs_train = pd.read_csv(features_train.path)\n",
    "    trgt_train = pd.read_csv(target_train.path)\n",
    "    ftrs_test = pd.read_csv(features_test.path)\n",
    "    trgt_test = pd.read_csv(target_test.path)\n",
    "    \n",
    "    ################################################\n",
    "    ## TODO: add and train any sklearn model here ##\n",
    "    ################################################\n",
    "    ml_model = \"\"\n",
    "    \n",
    "    with open(model.path, \"wb\") as f:\n",
    "        pickle.dump(ml_model, f)\n",
    "    \n",
    "    # Calculate and export squared_error\n",
    "    squared_error = ml_model.score(ftrs_test, trgt_test)\n",
    "    metrics.log_metric(\"squared error\", (squared_error * 100.0))\n",
    "    metrics.log_metric(\"framework\", \"Scikit Learn\")\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b79d6098-3228-4f96-80ab-e966a7863fde",
   "metadata": {},
   "outputs": [],
   "source": [
    "@component(\n",
    "    base_image='python:3.9-slim',\n",
    "    packages_to_install=[\n",
    "        'google-cloud-aiplatform==1.16.1', \n",
    "        'kfp-server-api==1.8'\n",
    "    ]\n",
    ")\n",
    "def deploy_model(\n",
    "    trained_model: Input[Model],\n",
    "    project_id: str,\n",
    "    region: str,\n",
    "    endpoint_name: str,\n",
    "    model_display_name: str\n",
    "):\n",
    "    \"\"\"\n",
    "    Deploy model to Vertex AI endpoint\n",
    "    :param trained_model:       Model artifact to deploy\n",
    "    :param feature_data_scaler: Scaler artifact to scale input data\n",
    "    :param target_data_scaler:  Scaler artifact to scale target data\n",
    "    :param region:              Region to init AI Platform\n",
    "    :param project_id:          Project ID\n",
    "    :param endpoint_name:       Name of endpoint\n",
    "    :param model_display_name:  Display name of model\n",
    "    :param serving_image:       Name of serving image to use for custom predictions\n",
    "    :param service_account:     Service account that will use the serving image\n",
    "    \"\"\"\n",
    "\n",
    "    import google.cloud.aiplatform as aip\n",
    "\n",
    "    def extract_or_create_endpoint(endpoint_display_name: str):\n",
    "        \"\"\"\n",
    "        Looks for an existing Vertex AI Endpoint with a given display name. If not found, creates a new one\n",
    "        \n",
    "        :param endpoint_display_name: display name of the Vertex AI Endpoint\n",
    "        :return: created/retrieved endpoint object\n",
    "        \"\"\"\n",
    "        existing_endpoints = aip.Endpoint.list()\n",
    "        found_endpoint = False\n",
    "        i = 0\n",
    "        while i < len(existing_endpoints) and not found_endpoint:\n",
    "            endpoint_retrieved = existing_endpoints[i]\n",
    "            if endpoint_retrieved.display_name == endpoint_display_name:\n",
    "                endpoint = endpoint_retrieved\n",
    "                found_endpoint = True\n",
    "            i += 1\n",
    "        if not found_endpoint:\n",
    "            endpoint = aip.Endpoint.create(display_name=endpoint_display_name)\n",
    "        return endpoint\n",
    "\n",
    "    def model_retrieve(model_display_name: str):\n",
    "        \"\"\"\n",
    "        Looks for an existing Vertex AI Model with a given display name\n",
    "        :param model_display_name: display name of the Vertex AI Mndpoint\n",
    "        :return: Model if found, else None\n",
    "        \"\"\"\n",
    "        models = aip.Model.list()\n",
    "        for model in models:\n",
    "            if model.display_name == model_display_name:\n",
    "                return model\n",
    "        return None\n",
    "\n",
    "    aip.init(\n",
    "        project=project_id,\n",
    "        location=region,\n",
    "    )\n",
    "\n",
    "    endpoint = extract_or_create_endpoint(endpoint_name)\n",
    "    artifact_uri = trained_model.path\n",
    "\n",
    "    model = model_retrieve(model_display_name)\n",
    "\n",
    "    model = aip.Model.upload_scikit_learn_model_file(\n",
    "        model_file_path = trained_model.path\n",
    "    )\n",
    "\n",
    "    model.deploy(\n",
    "        endpoint=endpoint,\n",
    "        machine_type=\"n1-standard-4\",\n",
    "        min_replica_count=1,\n",
    "        max_replica_count=1,\n",
    "        accelerator_type=None,\n",
    "        accelerator_count=None,\n",
    "        traffic_split={\"0\": 100},\n",
    "        deployed_model_display_name=model_display_name\n",
    "    )\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f0c39e3-715f-4872-8428-7264e294e3cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "@pipeline(name=\"pipeline-sklearn\")\n",
    "def my_pipeline(\n",
    "    project_id: str,\n",
    "    region: str,\n",
    "    endpoint_name: str,\n",
    "    model_display_name: str\n",
    "):\n",
    "    \n",
    "    source_data = get_source_data()\n",
    "    \n",
    "    processed_source_data = process_source_data(\n",
    "        source_data = source_data.outputs[\"source_data\"]\n",
    "    )\n",
    "    \n",
    "    trained_model = train_model(\n",
    "        features_train = processed_source_data.outputs[\"features_train\"],\n",
    "        target_train = processed_source_data.outputs[\"target_train\"],\n",
    "        features_test = processed_source_data.outputs[\"features_test\"],\n",
    "        target_test = processed_source_data.outputs[\"target_test\"]\n",
    "    )\n",
    "    \n",
    "    deploy_model(\n",
    "        trained_model = trained_model.outputs[\"model\"],\n",
    "        project_id = project_id,\n",
    "        region = region,\n",
    "        endpoint_name = endpoint_name,\n",
    "        model_display_name = model_display_name\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c82bbd90-aa9d-407d-aeb0-6a3a5c895352",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "85e0af86-217c-4982-8d00-83a999e8270e",
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "\n",
    "from google.cloud import aiplatform\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    compiler.Compiler().compile(my_pipeline,package_path=\"pipeline.json\")\n",
    "\n",
    "    model_display_name = \"predict-house-prices\"\n",
    "    project_id = \"dt-vtk-workshop-1\"\n",
    "    region = \"europe-west1\"\n",
    "    pipeline_root_bucket = \"gs://gcs-be-dt-vtk-workshop-1-pipelines\"\n",
    "    \n",
    "    run = aiplatform.PipelineJob(\n",
    "        project=project_id,\n",
    "        location=region,\n",
    "        display_name=model_display_name,\n",
    "        template_path=\"pipeline.json\",\n",
    "        job_id=f\"test-pipeline-{datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}\",\n",
    "        enable_caching=True,\n",
    "        pipeline_root=pipeline_root_bucket,\n",
    "        parameter_values={\n",
    "            \"project_id\": project_id,\n",
    "            \"region\": region,\n",
    "            \"endpoint_name\": \"endp-predict-house-prices\",\n",
    "            \"model_display_name\": model_display_name\n",
    "        },\n",
    "    )\n",
    "\n",
    "    run.submit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b50c7061-e9ed-4280-bd2e-a69fdbb8e9ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Use the model to do a prediction\n",
    "\n",
    "def endpoint_predict_sample(\n",
    "    to_predict: list,\n",
    "    project: str = project_id, \n",
    "    location: str = region, \n",
    "    endpoint: str = \"1833536794386235392\" # example: \"4580732567082237952\"\n",
    "):\n",
    "    aiplatform.init(project=project, location=location)\n",
    "\n",
    "    endpoint = aiplatform.Endpoint(endpoint)\n",
    "\n",
    "    prediction = endpoint.predict(instances=to_predict)\n",
    "    print(prediction)\n",
    "    return prediction\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "425c6da1-3fa4-4aa1-add2-6d922f21fbcd",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_sample_requests_list = [[3450,1,1,1,True,False,False,True,False,0,False]]\n",
    "prediction = endpoint_predict_sample(df_sample_requests_list)\n",
    "print(prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9dc63757-4ec4-4ae8-a8d6-6817c8351be4",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "153e403d-a709-43ef-8b09-bb25d17ae15a",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "environment": {
   "kernel": "python3",
   "name": "tf2-cpu.2-11.m107",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-cpu.2-11:m107"
  },
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}