00b-load-core-data-to-bq.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data loading and prep\n",
    "\n",
    "**objectives**\n",
    "* load the songs from the zip file\n",
    "* perform transformations to prepare the data for two-tower training\n",
    "\n",
    "**steps**\n",
    "1. Create a bq dataset\n",
    "2. Load the million playlist data to Big Query\n",
    "3. Create pipelines to download audio and artist features for training"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Notebook Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# naming convention for all cloud resources\n",
    "VERSION        = \"v1\"                  # TODO\n",
    "PREFIX         = f'ndr-{VERSION}'      # TODO\n",
    "\n",
    "print(f\"PREFIX = {PREFIX}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# staging GCS\n",
    "GCP_PROJECTS             = !gcloud config get-value project\n",
    "PROJECT_ID               = GCP_PROJECTS[0]\n",
    "\n",
    "# GCS bucket and paths\n",
    "BUCKET_NAME              = f'{PREFIX}-{PROJECT_ID}-bucket'\n",
    "BUCKET_URI               = f'gs://{BUCKET_NAME}'\n",
    "\n",
    "config = !gsutil cat {BUCKET_URI}/config/notebook_env.py\n",
    "print(config.n)\n",
    "exec(config.n)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### env variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Set your variables for your project, region, and dataset name\n",
    "# SOURCE_BUCKET = 'spotify-million-playlist-dataset'\n",
    "# PROJECT_ID = 'hybrid-vertex'\n",
    "# REGION = 'us-central1'\n",
    "# BQ_DATASET = 'spotify_e2e_test'\n",
    "\n",
    "import time\n",
    "from google.cloud import bigquery\n",
    "\n",
    "bigquery_client = bigquery.Client(project=PROJECT_ID)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### create BigQuery dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a bigquery dataset (one time operation)\n",
    "# Construct a full Dataset object to send to the API.\n",
    "dataset = bigquery.Dataset(f\"`{PROJECT_ID}.{BQ_DATASET}`\")\n",
    "dataset.location = REGION\n",
    "\n",
    "# Send the dataset to the API for creation, with an explicit timeout.\n",
    "# Raises google.api_core.exceptions.Conflict if the Dataset already\n",
    "# exists within the project.\n",
    "dataset = bigquery_client.create_dataset(BQ_DATASET, timeout=30)  # Make an API request.\n",
    "\n",
    "print(\"Created dataset {}.{}\".format(bigquery_client.project, dataset.dataset_id))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Download data from GCS\n",
    "\n",
    "(also `curl` from source, see readme)\n",
    "\n",
    "* This step can take up to 30 minutes\n",
    "* Iteration occurs over 1000 json files if you are using the full dataset\n",
    "* This should give you a `playlists` bq data set with 1,076,000 rows (playlists)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# !gsutil cp gs://{SOURCE_BUCKET}/spotify_million_playlist_dataset.zip .\n",
    "# !unzip -n spotify_million_playlist_dataset.zip "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## BigQuery data prep"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import playlists\n",
    "* load JSON files to BigQuery\n",
    "* playlists are nested as one large string that needs to be parsed (use json compatible functionality for BQ)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import json\n",
    "import pandas as pd\n",
    "from tqdm import tqdm\n",
    "import multiprocessing\n",
    "\n",
    "data_files = os.listdir('data')\n",
    "\n",
    "def load_data(filename: str):\n",
    "    \n",
    "    with open(f'data/{filename}') as f:\n",
    "        json_dict = json.load(f)\n",
    "        df = pd.DataFrame(json_dict['playlists'])\n",
    "        df['tracks'] = df['tracks'].map(str)\n",
    "        #write to bq\n",
    "    return df\n",
    "        \n",
    "# make sure there is not already existing data in the playlists table\n",
    "# loops over json files - converts to pandas then upload/appends\n",
    "\n",
    "### add this if you want to limit to smaller number of playlists - this scales significantly later!\n",
    "n_playliststs_limit = None #add if you want to use in for loop: while counter <= n_playliststs_limit:\n",
    "total_files = len(data_files)\n",
    "count = 0\n",
    "batch_size = 15\n",
    "\n",
    "for filename in tqdm(data_files):\n",
    "    if count == 0 or (count-1) % batch_size == 0:\n",
    "        append_df = load_data(filename)\n",
    "        count += 1\n",
    "    \n",
    "    if count % batch_size == 0 or count == total_files:\n",
    "        df = load_data(filename) \n",
    "        append_df = pd.concat([df, append_df])\n",
    "        count += 1\n",
    "        append_df.to_gbq(\n",
    "            destination_table=f'{BQ_DATASET}.playlists', \n",
    "            project_id=PROJECT_ID, # TODO: param\n",
    "            location='US', \n",
    "            progress_bar=False, \n",
    "            reauth=True, \n",
    "            if_exists='append'\n",
    "        )\n",
    "    else:\n",
    "        df = load_data(filename) \n",
    "        append_df = pd.concat([df, append_df])\n",
    "        count += 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### check table\n",
    "\n",
    "<!-- ![](img/tracks-string.png) -->\n",
    "<img\n",
    "  src=\"img/tracks-string.png\"\n",
    "  alt=\"Alt text\"\n",
    "  title=\"BQ table: playlists\"\n",
    "  style=\"display: inline-block; margin: 0 auto; max-width: 1200px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Nested playlists\n",
    "\n",
    "* run parameterized queries to shape the data\n",
    "* This query formats the json strings to be read as Bigquery structs, to be manipulated in subsequent queries\n",
    "* Creates `playlists_nested` by parsing the string data to a struct with arrays"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 148 ms, sys: 51.1 ms, total: 199 ms\n",
      "Wall time: 36.1 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<google.cloud.bigquery.table._EmptyRowIterator at 0x7f495488fe90>"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "json_extract_query = f\"\"\"create or replace table `{PROJECT_ID}.{BQ_DATASET}.playlists_nested` as (\n",
    "with json_parsed as (SELECT * except(tracks), JSON_EXTRACT_ARRAY(tracks) as json_data FROM `{PROJECT_ID}.{BQ_DATASET}.playlists` )\n",
    "\n",
    "select json_parsed.* except(json_data),\n",
    "ARRAY(SELECT AS STRUCT\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.pos\") as pos, \n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.artist_name\") as artist_name,\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.track_uri\") as track_uri,\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.artist_uri\") as artist_uri,\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.track_name\") as track_name,\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.album_uri\") as album_uri,\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.duration_ms\") as duration_ms,\n",
    "JSON_EXTRACT_SCALAR(json_data, \"$.album_name\") as album_name\n",
    "from json_parsed.json_data\n",
    ") as tracks,\n",
    "from json_parsed) \"\"\"\n",
    "\n",
    "bigquery_client.query(json_extract_query).result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### check table\n",
    "\n",
    "<!-- ![](img/playlists-nested.png) -->\n",
    "<img\n",
    "  src=\"img/playlists-nested.png\"\n",
    "  alt=\"Alt text\"\n",
    "  title=\"BQ table: playlists_nested\"\n",
    "  style=\"display: inline-block; margin: 0 auto; max-width: 1200px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Unique tracks\n",
    "\n",
    "* This table will then be used to call the Spotify API and enrich with additional data about each track and artist"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 24.3 ms, sys: 29.4 ms, total: 53.7 ms\n",
      "Wall time: 10.9 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<google.cloud.bigquery.table._EmptyRowIterator at 0x7f4955f3c9d0>"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "unique_tracks_sql = f\"\"\"create or replace table `{PROJECT_ID}.{BQ_DATASET}.tracks_unique` as (\n",
    "SELECT distinct \n",
    "    track.track_uri,\n",
    "    track.album_uri,\n",
    "    track.artist_uri, \n",
    "FROM `{PROJECT_ID}.{BQ_DATASET}.playlists_nested`, UNNEST(tracks) as track)\n",
    "\"\"\"\n",
    "\n",
    "bigquery_client.query(unique_tracks_sql).result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Data loading complete**\n",
    "\n",
    "* We now have our unique id tables we will use for grabbing additional audio and artist features "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "CVKyuWRezBGF"
   },
   "source": [
    "# Spotify API Feature Extraction Pipeline\n",
    "___________"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "IgGWdClhze9D",
    "tags": []
   },
   "source": [
    "**references**\n",
    "* Spotify Mlllion Playlist Dataset Challenge [Homepage](https://www.aicrowd.com/challenges/spotify-million-playlist-dataset-challenge)\n",
    "* [Spotify Web API docs](https://developer.spotify.com/documentation/web-api/reference/#/)\n",
    "\n",
    "**Community Examples**\n",
    "* [Extracting song lists](https://github.com/tojhe/recsys-spotify/blob/master/processing/songlist_extraction.py)\n",
    "* [construct audio features with Spotify API](https://github.com/tojhe/recsys-spotify/blob/master/processing/audio_features_construction.py)\n",
    "* [Using Spotify API](https://towardsdatascience.com/extracting-song-data-from-the-spotify-api-using-python-b1e79388d50)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Spotify Developer Dashboard**\n",
    "<!-- ![](img/spotify-dev-console.png) -->\n",
    "<!-- <img src=\"https://github.com/jswortz/spotify_mpd_two_tower/blob/main/img/spotify-dev-console.png\"  width=\"600\" height=\"300\"> -->\n",
    "\n",
    "<img\n",
    "  src=\"img/spotify-dev-console.png\"\n",
    "  alt=\"Alt text\"\n",
    "  title=\"Spotify Developer Dashboard\"\n",
    "  style=\"display: inline-block; margin: 0 auto; max-width: 900px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**TODO - required**\n",
    "* In your repo, create `spotipy_secret_creds.py`,\n",
    "* assign file to .gitignore\n",
    "* define variables:\n",
    "> * SPOTIPY_CLIENT_ID='YOUR_CLIENT_ID'\n",
    "> * SPOTIPY_CLIENT_SECRET='YOUR_CLIENT_SECRET'\n",
    "> * SPOTIFY_USERNAME='YOUR_USERNAME'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# local py file\n",
    "import spotipy_secret_creds as creds\n",
    "\n",
    "SPOTIPY_CLIENT_ID=creds.SPOTIPY_CLIENT_ID\n",
    "SPOTIPY_CLIENT_SECRET=creds.SPOTIPY_CLIENT_SECRET"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**local json credentials - optional**\n",
    "* if you are concerned about visibility to api keys (credentials), please see [GCP Secret Manager](https://cloud.google.com/secret-manager)\n",
    "* Below is an example if you were to add the json file to secret manager (keys: `secret`, `id`)\n",
    "\n",
    "```python\n",
    "from google.cloud import secretmanager\n",
    "\n",
    "###Note you copy/paste this from secret manager in console\n",
    "SECRET_VERSION = 'projects/934903580331/secrets/spotify-creds1/versions/1'\n",
    "\n",
    "sm_client = secretmanager.SecretManagerServiceClient()\n",
    "\n",
    "name = sm_client.secret_path(PROJECT_ID, SECRET_ID)\n",
    "\n",
    "response = client.access_secret_version(request={\"name\": SECRET_VERSION})   \n",
    "\n",
    "payload = json.loads(response.payload.data.decode(\"UTF-8\"))\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get spotify credentials\n",
    "# This file has id and secret stored as attributes\n",
    "\n",
    "###########################################\n",
    "### CAUTION THIS APPROACH WILL HAVE THE CREDENTIALS APPEAR IN THE CONSOLE - \n",
    "### USE SECRET MANAGER APPROACH IN EACH COMPONENT AS NEEDED (PROVIDED ABOVE)\n",
    "###########################################\n",
    "\n",
    "# # json \n",
    "# creds = open('spotify-creds.json')\n",
    "# spotify_creds = json.load(creds)\n",
    "# creds.close()\n",
    "\n",
    "# SPOTIPY_CLIENT_ID=spotify_creds['id']\n",
    "# SPOTIPY_CLIENT_SECRET=spotify_creds['secret']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### pip & package"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pip install -U spotipy google-cloud-storage google-cloud-aiplatform gcsfs --user -q\n",
    "# pip install --user kfp google-cloud-pipeline-components --upgrade -q\n",
    "# pip install --user -q google-cloud-secret-manager\n",
    "# pip install spotipy --user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "irhbKrY3fcG8",
    "outputId": "2008a7ca-303d-4943-e56b-a57040c9abd7"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "KFP SDK version: 1.8.20\n",
      "google_cloud_pipeline_components version: 1.0.42\n",
      "aiplatform SDK version: 1.23.0\n"
     ]
    }
   ],
   "source": [
    "! python3 -c \"import kfp; print('KFP SDK version: {}'.format(kfp.__version__))\"\n",
    "! python3 -c \"import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))\"\n",
    "! python3 -c \"import google.cloud.aiplatform; print('aiplatform SDK version: {}'.format(google.cloud.aiplatform.__version__))\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "id": "GqhJlsodR-xX"
   },
   "outputs": [],
   "source": [
    "import spotipy\n",
    "from spotipy.oauth2 import SpotifyClientCredentials\n",
    "\n",
    "import re\n",
    "from tqdm import tqdm\n",
    "\n",
    "import os\n",
    "\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import json\n",
    "\n",
    "import time\n",
    "\n",
    "import gcsfs\n",
    "\n",
    "# GCP\n",
    "# from google.cloud import aiplatform\n",
    "from google.cloud import aiplatform as vertex_ai\n",
    "from google.cloud import storage\n",
    "\n",
    "# Pipelines\n",
    "from typing import Any, Callable, Dict, NamedTuple, Optional, List\n",
    "from google_cloud_pipeline_components import aiplatform as gcc_aip\n",
    "from google_cloud_pipeline_components.types import artifact_types\n",
    "\n",
    "# Kubeflow SDK\n",
    "# TODO: fix these\n",
    "from kfp.v2 import dsl\n",
    "import kfp\n",
    "import kfp.v2.dsl\n",
    "from kfp.v2.google import client as pipelines_client\n",
    "from kfp.v2.dsl import (Artifact, Dataset, Input, InputPath, Model, Output,\n",
    "                        OutputPath, component)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "jCbr2tLuzTJK"
   },
   "source": [
    "### env variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "7wwktfnCyzWO",
    "outputId": "6afe4a17-ebc2-40dc-d23a-1f447f394046"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PIPELINE_TAG: pipe-v1-spotify-feature-enrich\n"
     ]
    }
   ],
   "source": [
    "PROJECT_ID = 'hybrid-vertex' #update\n",
    "LOCATION = 'us-central1' \n",
    "\n",
    "BUCKET_NAME = 'matching-engine-content'\n",
    "\n",
    "VERSION = 'v1'\n",
    "PIPELINE_VERSION = f'pipe-v1' # pipeline code\n",
    "PIPELINE_TAG = f'{PIPELINE_VERSION}-spotify-feature-enrich'\n",
    "\n",
    "print(\"PIPELINE_TAG:\", PIPELINE_TAG)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**create bucket if needed**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !gsutil mb -l $LOCATION gs://$BUCKET_NAME"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "IuJk9A5I9ZfF"
   },
   "source": [
    "### client & credentials\n",
    "* Setup Vertex AI client for pipelines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "GCbHe1G_gICe",
    "outputId": "c3f7d046-4997-4b59-9bcc-44df8feba2ed"
   },
   "outputs": [],
   "source": [
    "# # Setup clients\n",
    "vertex_ai.init(\n",
    "    project=PROJECT_ID, \n",
    "    location=LOCATION, \n",
    "    staging_bucket=BUCKET_NAME\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "QbeNm6_whXMG"
   },
   "source": [
    "## Create pipeline components"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# REPO_SRC = 'src'\n",
    "# PIPELINES_SUB_DIR = 'feature_pipes'\n",
    "\n",
    "# ! rm -rf {REPO_SRC}/{PIPELINES_SUB_DIR}\n",
    "! mkdir {REPO_SRC}/{PIPELINES_SUB_DIR}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "8m3D974DjTbE"
   },
   "source": [
    "### component: track audio features\n",
    "\n",
    "[Link to artist API and related features we will pull](https://developer.spotify.com/documentation/web-api/reference/#/operations/get-audio-features)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing src/feature_pipes/call_spotify_api_audio.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile {REPO_SRC}/{PIPELINES_SUB_DIR}/call_spotify_api_audio.py\n",
    "\n",
    "import kfp\n",
    "from typing import Any, Callable, Dict, NamedTuple, Optional, List\n",
    "from kfp.v2.dsl import (Artifact, Dataset, Input, InputPath, Model, Output,\n",
    "                        OutputPath, component, Metrics)\n",
    "\n",
    "@kfp.v2.dsl.component(\n",
    "    base_image=\"python:3.9\",\n",
    "    packages_to_install=[\n",
    "        'fsspec', 'google-cloud-bigquery',\n",
    "        'google-cloud-storage',\n",
    "        'gcsfs',\n",
    "        'spotipy','requests','db-dtypes',\n",
    "        'numpy','pandas','pyarrow','absl-py', 'pandas-gbq==0.17.4',\n",
    "        'tqdm'\n",
    "    ]\n",
    ")\n",
    "def call_spotify_api_audio(\n",
    "    project: str,\n",
    "    location: str,\n",
    "    client_id: str,\n",
    "    batch_size: int,\n",
    "    batches_to_store: int,\n",
    "    target_table: str,\n",
    "    client_secret: str,\n",
    "    unique_table: str,\n",
    "    sleep_param: float,\n",
    ") -> NamedTuple('Outputs', [\n",
    "    ('done_message', str),\n",
    "]):\n",
    "    print(f'pip install complete')\n",
    "    import os\n",
    "    \n",
    "    import spotipy\n",
    "    from spotipy.oauth2 import SpotifyClientCredentials\n",
    "    \n",
    "    import re\n",
    "    import warnings\n",
    "    warnings.simplefilter(action='ignore', category=FutureWarning)\n",
    "    \n",
    "    import pandas as pd\n",
    "    import json\n",
    "    import time\n",
    "    import numpy as np\n",
    "    \n",
    "    from requests.exceptions import ReadTimeout, HTTPError, ConnectionError, RequestException\n",
    "    from absl import logging\n",
    "    \n",
    "    from google.cloud import storage\n",
    "    import gcsfs\n",
    "    from google.cloud import bigquery\n",
    "    \n",
    "    import pandas_gbq\n",
    "    from multiprocessing import Process\n",
    "    from tqdm import tqdm\n",
    "    from tqdm.contrib.logging import logging_redirect_tqdm\n",
    "    \n",
    "    from google.cloud.exceptions import NotFound\n",
    "\n",
    "    import multiprocessing\n",
    "\n",
    "    # print(f'package import complete')\n",
    "\n",
    "    logging.set_verbosity(logging.INFO)\n",
    "    logging.info(f'package import complete')\n",
    "\n",
    "    \n",
    "    bq_client = bigquery.Client(\n",
    "      project=project, location=location\n",
    "    )\n",
    "    \n",
    "    def spot_audio_features(uri, client_id, client_secret):\n",
    "\n",
    "        # Authenticate\n",
    "        client_credentials_manager = SpotifyClientCredentials(\n",
    "            client_id=client_id, \n",
    "            client_secret=client_secret\n",
    "        )\n",
    "        sp = spotipy.Spotify(\n",
    "            client_credentials_manager = client_credentials_manager, \n",
    "            requests_timeout=10, \n",
    "            retries=10\n",
    "        )\n",
    "        ############################################################################\n",
    "        # Create Track Audio Features DF\n",
    "        ############################################################################\n",
    "        \n",
    "        uri_stripped = [u.replace('spotify:track:', '') for u in uri] #fix the quotes \n",
    "        #getting track popularity\n",
    "        tracks = sp.tracks(uri_stripped)\n",
    "        #Audio features\n",
    "        time.sleep(sleep_param)\n",
    "    \n",
    "        a_feats = sp.audio_features(uri)\n",
    "        features = pd.json_normalize(a_feats)#.to_dict('list')\n",
    "        \n",
    "        features['track_pop'] = pd.json_normalize(tracks['tracks'])['popularity']\n",
    "        \n",
    "        features['track_uri'] = uri\n",
    "        return features\n",
    "\n",
    "    bq_client = bigquery.Client(\n",
    "        project=project, \n",
    "        location='US'\n",
    "    )\n",
    "    \n",
    "    #check if target table exists and if so return a list to not duplicate records\n",
    "    try:\n",
    "        bq_client.get_table(target_table)  # Make an API request.\n",
    "        logging.info(\"Table {} already exists.\".format(target_table))\n",
    "        target_table_incomplete_query = f\"select distinct track_uri from `{target_table}`\"\n",
    "        loaded_tracks_df = bq_client.query(target_table_incomplete_query).result().to_dataframe()\n",
    "        loaded_tracks = loaded_tracks_df.track_uri.to_list()\n",
    "        \n",
    "    except NotFound:\n",
    "        logging.info(\"Table {} is not found.\".format(target_table))\n",
    "    \n",
    "    query = f\"select distinct track_uri from `{unique_table}`\" \n",
    "\n",
    "    #refactor\n",
    "    schema = [{'name':'danceability', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'energy', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'key', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'loudness', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'mode', 'type': 'INTEGER', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'speechiness', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'acousticness', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'instrumentalness', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'liveness', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'valence', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'tempo', 'type': 'FLOAT', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'type', 'type': 'STRING', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'id', 'type': 'STRING', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'uri', 'type': 'STRING', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'track_href', 'type': 'STRING', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'analysis_url', 'type': 'STRING', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'duration_ms', 'type': 'INTEGER', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'time_signature', 'type': 'INTEGER', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'track_pop', 'type': 'INTEGER', \"mode\": \"NULLABLE\"},\n",
    "        {'name':'track_uri', 'type': 'STRING', \"mode\": \"REQUIRED\"},\n",
    "    ]\n",
    "    \n",
    "    tracks = bq_client.query(query).result().to_dataframe()\n",
    "    track_list = tracks.track_uri.to_list()\n",
    "    logging.info(f'finished downloading tracks')\n",
    "    \n",
    "    \n",
    "    ### This section is used when there are tracks already loaded into BQ and you want to resume loading the data\n",
    "    try:\n",
    "        track_list = list(set(track_list) - set(loaded_tracks)) #sets the new track list to remove already loaded data in BQ\n",
    "    except:\n",
    "        pass\n",
    "    \n",
    "\n",
    "    from tqdm import tqdm\n",
    "    def process_track_list(track_list):\n",
    "        \n",
    "        uri_list_length = len(track_list)-1 #starting count at zero\n",
    "        inner_batch_count = 0 #avoiding calling the api on 0th iteration\n",
    "        uri_batch = []\n",
    "        \n",
    "        for i, uri in enumerate(tqdm(track_list)):\n",
    "            uri_batch.append(uri)\n",
    "            if (len(uri_batch) == batch_size or uri_list_length == i) and i > 0: #grab a batch of 50 songs\n",
    "                    # logging.info(f\"appending final record for nth song at: {inner_batch_count} \\n i: {i} \\n uri_batch length: {len(uri_batch)}\")\n",
    "                    ### Try catch block for function\n",
    "                try:\n",
    "                    audio_featureDF = spot_audio_features(uri_batch, client_id, client_secret)\n",
    "                    time.sleep(sleep_param)\n",
    "                    uri_batch = []\n",
    "                except ReadTimeout:\n",
    "                    logging.info(\"'Spotify timed out... trying again...'\")\n",
    "                    audio_featureDF = spot_audio_features(uri_batch, client_id, client_secret)\n",
    "                    \n",
    "                    uri_batch = []\n",
    "                    time.sleep(sleep_param)\n",
    "                \n",
    "                except HTTPError as err: #JW ADDED\n",
    "                    logging.info(f\"HTTP Error: {err}\")\n",
    "                \n",
    "                except spotipy.exceptions.SpotifyException as spotify_error: #jw_added\n",
    "                    logging.info(f\"Spotify error: {spotify_error}\")\n",
    "                    \n",
    "                # Accumulate batches on the machine before writing to BQ\n",
    "                # if inner_batch_count <= batches_to_store or uri_list_length == i:\n",
    "                if inner_batch_count == 0:\n",
    "                    appended_data = audio_featureDF\n",
    "                    # logging.info(f\"creating new appended data at IBC: {inner_batch_count} \\n i: {i}\")\n",
    "                    inner_batch_count += 1\n",
    "                elif uri_list_length == i or inner_batch_count == batches_to_store: #send the batches to bq\n",
    "                    appended_data = pd.concat([audio_featureDF, appended_data])\n",
    "                    inner_batch_count = 0\n",
    "                    appended_data.to_gbq(\n",
    "                        destination_table=target_table, \n",
    "                        project_id=f'{project}', \n",
    "                        location='US', \n",
    "                        table_schema=schema,\n",
    "                        progress_bar=False, \n",
    "                        reauth=False, \n",
    "                        if_exists='append'\n",
    "                    )\n",
    "                    logging.info(f'{i+1} of {uri_list_length} complete!')\n",
    "                else:\n",
    "                    appended_data = pd.concat([audio_featureDF, appended_data])\n",
    "                    inner_batch_count += 1\n",
    "\n",
    "        logging.info(f'audio features appended')\n",
    "    \n",
    "    #multiprocessing portion - we will loop based on the modulus of the track_uri list\n",
    "    #chunk the list \n",
    "    \n",
    "    # Yield successive n-sized\n",
    "    # chunks from l.\n",
    "    def divide_chunks(l, n):\n",
    "        # looping till length l\n",
    "        for i in range(0, len(l), n):\n",
    "            yield l[i:i + n]\n",
    "            \n",
    "    n_cores = multiprocessing.cpu_count() \n",
    "    chunked_tracks = list(divide_chunks(track_list, int(len(track_list)/n_cores))) #produces a list of lists chunked evenly by groups of n_cores\n",
    "    \n",
    "    logging.info(\n",
    "        f\"\"\"\n",
    "        total tracks downloaded: {len(track_list)}\\n\n",
    "        length of chunked_tracks: {len(chunked_tracks)}\\n \n",
    "        and inner dims: {[len(x) for x in chunked_tracks]}\n",
    "        \"\"\"\n",
    "    )\n",
    "\n",
    "    procs = []\n",
    "    \n",
    "    def create_job(target, *args):\n",
    "        p = multiprocessing.Process(target=target, args=args)\n",
    "        p.start()\n",
    "        return p\n",
    "\n",
    "    # starting process with arguments\n",
    "    for track_chunk in chunked_tracks:\n",
    "        proc = create_job(process_track_list, track_chunk)\n",
    "        time.sleep(np.pi)\n",
    "        procs.append(proc)\n",
    "\n",
    "    # complete the processes\n",
    "    for proc in procs:\n",
    "        proc.join()\n",
    "        \n",
    "    # process_track_list(track_list) #single thread\n",
    "     \n",
    "    logging.info(f'audio features appended')\n",
    "    \n",
    "    return (\n",
    "          f'DONE',\n",
    "      )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "_hppPSwkh1x2",
    "tags": []
   },
   "source": [
    "### component: artist metadata \n",
    "\n",
    "[Link to artist API and related features we will pull](https://developer.spotify.com/documentation/web-api/reference/#/operations/get-an-artist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "id": "JYbgsLrtxPHl",
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing src/feature_pipes/call_spotify_api_artist.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile {REPO_SRC}/{PIPELINES_SUB_DIR}/call_spotify_api_artist.py\n",
    "\n",
    "import kfp\n",
    "from typing import Any, Callable, Dict, NamedTuple, Optional, List\n",
    "from kfp.v2.dsl import (Artifact, Dataset, Input, InputPath, Model, Output,\n",
    "                        OutputPath, component, Metrics)\n",
    "\n",
    "### Artist tracks api call\n",
    "@kfp.v2.dsl.component(\n",
    "    base_image=\"python:3.9\",\n",
    "    packages_to_install=[\n",
    "        'fsspec',' google-cloud-bigquery',\n",
    "        'google-cloud-storage',\n",
    "        'gcsfs', 'tqdm',\n",
    "        'spotipy','requests','db-dtypes',\n",
    "        'numpy','pandas','pyarrow','absl-py', 'pandas-gbq==0.17.4',\n",
    "        'google-cloud-secret-manager'\n",
    "    ]\n",
    ")\n",
    "def call_spotify_api_artist(\n",
    "    project: str,\n",
    "    location: str,\n",
    "    unique_table: str,\n",
    "    batch_size: int,\n",
    "    batches_to_store: int,\n",
    "    client_id: str,\n",
    "    client_secret: str,\n",
    "    sleep_param: float,\n",
    "    target_table: str,\n",
    ") -> NamedTuple('Outputs', [\n",
    "    ('done_message', str),\n",
    "]):\n",
    "    print(f'pip install complete')\n",
    "    \n",
    "    import os\n",
    "    \n",
    "    import spotipy\n",
    "    from spotipy.oauth2 import SpotifyClientCredentials\n",
    "    \n",
    "    import re\n",
    "    import warnings\n",
    "    warnings.simplefilter(action='ignore', category=FutureWarning)\n",
    "    \n",
    "    import pandas as pd\n",
    "    import json\n",
    "    import time\n",
    "    import numpy as np\n",
    "    \n",
    "    from google.cloud import storage\n",
    "    import gcsfs\n",
    "    from google.cloud import bigquery\n",
    "\n",
    "    from requests.exceptions import ReadTimeout, HTTPError, ConnectionError, RequestException\n",
    "    from absl import logging\n",
    "\n",
    "    import pandas_gbq\n",
    "    pd.options.mode.chained_assignment = None  # default='warn'\n",
    "    from google.cloud.exceptions import NotFound\n",
    "    \n",
    "    from multiprocessing import Process\n",
    "    import multiprocessing\n",
    "\n",
    "    logging.set_verbosity(logging.INFO)\n",
    "    logging.info(f'package import complete')\n",
    "\n",
    "    storage_client = storage.Client(\n",
    "        project=project\n",
    "    )\n",
    "    \n",
    "    logging.info(f'spotipy auth complete')\n",
    "    \n",
    "    def spot_artist_features(uri, client_id, client_secret):\n",
    "\n",
    "        # Authenticate\n",
    "        client_credentials_manager = SpotifyClientCredentials(\n",
    "            client_id=client_id, \n",
    "            client_secret=client_secret\n",
    "        )\n",
    "        sp = spotipy.Spotify(\n",
    "            client_credentials_manager = client_credentials_manager, \n",
    "            requests_timeout=2, \n",
    "            retries=1 )\n",
    "\n",
    "        ############################################################################\n",
    "        # Create Track Audio Features DF\n",
    "        ############################################################################ \n",
    "\n",
    "        # uri = [u.replace('spotify:artist:', '') for u in uri] #fix the quotes \n",
    "\n",
    "        artists = sp.artists(uri)\n",
    "        features = pd.json_normalize(artists['artists'])\n",
    "        smaller_features = features[['genres', 'popularity', 'name', 'followers.total']]\n",
    "        smaller_features.columns = ['genres_list',  'artist_pop', 'name',  'followers']\n",
    "        smaller_features['artist_uri'] = uri\n",
    "        smaller_features['genres'] = smaller_features['genres_list'].map(lambda x: f\"{x}\")\n",
    "        return smaller_features[['genres', 'artist_pop', 'artist_uri', 'followers']]\n",
    "        \n",
    "\n",
    "    bq_client = bigquery.Client(\n",
    "      project=project, location='US'\n",
    "    )\n",
    "\n",
    "    #check if target table exists and if so return a list to not duplicate records\n",
    "    try:\n",
    "        bq_client.get_table(target_table)  # Make an API request.\n",
    "        logging.info(\"Table {} already exists.\".format(target_table))\n",
    "        target_table_incomplete_query = f\"select distinct artist_uri from `{target_table}`\"\n",
    "        loaded_tracks_df = bq_client.query(target_table_incomplete_query).result().to_dataframe()\n",
    "        loaded_tracks = loaded_tracks_df.artist_uri.to_list()\n",
    "        \n",
    "    except NotFound:\n",
    "        logging.info(\"Table {} is not found.\".format(target_table))\n",
    "        \n",
    "        \n",
    "    query = f\"select distinct artist_uri from `{unique_table}`\"\n",
    "    \n",
    "\n",
    "    schema = [\n",
    "        {'name': 'artist_pop', 'type': 'INTEGER'},\n",
    "        {'name':'genres', 'type': 'STRING'},\n",
    "        {'name':'followers', 'type': 'INTEGER'},\n",
    "        {'name':'artist_uri', 'type': 'STRING'}\n",
    "    ]\n",
    "    \n",
    "    \n",
    "    tracks = bq_client.query(query).result().to_dataframe()\n",
    "    track_list = tracks.artist_uri.to_list()\n",
    "    logging.info(f'finished downloading tracks')\n",
    "    \n",
    "    from tqdm import tqdm\n",
    "    def process_track_list(track_list):\n",
    "        uri_list_length = len(track_list)-1 #starting count at zero\n",
    "        inner_batch_count = 0 #avoiding calling the api on 0th iteration\n",
    "        uri_batch = []\n",
    "        for i, uri in enumerate(tqdm(track_list)):\n",
    "            uri_batch.append(uri)\n",
    "            if (len(uri_batch) == batch_size or uri_list_length == i) and i > 0: #grab a batch of 50 songs\n",
    "                ### Try catch block for function\n",
    "                try:\n",
    "                    audio_featureDF = spot_artist_features(uri_batch, client_id, client_secret)\n",
    "                    time.sleep(sleep_param)\n",
    "                    uri_batch = []\n",
    "                \n",
    "                except ReadTimeout:\n",
    "                    logging.info(\"'Spotify timed out... trying again...'\")\n",
    "                    audio_featureDF = spot_artist_features(uri_batch, client_id, client_secret)\n",
    "                    uri_batch = []\n",
    "                    time.sleep(sleep_param)\n",
    "                \n",
    "                except HTTPError as err: #JW ADDED\n",
    "                    logging.info(f\"HTTP Error: {err}\")\n",
    "                \n",
    "                except spotipy.exceptions.SpotifyException as spotify_error: #jw_added\n",
    "                    logging.info(f\"Spotify error: {spotify_error}\")\n",
    "                \n",
    "                # Accumulate batches on the machine before writing to BQ\n",
    "                # if inner_batch_count <= batches_to_store or uri_list_length == i:\n",
    "                \n",
    "                if inner_batch_count == 0:\n",
    "                    appended_data = audio_featureDF\n",
    "                    # logging.info(f\"creating new appended data at IBC: {inner_batch_count} \\n i: {i}\")\n",
    "                    inner_batch_count += 1\n",
    "                \n",
    "                elif uri_list_length == i or inner_batch_count == batches_to_store: #send the batches to bq\n",
    "                    appended_data = pd.concat([audio_featureDF, appended_data])\n",
    "                    inner_batch_count = 0\n",
    "                    appended_data.to_gbq(\n",
    "                        destination_table=target_table, \n",
    "                        project_id=f'{project}', \n",
    "                        location='US', \n",
    "                        table_schema=schema,\n",
    "                        progress_bar=False, \n",
    "                        reauth=False, \n",
    "                        if_exists='append'\n",
    "                    )\n",
    "                    logging.info(f'{i+1} of {uri_list_length} complete!')\n",
    "                \n",
    "                else:\n",
    "                    appended_data = pd.concat([audio_featureDF, appended_data])\n",
    "                    inner_batch_count += 1\n",
    "\n",
    "        logging.info(f'audio features appended')\n",
    "    \n",
    "    #multiprocessing portion - we will loop based on the modulus of the track_uri list\n",
    "    #chunk the list \n",
    "    \n",
    "    # Yield successive n-sized\n",
    "    # chunks from l.\n",
    "    def divide_chunks(l, n):\n",
    "        # looping till length l\n",
    "        for i in range(0, len(l), n):\n",
    "            yield l[i:i + n]\n",
    "            \n",
    "    n_cores = multiprocessing.cpu_count() \n",
    "    chunked_tracks = list(divide_chunks(track_list, int(len(track_list)/n_cores))) # produces a list of lists chunked evenly by groups of n_cores\n",
    "    \n",
    "    logging.info(\n",
    "        f\"\"\"total tracks downloaded: {len(track_list)}\\n\n",
    "        length of chunked_tracks: {len(chunked_tracks)}\\n \n",
    "        and inner dims: {[len(x) for x in chunked_tracks]}\n",
    "        \"\"\"\n",
    "    )\n",
    "    \n",
    "    procs = []\n",
    "    def create_job(target, *args):\n",
    "        p = multiprocessing.Process(target=target, args=args)\n",
    "        p.start()\n",
    "        return p\n",
    "\n",
    "    # starting process with arguments\n",
    "    for track_chunk in chunked_tracks:\n",
    "        proc = create_job(process_track_list, track_chunk)\n",
    "        time.sleep(np.pi)\n",
    "        procs.append(proc)\n",
    "\n",
    "    # complete the processes\n",
    "    for proc in procs:\n",
    "        proc.join()\n",
    "        \n",
    "    # process_track_list(track_list)\n",
    "    logging.info(f'artist features appended')\n",
    "    \n",
    "    return (\n",
    "          f'DONE',\n",
    "      )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Ma5qzpGkuYhD"
   },
   "source": [
    "## Build pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "REPO_SRC: src\n",
      "PIPELINES_SUB_DIR: feature_pipes\n"
     ]
    }
   ],
   "source": [
    "print(f'REPO_SRC: {REPO_SRC}')\n",
    "print(f'PIPELINES_SUB_DIR: {PIPELINES_SUB_DIR}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "id": "Buwtyt7rugt4"
   },
   "outputs": [],
   "source": [
    "from src.feature_pipes import call_spotify_api_audio, call_spotify_api_artist\n",
    "\n",
    "@kfp.v2.dsl.pipeline(\n",
    "  name=f'spotify-feature-enrichment-{PIPELINE_TAG}'.replace('_', '-')\n",
    ")\n",
    "def pipeline(\n",
    "    project: str,\n",
    "    location: str,\n",
    "    unique_table: str,\n",
    "    target_table_audio: str,\n",
    "    target_table_artist: str,\n",
    "    target_table_popularity: str,\n",
    "    batch_size: int,\n",
    "    batches_to_store: int,\n",
    "    sleep_param: float,\n",
    "    spotify_id: str = SPOTIPY_CLIENT_ID, # = spotify_creds['id'],\n",
    "    spotify_secret: str = SPOTIPY_CLIENT_SECRET, # = spotify_creds['secret'],\n",
    "):\n",
    "    \n",
    "    call_spotify_api_artist_op = (\n",
    "        call_spotify_api_artist.call_spotify_api_artist(\n",
    "            project=project,\n",
    "            location=location,\n",
    "            client_id=spotify_id,\n",
    "            client_secret=spotify_secret,\n",
    "            batch_size=batch_size,\n",
    "            sleep_param=sleep_param,\n",
    "            unique_table=unique_table,\n",
    "            target_table=target_table_artist,\n",
    "            batches_to_store=batches_to_store,\n",
    "        )\n",
    "        .set_display_name(\"Get Artist Features From Spotify API\")\n",
    "    )\n",
    "\n",
    "    call_spotify_api_audio_op = (\n",
    "        call_spotify_api_audio.call_spotify_api_audio(\n",
    "            project=project,\n",
    "            location=location,\n",
    "            client_id=spotify_id,\n",
    "            client_secret=spotify_secret,\n",
    "            batch_size=batch_size,\n",
    "            sleep_param=sleep_param,\n",
    "            unique_table=unique_table,\n",
    "            target_table=target_table_audio,\n",
    "            batches_to_store=batches_to_store,\n",
    "        )\n",
    "        .set_display_name(\"Get Track Audio Features From Spotify API\")\n",
    "        .after(call_spotify_api_artist_op)\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compile pipeline \n",
    "* compiles to json\n",
    "* store in GCS bucket for tracking"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "RoBsfT1IyIXd",
    "outputId": "50a65461-1c7c-44c0-9362-5524aaac21c4"
   },
   "outputs": [],
   "source": [
    "PIPELINE_JSON_SPEC_LOCAL = \"custom_track_meta_pipeline_spec.json\"\n",
    "\n",
    "! rm -f $PIPELINE_JSON_SPEC_LOCAL\n",
    "\n",
    "kfp.v2.compiler.Compiler().compile(\n",
    "  pipeline_func=pipeline, \n",
    "  package_path=PIPELINE_JSON_SPEC_LOCAL,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PIPELINES_FILEPATH: gs://matching-engine-content/v1/pipeline_root/custom_track_meta_pipeline_spec.json\n"
     ]
    }
   ],
   "source": [
    "PIPELINE_ROOT_PATH = f'gs://{BUCKET_NAME}/{VERSION}/pipeline_root'\n",
    "\n",
    "PIPELINES_FILEPATH = f'{PIPELINE_ROOT_PATH}/{PIPELINE_JSON_SPEC_LOCAL}'\n",
    "\n",
    "print(\"PIPELINES_FILEPATH:\", PIPELINES_FILEPATH)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Copying file://custom_track_meta_pipeline_spec.json [Content-Type=application/json]...\n",
      "/ [1 files][ 24.9 KiB/ 24.9 KiB]                                                \n",
      "Operation completed over 1 objects/24.9 KiB.                                     \n"
     ]
    }
   ],
   "source": [
    "!gsutil cp $PIPELINE_JSON_SPEC_LOCAL $PIPELINES_FILEPATH"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Submit pipeline\n",
    "\n",
    "* Set pipeline parameters dict\n",
    "* create pipeline job\n",
    "* submit"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### set pipeline params"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "id": "evUgOHllykr5"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'project': 'hybrid-vertex',\n",
       " 'location': 'us-central1',\n",
       " 'unique_table': 'hybrid-vertex.spotify_e2e_test.tracks_unique',\n",
       " 'target_table_audio': 'hybrid-vertex.spotify_e2e_test.audio_features',\n",
       " 'target_table_artist': 'hybrid-vertex.spotify_e2e_test.artist_features',\n",
       " 'target_table_popularity': 'hybrid-vertex.spotify_e2e_test.track_popularity',\n",
       " 'batch_size': 50,\n",
       " 'batches_to_store': 400,\n",
       " 'sleep_param': 0.5}"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# BUCKET_NAME = 'matching-engine-content'\n",
    "\n",
    "ideal_batch_size = 20_000\n",
    "bts = ideal_batch_size / 50\n",
    "\n",
    "PIPELINE_PARAMETERS = dict(\n",
    "    project = PROJECT_ID,\n",
    "    location = 'us-central1',\n",
    "    unique_table = f'{PROJECT_ID}.{BQ_DATASET}.tracks_unique', \n",
    "    target_table_audio = f'{PROJECT_ID}.{BQ_DATASET}.audio_features',\n",
    "    target_table_artist = f'{PROJECT_ID}.{BQ_DATASET}.artist_features',\n",
    "    target_table_popularity = f'{PROJECT_ID}.{BQ_DATASET}.track_popularity',\n",
    "    batch_size = 50,\n",
    "    batches_to_store = int(bts),\n",
    "    sleep_param = .5,\n",
    "    # spotify_id = SPOTIPY_CLIENT_ID,\n",
    "    # spotify_secret = SPOTIPY_CLIENT_SECRET\n",
    ")\n",
    "\n",
    "PIPELINE_PARAMETERS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### create pipeline job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job = aiplatform.PipelineJob(\n",
    "    display_name = f'spotify-feature-enrichment-{PIPELINE_TAG}'.replace('_', '-'),\n",
    "    template_path = PIPELINE_JSON_SPEC_LOCAL,\n",
    "    pipeline_root = f'gs://{BUCKET_NAME}/{VERSION}',\n",
    "    parameter_values = PIPELINE_PARAMETERS,\n",
    "    project = PROJECT_ID,\n",
    "    location = LOCATION,\n",
    "    enable_caching=False\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### submit pipeline job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating PipelineJob\n",
      "PipelineJob created. Resource name: projects/934903580331/locations/us-central1/pipelineJobs/spotify-feature-enrichment-v13-spotify-feature-enrich-20230307013206\n",
      "To use this PipelineJob in another session:\n",
      "pipeline_job = aiplatform.PipelineJob.get('projects/934903580331/locations/us-central1/pipelineJobs/spotify-feature-enrichment-v13-spotify-feature-enrich-20230307013206')\n",
      "View Pipeline Job:\n",
      "https://console.cloud.google.com/vertex-ai/locations/us-central1/pipelines/runs/spotify-feature-enrichment-v13-spotify-feature-enrich-20230307013206?project=934903580331\n"
     ]
    }
   ],
   "source": [
    "job.submit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Vertex Pipeline UI (console) \n",
    "\n",
    "* It may take a couple of runs with different application credientials for the audio tracks\n",
    "* This module can resume from where data was already loaded to BQ\n",
    "\n",
    "<!-- ![](img/feature-enrich-pipeline.png) -->\n",
    "\n",
    "<img\n",
    "  src=\"img/feature-enrich-pipeline.png\"\n",
    "  alt=\"Alt text\"\n",
    "  title=\"Feature Enrichment Pipeline\"\n",
    "  style=\"display: inline-block; margin: 0 auto; max-width: 900px\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## GitIgnore"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting .gitignore\n"
     ]
    }
   ],
   "source": [
    "%%writefile .gitignore\n",
    "*.cpython-310.pyc\n",
    "*-checkpoint.*\n",
    "*-checkpoint.md\n",
    "*-checkpoint.py\n",
    "*.ipynb_checkpoints\n",
    "# .gcloudignore\n",
    "# .git\n",
    "# .github\n",
    ".ipynb_checkpoints/*\n",
    "*__pycache__\n",
    "# *cpython-37.pyc\n",
    "# .gitignore\n",
    "# .DS_Store"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Next Steps\n",
    "\n",
    "Now that all data is loaded in BigQuery, the [01-bq-data-prep.ipynb](01-bq-data-prep.ipynb) notebook will finish feature prep"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "Extract-spotify-features-pipeline-parallel-for.ipynb",
   "provenance": [],
   "toc_visible": true
  },
  "environment": {
   "kernel": "python3",
   "name": "tf2-gpu.2-10.m103",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-gpu.2-10:m103"
  },
  "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.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}