00_prep_bq_and_datastore.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "KAdAvk8LquRK"
   },
   "source": [
    "# (Prerequisite) Importing the Sample Data to BigQuery and Datastore\n",
    "\n",
    "This multi-part tutorial shows how to use Matrix Factorization algorithm in BigQuery ML to generate embeddings for items based on their co-occurrence statistics. The generated item embeddings can be then used to find similar items.\n",
    "\n",
    "The prerequisite part covers:\n",
    "\n",
    "1. Copy the `bigquery-samples dataset.playlists` public data to your dataset.\n",
    "2. Export the songs information to Datastore."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "C9tnKiq4q6as"
   },
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "xSNZnVZbEVO_"
   },
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "EMtPGiyVtZTj"
   },
   "outputs": [],
   "source": [
    "import os\n",
    "from datetime import datetime\n",
    "import apache_beam as beam\n",
    "from apache_beam.io.gcp.datastore.v1new.datastoreio import WriteToDatastore"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "bHAObQoaEfC2"
   },
   "source": [
    "### Configure GCP environment settings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "81apxD89q6Co"
   },
   "outputs": [],
   "source": [
    "PROJECT_ID = 'chernikov-playground'\n",
    "BUCKET = 'cooltacular-gcp1'\n",
    "DF_REGION = 'us-west1'\n",
    "\n",
    "BQ_DATASET_NAME = 'recommendations'\n",
    "BQ_TABLE_NAME = 'playlist'\n",
    "BQ_REGION = 'US' # Change to your BigQuery region.\n",
    "DS_KIND = 'song'\n",
    "\n",
    "!gcloud config set project $PROJECT_ID"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "kgjoyd3CrBQi"
   },
   "source": [
    "### Authenticate your GCP account\n",
    "This is required if you run the notebook in Colab"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "hU_altC3pTmd"
   },
   "outputs": [],
   "source": [
    "try:\n",
    "  from google.colab import auth\n",
    "  auth.authenticate_user()\n",
    "  print(\"Colab user is authenticated.\")\n",
    "except: pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4XaUBOKTkzKi"
   },
   "source": [
    "## Copy the BigQuery Playlists data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "lBlQ3RrinurU"
   },
   "source": [
    "### Create BQ Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "d5OHoo3Eowfh"
   },
   "outputs": [],
   "source": [
    "!bq mk -f --dataset \\\n",
    "    --location={BQ_REGION} \\\n",
    "    --project_id={PROJECT_ID} \\\n",
    "    --headless=True \\\n",
    "    {PROJECT_ID}:{BQ_DATASET_NAME}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "WLLs69WFEqTU"
   },
   "source": [
    "### Implement data copying Beam pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "cSwEWeWSkzSA"
   },
   "outputs": [],
   "source": [
    "def run_copy_bq_data_pipeline(args):\n",
    "\n",
    "  schema = 'list_Id:INT64, track_Id:INT64, track_title:STRING, track_artist:STRING'\n",
    "\n",
    "  query = '''\n",
    "    SELECT \n",
    "      id list_Id, \n",
    "      tracks_data_id track_Id, \n",
    "      tracks_data_title track_title,\n",
    "      tracks_data_artist_name track_artist\n",
    "    FROM `bigquery-samples.playlists.playlist`\n",
    "    WHERE tracks_data_title IS NOT NULL AND tracks_data_id > 0\n",
    "    GROUP BY list_Id, track_Id, track_title, track_artist;\n",
    "  '''\n",
    "\n",
    "  pipeline_options = beam.options.pipeline_options.PipelineOptions(**args)\n",
    "  with beam.Pipeline(options=pipeline_options) as pipeline:\n",
    "\n",
    "    _ = (\n",
    "        pipeline\n",
    "        | 'ReadFromBigQuery' >> beam.io.Read(beam.io.BigQuerySource(\n",
    "            project=PROJECT_ID, query=query, use_standard_sql=True))\n",
    "        | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(\n",
    "            table=BQ_TABLE_NAME, dataset=BQ_DATASET_NAME, project=PROJECT_ID,\n",
    "            schema=schema, \n",
    "            create_disposition='CREATE_IF_NEEDED',\n",
    "            write_disposition='WRITE_TRUNCATE'\n",
    "        )\n",
    "    )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "G-PAwjZmEwq8"
   },
   "source": [
    "### Run the Beam pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "gLvUGHUMnVDm"
   },
   "outputs": [],
   "source": [
    "DATASET = 'playlist'\n",
    "RUNNER = 'DataflowRunner'\n",
    "\n",
    "job_name = f'copy-bigquery-{datetime.utcnow().strftime(\"%y%m%d%H%M%S\")}'\n",
    "\n",
    "args = {\n",
    "    'job_name': job_name,\n",
    "    'runner': RUNNER,\n",
    "    'project': PROJECT_ID,\n",
    "    'temp_location': f'gs://{BUCKET}/dataflow_tmp',\n",
    "    'region': DF_REGION\n",
    "}\n",
    "\n",
    "print(\"Pipeline args are set.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "uTpFW5HmnYtq"
   },
   "outputs": [],
   "source": [
    "print(\"Running pipeline...\")\n",
    "%time run_copy_bq_data_pipeline(args)\n",
    "print(\"Pipeline is done.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "EO6yUMUwuxr-"
   },
   "source": [
    "### Create a view to abstract the source table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "z4Qya0Mtux6b"
   },
   "outputs": [],
   "source": [
    "%%bigquery  --project $PROJECT_ID\n",
    "\n",
    "CREATE OR REPLACE VIEW `recommendations.vw_item_groups`\n",
    "AS\n",
    "SELECT\n",
    "  list_Id AS group_Id,\n",
    "  track_Id AS item_Id\n",
    "FROM  \n",
    "  `recommendations.playlist` "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Wr5mm4J3rehj"
   },
   "source": [
    "## Load the Tracks information to Datastore"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "HvBuPJTcresj"
   },
   "source": [
    "### Implement Beam Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "yOANhNR3q_Tc"
   },
   "outputs": [],
   "source": [
    "def create_entity(song_info, kind):\n",
    "\n",
    "  from apache_beam.io.gcp.datastore.v1new.types import Entity\n",
    "  from apache_beam.io.gcp.datastore.v1new.types import Key\n",
    "\n",
    "  track_Id = song_info.pop(\"track_Id\")\n",
    "  key = Key([kind, track_Id])\n",
    "  song_entity = Entity(key)\n",
    "  song_entity.set_properties(song_info)\n",
    "  return song_entity\n",
    "\n",
    "def run_export_to_datatore_pipeline(args):\n",
    "\n",
    "    query = f'''\n",
    "      SELECT  \n",
    "        track_Id, \n",
    "        MAX(track_title) track_title, \n",
    "        MAX(track_artist) artist\n",
    "      FROM \n",
    "        `{BQ_DATASET_NAME}.{BQ_TABLE_NAME}`\n",
    "      GROUP BY track_Id\n",
    "    '''\n",
    "\n",
    "    pipeline_options = beam.options.pipeline_options.PipelineOptions(**args)\n",
    "    with beam.Pipeline(options=pipeline_options) as pipeline:\n",
    "\n",
    "      _ = (\n",
    "        pipeline\n",
    "        | 'ReadFromBigQuery' >> beam.io.Read(beam.io.BigQuerySource(\n",
    "            project=PROJECT_ID, query=query, use_standard_sql=True))\n",
    "        | 'ConvertToDatastoreEntity' >> beam.Map(create_entity, DS_KIND)\n",
    "        | 'WriteToDatastore' >> WriteToDatastore(project=PROJECT_ID)\n",
    "      )\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "Cs_Fyqu7rdsU"
   },
   "source": [
    "### Run pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "D0zcYCsyrdzH"
   },
   "outputs": [],
   "source": [
    "import os\n",
    "from datetime import datetime\n",
    "\n",
    "DATASET = 'playlist'\n",
    "RUNNER = 'DataflowRunner'\n",
    "\n",
    "job_name = f'load-datastore-{datetime.utcnow().strftime(\"%y%m%d%H%M%S\")}'\n",
    "\n",
    "args = {\n",
    "    'job_name': job_name,\n",
    "    'runner': RUNNER,\n",
    "    'project': PROJECT_ID,\n",
    "    'temp_location': f'gs://{BUCKET}/dataflow_tmp',\n",
    "    'region': DF_REGION\n",
    "}\n",
    "\n",
    "print(\"Pipeline args are set.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "74x5kPvQsG7i"
   },
   "outputs": [],
   "source": [
    "print(\"Running pipeline...\")\n",
    "%time run_export_to_datatore_pipeline(args)\n",
    "print(\"Pipeline is done.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "y9AGcKZRfSc-"
   },
   "source": [
    "## License\n",
    "\n",
    "Copyright 2020 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. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0\n",
    "\n",
    "Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. \n",
    "\n",
    "See the License for the specific language governing permissions and limitations under the License.\n",
    "\n",
    "**This is not an official Google product but sample code provided for an educational purpose**"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "00-bq-to-datastore.ipynb",
   "provenance": [],
   "toc_visible": true
  },
  "environment": {
   "name": "tf2-gpu.2-3.m61",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/tf2-gpu.2-3:m61"
  },
  "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.7.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}