hybrid_recommendations_preproc.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "# Neural network hybrid recommendation system on Google Analytics data preprocessing\n",
    "\n",
    "This notebook demonstrates how to implement a hybrid recommendation system using a neural network to combine content-based and collaborative filtering recommendation models using Google Analytics data. We are going to use the learned user embeddings from [wals.ipynb](../wals.ipynb) and combine that with our previous content-based features from [content_based_using_neural_networks.ipynb](../content_based_using_neural_networks.ipynb)\n",
    "\n",
    "First we are going to preprocess our data using BigQuery and Cloud Dataflow to be used in our later neural network hybrid recommendation model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Apache Beam only works in Python 2 at the moment, so we're going to switch to the Python 2 kernel. In the above menu, click the dropdown arrow and select `python2`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "conda update -y -n base -c defaults conda\n",
    "source activate py2env\n",
    "pip uninstall -y google-cloud-dataflow\n",
    "conda install -y pytz\n",
    "pip install apache-beam[gcp]==2.9.0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Now restart notebook's session kernel!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "# Import helpful libraries and setup our project, bucket, and region\n",
    "import os\n",
    "\n",
    "PROJECT = \"cloud-training-demos\" # REPLACE WITH YOUR PROJECT ID\n",
    "BUCKET = \"cloud-training-demos-ml\" # REPLACE WITH YOUR BUCKET NAME\n",
    "REGION = \"us-central1\" # REPLACE WITH YOUR BUCKET REGION e.g. us-central1\n",
    "\n",
    "# Do not change these\n",
    "os.environ[\"PROJECT\"] = PROJECT\n",
    "os.environ[\"BUCKET\"] = BUCKET\n",
    "os.environ[\"REGION\"] = REGION\n",
    "os.environ[\"TFVERSION\"] = \"1.13\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Updated property [core/project].\n",
      "Updated property [compute/region].\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "gcloud  config  set project $PROJECT\n",
    "gcloud config set compute/region $REGION"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<h2> Create ML dataset using Dataflow </h2>\n",
    "Let's use Cloud Dataflow to read in the BigQuery data, do some preprocessing, and write it out as CSV files.\n",
    "\n",
    "First, let's create our hybrid dataset query that we will use in our Cloud Dataflow pipeline. This will combine some content-based features and the user and item embeddings learned from our WALS Matrix Factorization Collaborative filtering lab that we extracted from our trained WALSMatrixFactorization Estimator and uploaded to BigQuery."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "query_hybrid_dataset = \"\"\"\n",
    "WITH CTE_site_history AS (\n",
    "    SELECT\n",
    "        fullVisitorId as visitor_id,\n",
    "        (SELECT MAX(IF(index = 10, value, NULL)) FROM UNNEST(hits.customDimensions)) AS content_id,\n",
    "        (SELECT MAX(IF(index = 7, value, NULL)) FROM UNNEST(hits.customDimensions)) AS category, \n",
    "        (SELECT MAX(IF(index = 6, value, NULL)) FROM UNNEST(hits.customDimensions)) AS title,\n",
    "        (SELECT MAX(IF(index = 2, value, NULL)) FROM UNNEST(hits.customDimensions)) AS author_list,\n",
    "        SPLIT(RPAD((SELECT MAX(IF(index = 4, value, NULL)) FROM UNNEST(hits.customDimensions)), 7), '.') AS year_month_array,\n",
    "        LEAD(hits.customDimensions, 1) OVER (PARTITION BY fullVisitorId ORDER BY hits.time ASC) AS nextCustomDimensions\n",
    "    FROM \n",
    "        `cloud-training-demos.GA360_test.ga_sessions_sample`,   \n",
    "         UNNEST(hits) AS hits\n",
    "    WHERE \n",
    "        # only include hits on pages\n",
    "        hits.type = \"PAGE\"\n",
    "        AND\n",
    "        fullVisitorId IS NOT NULL\n",
    "        AND\n",
    "        hits.time != 0\n",
    "        AND\n",
    "        hits.time IS NOT NULL\n",
    "        AND\n",
    "        (SELECT MAX(IF(index = 10, value, NULL)) FROM UNNEST(hits.customDimensions)) IS NOT NULL\n",
    "),\n",
    "CTE_training_dataset AS (\n",
    "    SELECT\n",
    "        (SELECT MAX(IF(index=10, value, NULL)) FROM UNNEST(nextCustomDimensions)) AS next_content_id,\n",
    "\n",
    "        visitor_id,\n",
    "        content_id,\n",
    "        category,\n",
    "        REGEXP_REPLACE(title, r\",\", \"\") AS title,\n",
    "        REGEXP_EXTRACT(author_list, r\"^[^,]+\") AS author,\n",
    "        DATE_DIFF(DATE(CAST(year_month_array[OFFSET(0)] AS INT64), CAST(year_month_array[OFFSET(1)] AS INT64), 1), DATE(1970, 1, 1), MONTH) AS months_since_epoch\n",
    "    FROM\n",
    "        CTE_site_history\n",
    "    WHERE\n",
    "        (SELECT MAX(IF(index=10, value, NULL)) FROM UNNEST(nextCustomDimensions)) IS NOT NULL)\n",
    "\n",
    "SELECT\n",
    "    CAST(next_content_id AS STRING) AS next_content_id,\n",
    "\n",
    "    CAST(training_dataset.visitor_id AS STRING) AS visitor_id,\n",
    "    CAST(training_dataset.content_id AS STRING) AS content_id,\n",
    "    CAST(IFNULL(category, 'None') AS STRING) AS category,\n",
    "    CONCAT(\"\\\\\"\", REPLACE(TRIM(CAST(IFNULL(title, 'None') AS STRING)), \"\\\\\"\",\"\"), \"\\\\\"\") AS title,\n",
    "    CAST(IFNULL(author, 'None') AS STRING) AS author,\n",
    "    CAST(months_since_epoch AS STRING) AS months_since_epoch,\n",
    "\n",
    "    IFNULL(user_factors._0, 0.0) AS user_factor_0,\n",
    "    IFNULL(user_factors._1, 0.0) AS user_factor_1,\n",
    "    IFNULL(user_factors._2, 0.0) AS user_factor_2,\n",
    "    IFNULL(user_factors._3, 0.0) AS user_factor_3,\n",
    "    IFNULL(user_factors._4, 0.0) AS user_factor_4,\n",
    "    IFNULL(user_factors._5, 0.0) AS user_factor_5,\n",
    "    IFNULL(user_factors._6, 0.0) AS user_factor_6,\n",
    "    IFNULL(user_factors._7, 0.0) AS user_factor_7,\n",
    "    IFNULL(user_factors._8, 0.0) AS user_factor_8,\n",
    "    IFNULL(user_factors._9, 0.0) AS user_factor_9,\n",
    "\n",
    "    IFNULL(item_factors._0, 0.0) AS item_factor_0,\n",
    "    IFNULL(item_factors._1, 0.0) AS item_factor_1,\n",
    "    IFNULL(item_factors._2, 0.0) AS item_factor_2,\n",
    "    IFNULL(item_factors._3, 0.0) AS item_factor_3,\n",
    "    IFNULL(item_factors._4, 0.0) AS item_factor_4,\n",
    "    IFNULL(item_factors._5, 0.0) AS item_factor_5,\n",
    "    IFNULL(item_factors._6, 0.0) AS item_factor_6,\n",
    "    IFNULL(item_factors._7, 0.0) AS item_factor_7,\n",
    "    IFNULL(item_factors._8, 0.0) AS item_factor_8,\n",
    "    IFNULL(item_factors._9, 0.0) AS item_factor_9,\n",
    "\n",
    "    FARM_FINGERPRINT(CONCAT(CAST(visitor_id AS STRING), CAST(content_id AS STRING))) AS hash_id\n",
    "FROM\n",
    "    CTE_training_dataset AS training_dataset\n",
    "LEFT JOIN\n",
    "    `cloud-training-demos.GA360_test.user_factors` AS user_factors\n",
    "        ON CAST(training_dataset.visitor_id AS FLOAT64) = CAST(user_factors.user_id AS FLOAT64)\n",
    "LEFT JOIN\n",
    "    `cloud-training-demos.GA360_test.item_factors` AS item_factors\n",
    "        ON CAST(training_dataset.content_id AS STRING) = CAST(item_factors.item_id AS STRING)\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Let's pull a sample of our data into a dataframe to see what it looks like."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>next_content_id</th>\n",
       "      <th>visitor_id</th>\n",
       "      <th>content_id</th>\n",
       "      <th>category</th>\n",
       "      <th>title</th>\n",
       "      <th>author</th>\n",
       "      <th>months_since_epoch</th>\n",
       "      <th>user_factor_0</th>\n",
       "      <th>user_factor_1</th>\n",
       "      <th>user_factor_2</th>\n",
       "      <th>...</th>\n",
       "      <th>item_factor_1</th>\n",
       "      <th>item_factor_2</th>\n",
       "      <th>item_factor_3</th>\n",
       "      <th>item_factor_4</th>\n",
       "      <th>item_factor_5</th>\n",
       "      <th>item_factor_6</th>\n",
       "      <th>item_factor_7</th>\n",
       "      <th>item_factor_8</th>\n",
       "      <th>item_factor_9</th>\n",
       "      <th>hash_id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>299954138</td>\n",
       "      <td>1000025265994336570</td>\n",
       "      <td>299816215</td>\n",
       "      <td>News</td>\n",
       "      <td>\"Fahnenskandal von Mailand: Die Austria zeigt ...</td>\n",
       "      <td>Alexander Strecha</td>\n",
       "      <td>574</td>\n",
       "      <td>0.000712</td>\n",
       "      <td>0.001678</td>\n",
       "      <td>-0.001852</td>\n",
       "      <td>...</td>\n",
       "      <td>-2.434206e-24</td>\n",
       "      <td>-3.109219e-24</td>\n",
       "      <td>-7.968385e-24</td>\n",
       "      <td>-1.769454e-24</td>\n",
       "      <td>-1.478184e-24</td>\n",
       "      <td>-1.331895e-24</td>\n",
       "      <td>4.589061e-24</td>\n",
       "      <td>-2.270997e-24</td>\n",
       "      <td>3.567726e-24</td>\n",
       "      <td>-5768114586991797349</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>299826775</td>\n",
       "      <td>1000163602560555666</td>\n",
       "      <td>299933565</td>\n",
       "      <td>News</td>\n",
       "      <td>\"Koalitionsverhandler vor Konsens bei Krankenk...</td>\n",
       "      <td>Peter Temel</td>\n",
       "      <td>574</td>\n",
       "      <td>-0.000023</td>\n",
       "      <td>0.000019</td>\n",
       "      <td>-0.000042</td>\n",
       "      <td>...</td>\n",
       "      <td>-1.474965e-16</td>\n",
       "      <td>6.380229e-16</td>\n",
       "      <td>1.089593e-15</td>\n",
       "      <td>2.664141e-16</td>\n",
       "      <td>-6.541727e-16</td>\n",
       "      <td>5.789462e-16</td>\n",
       "      <td>-1.160644e-15</td>\n",
       "      <td>3.606335e-16</td>\n",
       "      <td>-3.140273e-16</td>\n",
       "      <td>7572436456843040598</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>299918278</td>\n",
       "      <td>1000163602560555666</td>\n",
       "      <td>299826775</td>\n",
       "      <td>Lifestyle</td>\n",
       "      <td>\"Auf Bank ausgeruht: Pensionist muss Strafe za...</td>\n",
       "      <td>Marlene Patsalidis</td>\n",
       "      <td>574</td>\n",
       "      <td>-0.000023</td>\n",
       "      <td>0.000019</td>\n",
       "      <td>-0.000042</td>\n",
       "      <td>...</td>\n",
       "      <td>-4.174247e-04</td>\n",
       "      <td>5.184471e-04</td>\n",
       "      <td>1.980996e-03</td>\n",
       "      <td>-1.876339e-03</td>\n",
       "      <td>-2.035266e-03</td>\n",
       "      <td>-9.295577e-04</td>\n",
       "      <td>-2.669745e-03</td>\n",
       "      <td>1.398294e-03</td>\n",
       "      <td>8.318340e-06</td>\n",
       "      <td>4505774231869461664</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>299853016</td>\n",
       "      <td>1000163602560555666</td>\n",
       "      <td>299918278</td>\n",
       "      <td>News</td>\n",
       "      <td>\"Skipässe in Wintersport-Hochburgen massiv teu...</td>\n",
       "      <td>Stefan Hofer</td>\n",
       "      <td>574</td>\n",
       "      <td>-0.000023</td>\n",
       "      <td>0.000019</td>\n",
       "      <td>-0.000042</td>\n",
       "      <td>...</td>\n",
       "      <td>1.557606e-03</td>\n",
       "      <td>2.223454e-02</td>\n",
       "      <td>-2.036940e-02</td>\n",
       "      <td>1.540217e-02</td>\n",
       "      <td>2.021701e-02</td>\n",
       "      <td>2.157036e-02</td>\n",
       "      <td>4.977403e-04</td>\n",
       "      <td>-1.841626e-03</td>\n",
       "      <td>-3.476710e-03</td>\n",
       "      <td>-8810985193337367461</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>298888038</td>\n",
       "      <td>1000163602560555666</td>\n",
       "      <td>299853016</td>\n",
       "      <td>News</td>\n",
       "      <td>\"Schröcksnadel gegen Werdenigg: Keine Aussprache\"</td>\n",
       "      <td>None</td>\n",
       "      <td>574</td>\n",
       "      <td>-0.000023</td>\n",
       "      <td>0.000019</td>\n",
       "      <td>-0.000042</td>\n",
       "      <td>...</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>5575481432772243221</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 28 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "  next_content_id           visitor_id content_id   category  \\\n",
       "0       299954138  1000025265994336570  299816215       News   \n",
       "1       299826775  1000163602560555666  299933565       News   \n",
       "2       299918278  1000163602560555666  299826775  Lifestyle   \n",
       "3       299853016  1000163602560555666  299918278       News   \n",
       "4       298888038  1000163602560555666  299853016       News   \n",
       "\n",
       "                                               title              author  \\\n",
       "0  \"Fahnenskandal von Mailand: Die Austria zeigt ...   Alexander Strecha   \n",
       "1  \"Koalitionsverhandler vor Konsens bei Krankenk...         Peter Temel   \n",
       "2  \"Auf Bank ausgeruht: Pensionist muss Strafe za...  Marlene Patsalidis   \n",
       "3  \"Skipässe in Wintersport-Hochburgen massiv teu...        Stefan Hofer   \n",
       "4  \"Schröcksnadel gegen Werdenigg: Keine Aussprache\"                None   \n",
       "\n",
       "  months_since_epoch  user_factor_0  user_factor_1  user_factor_2  ...  \\\n",
       "0                574       0.000712       0.001678      -0.001852  ...   \n",
       "1                574      -0.000023       0.000019      -0.000042  ...   \n",
       "2                574      -0.000023       0.000019      -0.000042  ...   \n",
       "3                574      -0.000023       0.000019      -0.000042  ...   \n",
       "4                574      -0.000023       0.000019      -0.000042  ...   \n",
       "\n",
       "   item_factor_1  item_factor_2  item_factor_3  item_factor_4  item_factor_5  \\\n",
       "0  -2.434206e-24  -3.109219e-24  -7.968385e-24  -1.769454e-24  -1.478184e-24   \n",
       "1  -1.474965e-16   6.380229e-16   1.089593e-15   2.664141e-16  -6.541727e-16   \n",
       "2  -4.174247e-04   5.184471e-04   1.980996e-03  -1.876339e-03  -2.035266e-03   \n",
       "3   1.557606e-03   2.223454e-02  -2.036940e-02   1.540217e-02   2.021701e-02   \n",
       "4   0.000000e+00   0.000000e+00   0.000000e+00   0.000000e+00   0.000000e+00   \n",
       "\n",
       "   item_factor_6  item_factor_7  item_factor_8  item_factor_9  \\\n",
       "0  -1.331895e-24   4.589061e-24  -2.270997e-24   3.567726e-24   \n",
       "1   5.789462e-16  -1.160644e-15   3.606335e-16  -3.140273e-16   \n",
       "2  -9.295577e-04  -2.669745e-03   1.398294e-03   8.318340e-06   \n",
       "3   2.157036e-02   4.977403e-04  -1.841626e-03  -3.476710e-03   \n",
       "4   0.000000e+00   0.000000e+00   0.000000e+00   0.000000e+00   \n",
       "\n",
       "               hash_id  \n",
       "0 -5768114586991797349  \n",
       "1  7572436456843040598  \n",
       "2  4505774231869461664  \n",
       "3 -8810985193337367461  \n",
       "4  5575481432772243221  \n",
       "\n",
       "[5 rows x 28 columns]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from google.cloud import bigquery\n",
    "bq = bigquery.Client(project = PROJECT)\n",
    "df_hybrid_dataset = bq.query(query_hybrid_dataset + \"LIMIT 100\").to_dataframe()\n",
    "df_hybrid_dataset.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user_factor_0</th>\n",
       "      <th>user_factor_1</th>\n",
       "      <th>user_factor_2</th>\n",
       "      <th>user_factor_3</th>\n",
       "      <th>user_factor_4</th>\n",
       "      <th>user_factor_5</th>\n",
       "      <th>user_factor_6</th>\n",
       "      <th>user_factor_7</th>\n",
       "      <th>user_factor_8</th>\n",
       "      <th>user_factor_9</th>\n",
       "      <th>...</th>\n",
       "      <th>item_factor_1</th>\n",
       "      <th>item_factor_2</th>\n",
       "      <th>item_factor_3</th>\n",
       "      <th>item_factor_4</th>\n",
       "      <th>item_factor_5</th>\n",
       "      <th>item_factor_6</th>\n",
       "      <th>item_factor_7</th>\n",
       "      <th>item_factor_8</th>\n",
       "      <th>item_factor_9</th>\n",
       "      <th>hash_id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>count</th>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>100.000000</td>\n",
       "      <td>...</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "      <td>1.000000e+02</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>mean</th>\n",
       "      <td>0.000003</td>\n",
       "      <td>0.001737</td>\n",
       "      <td>-0.000583</td>\n",
       "      <td>-0.000835</td>\n",
       "      <td>0.000241</td>\n",
       "      <td>0.000465</td>\n",
       "      <td>1.240610e-03</td>\n",
       "      <td>-0.000273</td>\n",
       "      <td>0.000269</td>\n",
       "      <td>0.000153</td>\n",
       "      <td>...</td>\n",
       "      <td>6.560215e-01</td>\n",
       "      <td>-4.308458e-01</td>\n",
       "      <td>3.807160e-03</td>\n",
       "      <td>-2.353299e-01</td>\n",
       "      <td>3.869609e-01</td>\n",
       "      <td>-6.329494e-01</td>\n",
       "      <td>-1.234204e-01</td>\n",
       "      <td>5.202034e-01</td>\n",
       "      <td>-2.226345e-01</td>\n",
       "      <td>-8.484932e+17</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>std</th>\n",
       "      <td>0.001702</td>\n",
       "      <td>0.002274</td>\n",
       "      <td>0.001475</td>\n",
       "      <td>0.001685</td>\n",
       "      <td>0.001810</td>\n",
       "      <td>0.001654</td>\n",
       "      <td>2.443893e-03</td>\n",
       "      <td>0.001819</td>\n",
       "      <td>0.001391</td>\n",
       "      <td>0.001901</td>\n",
       "      <td>...</td>\n",
       "      <td>4.542447e+00</td>\n",
       "      <td>3.203811e+00</td>\n",
       "      <td>1.326512e-01</td>\n",
       "      <td>1.456650e+00</td>\n",
       "      <td>2.643587e+00</td>\n",
       "      <td>4.557982e+00</td>\n",
       "      <td>7.201253e-01</td>\n",
       "      <td>3.763805e+00</td>\n",
       "      <td>1.570324e+00</td>\n",
       "      <td>5.309422e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>min</th>\n",
       "      <td>-0.005761</td>\n",
       "      <td>-0.000195</td>\n",
       "      <td>-0.003718</td>\n",
       "      <td>-0.005138</td>\n",
       "      <td>-0.009186</td>\n",
       "      <td>-0.003106</td>\n",
       "      <td>-1.258821e-03</td>\n",
       "      <td>-0.005378</td>\n",
       "      <td>-0.002282</td>\n",
       "      <td>-0.006065</td>\n",
       "      <td>...</td>\n",
       "      <td>-9.316617e-03</td>\n",
       "      <td>-2.270137e+01</td>\n",
       "      <td>-6.738291e-01</td>\n",
       "      <td>-1.019035e+01</td>\n",
       "      <td>-2.655937e-01</td>\n",
       "      <td>-3.237039e+01</td>\n",
       "      <td>-4.917779e+00</td>\n",
       "      <td>-1.516911e+00</td>\n",
       "      <td>-1.115637e+01</td>\n",
       "      <td>-8.901887e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>25%</th>\n",
       "      <td>-0.000588</td>\n",
       "      <td>0.000020</td>\n",
       "      <td>-0.001785</td>\n",
       "      <td>-0.000553</td>\n",
       "      <td>-0.000412</td>\n",
       "      <td>-0.000133</td>\n",
       "      <td>-1.164185e-04</td>\n",
       "      <td>-0.000759</td>\n",
       "      <td>-0.000327</td>\n",
       "      <td>-0.000008</td>\n",
       "      <td>...</td>\n",
       "      <td>-1.474965e-16</td>\n",
       "      <td>-3.109219e-24</td>\n",
       "      <td>-1.572803e-15</td>\n",
       "      <td>-5.862354e-06</td>\n",
       "      <td>-7.421535e-07</td>\n",
       "      <td>-1.280417e-14</td>\n",
       "      <td>-6.903299e-04</td>\n",
       "      <td>-1.320764e-19</td>\n",
       "      <td>-3.410457e-16</td>\n",
       "      <td>-5.638007e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>50%</th>\n",
       "      <td>0.000000</td>\n",
       "      <td>0.001079</td>\n",
       "      <td>-0.000149</td>\n",
       "      <td>-0.000219</td>\n",
       "      <td>0.000105</td>\n",
       "      <td>0.000096</td>\n",
       "      <td>-3.539207e-07</td>\n",
       "      <td>-0.000214</td>\n",
       "      <td>0.000047</td>\n",
       "      <td>0.000011</td>\n",
       "      <td>...</td>\n",
       "      <td>-3.821710e-36</td>\n",
       "      <td>2.089385e-23</td>\n",
       "      <td>0.000000e+00</td>\n",
       "      <td>-1.769454e-24</td>\n",
       "      <td>-1.478184e-24</td>\n",
       "      <td>-4.871939e-37</td>\n",
       "      <td>-1.801567e-22</td>\n",
       "      <td>2.222428e-24</td>\n",
       "      <td>3.567726e-24</td>\n",
       "      <td>-1.995584e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>75%</th>\n",
       "      <td>0.000420</td>\n",
       "      <td>0.002404</td>\n",
       "      <td>0.000004</td>\n",
       "      <td>-0.000068</td>\n",
       "      <td>0.001072</td>\n",
       "      <td>0.001426</td>\n",
       "      <td>2.663703e-03</td>\n",
       "      <td>0.000534</td>\n",
       "      <td>0.000654</td>\n",
       "      <td>0.001292</td>\n",
       "      <td>...</td>\n",
       "      <td>5.076126e-06</td>\n",
       "      <td>6.361844e-06</td>\n",
       "      <td>1.089593e-15</td>\n",
       "      <td>1.561346e-20</td>\n",
       "      <td>1.101260e-15</td>\n",
       "      <td>1.879980e-13</td>\n",
       "      <td>4.589061e-24</td>\n",
       "      <td>1.807638e-06</td>\n",
       "      <td>4.006397e-06</td>\n",
       "      <td>3.609213e+18</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>max</th>\n",
       "      <td>0.005343</td>\n",
       "      <td>0.008446</td>\n",
       "      <td>0.005723</td>\n",
       "      <td>0.001581</td>\n",
       "      <td>0.003355</td>\n",
       "      <td>0.007909</td>\n",
       "      <td>1.273011e-02</td>\n",
       "      <td>0.002683</td>\n",
       "      <td>0.004856</td>\n",
       "      <td>0.006780</td>\n",
       "      <td>...</td>\n",
       "      <td>3.229185e+01</td>\n",
       "      <td>1.982741e+00</td>\n",
       "      <td>7.682434e-01</td>\n",
       "      <td>5.441631e-02</td>\n",
       "      <td>1.877586e+01</td>\n",
       "      <td>1.030793e+00</td>\n",
       "      <td>3.816447e-02</td>\n",
       "      <td>2.671273e+01</td>\n",
       "      <td>2.130968e-01</td>\n",
       "      <td>8.981579e+18</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>8 rows × 21 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "       user_factor_0  user_factor_1  user_factor_2  user_factor_3  \\\n",
       "count     100.000000     100.000000     100.000000     100.000000   \n",
       "mean        0.000003       0.001737      -0.000583      -0.000835   \n",
       "std         0.001702       0.002274       0.001475       0.001685   \n",
       "min        -0.005761      -0.000195      -0.003718      -0.005138   \n",
       "25%        -0.000588       0.000020      -0.001785      -0.000553   \n",
       "50%         0.000000       0.001079      -0.000149      -0.000219   \n",
       "75%         0.000420       0.002404       0.000004      -0.000068   \n",
       "max         0.005343       0.008446       0.005723       0.001581   \n",
       "\n",
       "       user_factor_4  user_factor_5  user_factor_6  user_factor_7  \\\n",
       "count     100.000000     100.000000   1.000000e+02     100.000000   \n",
       "mean        0.000241       0.000465   1.240610e-03      -0.000273   \n",
       "std         0.001810       0.001654   2.443893e-03       0.001819   \n",
       "min        -0.009186      -0.003106  -1.258821e-03      -0.005378   \n",
       "25%        -0.000412      -0.000133  -1.164185e-04      -0.000759   \n",
       "50%         0.000105       0.000096  -3.539207e-07      -0.000214   \n",
       "75%         0.001072       0.001426   2.663703e-03       0.000534   \n",
       "max         0.003355       0.007909   1.273011e-02       0.002683   \n",
       "\n",
       "       user_factor_8  user_factor_9  ...  item_factor_1  item_factor_2  \\\n",
       "count     100.000000     100.000000  ...   1.000000e+02   1.000000e+02   \n",
       "mean        0.000269       0.000153  ...   6.560215e-01  -4.308458e-01   \n",
       "std         0.001391       0.001901  ...   4.542447e+00   3.203811e+00   \n",
       "min        -0.002282      -0.006065  ...  -9.316617e-03  -2.270137e+01   \n",
       "25%        -0.000327      -0.000008  ...  -1.474965e-16  -3.109219e-24   \n",
       "50%         0.000047       0.000011  ...  -3.821710e-36   2.089385e-23   \n",
       "75%         0.000654       0.001292  ...   5.076126e-06   6.361844e-06   \n",
       "max         0.004856       0.006780  ...   3.229185e+01   1.982741e+00   \n",
       "\n",
       "       item_factor_3  item_factor_4  item_factor_5  item_factor_6  \\\n",
       "count   1.000000e+02   1.000000e+02   1.000000e+02   1.000000e+02   \n",
       "mean    3.807160e-03  -2.353299e-01   3.869609e-01  -6.329494e-01   \n",
       "std     1.326512e-01   1.456650e+00   2.643587e+00   4.557982e+00   \n",
       "min    -6.738291e-01  -1.019035e+01  -2.655937e-01  -3.237039e+01   \n",
       "25%    -1.572803e-15  -5.862354e-06  -7.421535e-07  -1.280417e-14   \n",
       "50%     0.000000e+00  -1.769454e-24  -1.478184e-24  -4.871939e-37   \n",
       "75%     1.089593e-15   1.561346e-20   1.101260e-15   1.879980e-13   \n",
       "max     7.682434e-01   5.441631e-02   1.877586e+01   1.030793e+00   \n",
       "\n",
       "       item_factor_7  item_factor_8  item_factor_9       hash_id  \n",
       "count   1.000000e+02   1.000000e+02   1.000000e+02  1.000000e+02  \n",
       "mean   -1.234204e-01   5.202034e-01  -2.226345e-01 -8.484932e+17  \n",
       "std     7.201253e-01   3.763805e+00   1.570324e+00  5.309422e+18  \n",
       "min    -4.917779e+00  -1.516911e+00  -1.115637e+01 -8.901887e+18  \n",
       "25%    -6.903299e-04  -1.320764e-19  -3.410457e-16 -5.638007e+18  \n",
       "50%    -1.801567e-22   2.222428e-24   3.567726e-24 -1.995584e+18  \n",
       "75%     4.589061e-24   1.807638e-06   4.006397e-06  3.609213e+18  \n",
       "max     3.816447e-02   2.671273e+01   2.130968e-01  8.981579e+18  \n",
       "\n",
       "[8 rows x 21 columns]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_hybrid_dataset.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Launching Dataflow job preprocess-hybrid-recommendation-features-190412-184419 ... hang on\n"
     ]
    }
   ],
   "source": [
    "import apache_beam as beam\n",
    "import datetime, os\n",
    "\n",
    "def to_csv(rowdict):\n",
    "    # Pull columns from BQ and create a line\n",
    "    import hashlib\n",
    "    import copy\n",
    "    CSV_COLUMNS = \"next_content_id,visitor_id,content_id,category,title,author,months_since_epoch\".split(\",\")\n",
    "    FACTOR_COLUMNS = [\"user_factor_{}\".format(i) for i in range(10)] + [\"item_factor_{}\".format(i) for i in range(10)]\n",
    "\n",
    "    # Write out rows for each input row for each column in rowdict\n",
    "    data = \",\".join([\"None\" if k not in rowdict else (rowdict[k].encode(\"utf-8\") if rowdict[k] is not None else \"None\") for k in CSV_COLUMNS])\n",
    "    data += \",\"\n",
    "    data += \",\".join([str(rowdict[k]) if k in rowdict else \"None\" for k in FACTOR_COLUMNS])\n",
    "    yield (\"{}\".format(data))\n",
    "  \n",
    "def preprocess(in_test_mode):\n",
    "    import shutil, os, subprocess\n",
    "    job_name = \"preprocess-hybrid-recommendation-features\" + \"-\" + datetime.datetime.now().strftime(\"%y%m%d-%H%M%S\")\n",
    "\n",
    "    if in_test_mode:\n",
    "        print(\"Launching local job ... hang on\")\n",
    "        OUTPUT_DIR = \"./preproc/features\"\n",
    "        shutil.rmtree(OUTPUT_DIR, ignore_errors=True)\n",
    "        os.makedirs(OUTPUT_DIR)\n",
    "    else:\n",
    "        print(\"Launching Dataflow job {} ... hang on\".format(job_name))\n",
    "        OUTPUT_DIR = \"gs://{0}/hybrid_recommendation/preproc/features/\".format(BUCKET)\n",
    "        try:\n",
    "            subprocess.check_call(\"gsutil -m rm -r {}\".format(OUTPUT_DIR).split())\n",
    "        except:\n",
    "            pass\n",
    "\n",
    "    options = {\n",
    "        \"staging_location\": os.path.join(OUTPUT_DIR, \"tmp\", \"staging\"),\n",
    "        \"temp_location\": os.path.join(OUTPUT_DIR, \"tmp\"),\n",
    "        \"job_name\": job_name,\n",
    "        \"project\": PROJECT,\n",
    "        \"teardown_policy\": \"TEARDOWN_ALWAYS\",\n",
    "        \"no_save_main_session\": True,\n",
    "        \"max_num_workers\": 6\n",
    "    }\n",
    "    opts = beam.pipeline.PipelineOptions(flags = [], **options)\n",
    "    if in_test_mode:\n",
    "        RUNNER = \"DirectRunner\"\n",
    "    else:\n",
    "        RUNNER = \"DataflowRunner\"\n",
    "    p = beam.Pipeline(RUNNER, options = opts)\n",
    "  \n",
    "    query = query_hybrid_dataset\n",
    "\n",
    "    if in_test_mode:\n",
    "        query = query + \" LIMIT 100\" \n",
    "\n",
    "    for step in [\"train\", \"eval\"]:\n",
    "        if step == \"train\":\n",
    "            selquery = \"SELECT * FROM ({}) WHERE ABS(MOD(hash_id, 10)) < 9\".format(query)\n",
    "        else:\n",
    "            selquery = \"SELECT * FROM ({}) WHERE ABS(MOD(hash_id, 10)) = 9\".format(query)\n",
    "\n",
    "        (p \n",
    "         | \"{}_read\".format(step) >> beam.io.Read(beam.io.BigQuerySource(query = selquery, use_standard_sql = True))\n",
    "         | \"{}_csv\".format(step) >> beam.FlatMap(to_csv)\n",
    "         | \"{}_out\".format(step) >> beam.io.Write(beam.io.WriteToText(os.path.join(OUTPUT_DIR, \"{}.csv\".format(step))))\n",
    "        )\n",
    "\n",
    "    job = p.run()\n",
    "    if in_test_mode:\n",
    "        job.wait_until_finish()\n",
    "        print(\"Done!\")\n",
    "    \n",
    "preprocess(in_test_mode = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Let's check our files to make sure everything went as expected"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "rm -rf features\n",
    "mkdir features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "!gsutil -m cp -r gs://{BUCKET}/hybrid_recommendation/preproc/features/*.csv* features/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "==> features/eval.csv-00000-of-00001 <==\n",
      "710535,951784927766849126,710535,News,\"Haus aus Marmor und Grabsteinen\",None,503,-0.00170100899413,0.00496714003384,0.0040482301265,0.000690933316946,3.52509268851e-05,-0.00172890012618,0.00153049221262,0.00100265210494,0.00228979066014,-0.00201142113656,-5.59943889043e-19,7.42678684608e-19,-1.3985523895e-19,3.42277049416e-19,1.11620765154e-18,2.17990091471e-18,-2.42801472173e-19,1.5953545546e-19,-1.10792405809e-18,-4.38625547901e-19\n",
      "299818044,6813364694829221327,711895,None,\"Impressum KURIER.at\",None,553,0.000617411220446,-0.000148811683175,-0.000547810224816,-0.000194783264305,-0.000416739669163,-1.85458611668e-05,-0.000259642780293,-0.000104108628875,-0.000167975216755,0.000182291900273,-6.17342042923,14.5652112961,17.5528583527,3.3229033947,-44.9284629822,29.9998893738,18.7066059113,-14.6920909882,-20.3173618317,-3.7755317688\n",
      "714241,8640555275627058154,711895,None,\"Impressum KURIER.at\",None,553,1.48057097249e-05,-1.99350433832e-05,-1.66832815012e-05,-7.71130453359e-06,-1.61893422046e-05,1.01407499642e-06,-8.83275151864e-06,-9.23535571928e-06,-1.45201977375e-06,2.7637850053e-06,-6.17342042923,14.5652112961,17.5528583527,3.3229033947,-44.9284629822,29.9998893738,18.7066059113,-14.6920909882,-20.3173618317,-3.7755317688\n",
      "\n",
      "==> features/train.csv-00000-of-00004 <==\n",
      "733180,4457498454949488072,709531,Lifestyle,\"Thunfisch um 566.000 Euro versteigert\",None,504,-0.00163161556702,0.00118563103024,-0.00206252280623,0.000174878616235,-0.000371386762708,8.65281617735e-05,-0.000263485824689,-0.000245551520493,-0.000876758887898,6.79314553054e-06,1.19178007307e-12,-1.84806575425e-12,-3.35243334829e-13,-1.08242776721e-12,-6.03147197985e-13,-1.35153433584e-12,-4.69212938902e-13,-1.2847801165e-12,-7.35558153015e-13,7.33802612857e-13\n",
      "299907275,5684576675894832060,709763,Lifestyle,\"Waris Dirie lässt die Hüllen fallen\",Christine Scharfetter,503,-0.00205975957215,0.00373387616128,0.00113246438559,-0.000466885016067,0.0015372610651,0.000233920087339,-0.00135274045169,0.0016922946088,0.000412207999034,0.00150463567115,0.000206881333725,2.95793543046e-05,-0.000242229856667,0.000110112487164,5.73702527618e-05,-0.00017226851196,0.000104810045741,-0.000244620197918,-4.43018470833e-05,1.5213982806e-05\n",
      "299826775,1136161616016806175,710022,News,\"Ein Haus mieten: 10 Fragen und Antworten\",Ursula Horvath,503,0.000131085354951,-0.00053922861116,-0.00105769280344,-0.00136430538259,-0.00015962823818,0.0014577775728,-0.00063491309993,-0.00256530218758,0.000615109165665,-0.000610241550021,5.99711598656e-21,-3.65828623058e-21,3.48770082798e-21,-4.89903735847e-21,2.13146753176e-21,-6.97418511884e-21,-4.76741668636e-21,-3.89970947445e-21,5.3616425589e-21,5.38330152391e-21\n",
      "\n",
      "==> features/train.csv-00001-of-00004 <==\n",
      "299848776,6263916486525708411,299816215,News,\"Fahnenskandal von Mailand: Die Austria zeigt Flagge\",Alexander Strecha,574,-0.00149928487372,0.0052962237969,-0.000465781748062,-0.000578803650569,-0.0023043777328,-0.00339295784943,0.00170589145273,0.00116466486361,-0.00106325943489,-0.00189968058839,1.91183576102e-24,-2.4342056474e-24,-3.10921873842e-24,-7.96838497783e-24,-1.7694536646e-24,-1.47818373585e-24,-1.33189549604e-24,4.5890606599e-24,-2.27099722864e-24,3.56772599578e-24\n",
      "299853016,6764738067233144695,299816215,News,\"Fahnenskandal von Mailand: Die Austria zeigt Flagge\",Alexander Strecha,574,-0.000273603363894,0.000381544639822,9.51748152147e-05,-0.00141440914012,0.00113855628297,0.0012986556394,0.00113242818043,-0.000180888702744,0.000338799087331,-0.000689418928232,1.91183576102e-24,-2.4342056474e-24,-3.10921873842e-24,-7.96838497783e-24,-1.7694536646e-24,-1.47818373585e-24,-1.33189549604e-24,4.5890606599e-24,-2.27099722864e-24,3.56772599578e-24\n",
      "299937546,7015635483731327405,299816215,News,\"Fahnenskandal von Mailand: Die Austria zeigt Flagge\",Alexander Strecha,574,0.000118457275676,0.000143930403283,-0.000417291623307,-0.000286566646537,6.25781685812e-05,-0.000413753732573,0.000216673768591,-0.000203418472665,0.000195310538402,-0.000217049237108,1.91183576102e-24,-2.4342056474e-24,-3.10921873842e-24,-7.96838497783e-24,-1.7694536646e-24,-1.47818373585e-24,-1.33189549604e-24,4.5890606599e-24,-2.27099722864e-24,3.56772599578e-24\n",
      "\n",
      "==> features/train.csv-00002-of-00004 <==\n",
      "299828023,5545423854390309177,299848776,News,\"Hauptversammlung: Standortbestimmung für Rapid\",Alexander Huber,574,-1.64460652741e-05,0.00020944190328,-0.000177512396476,-0.000143139332067,0.000116113253171,5.08286175318e-05,8.00898342277e-05,-0.000202351686312,-0.000146371210576,-2.5919995096e-05,0.0424923039973,0.00983688607812,0.0494144596159,0.00495202513412,-0.00325802061707,-0.0105106746778,0.0170645602047,-0.0371509939432,0.118978075683,-0.0505210384727\n",
      "178890709,7582231185927782147,299848776,News,\"Hauptversammlung: Standortbestimmung für Rapid\",Alexander Huber,574,-8.35954607783e-20,-3.79247366675e-21,-4.59933423212e-20,-6.89563608018e-20,5.26119859873e-20,-8.0288534861e-20,1.05168176057e-19,5.31674345375e-20,-5.33382570275e-20,2.43700833987e-20,0.0424923039973,0.00983688607812,0.0494144596159,0.00495202513412,-0.00325802061707,-0.0105106746778,0.0170645602047,-0.0371509939432,0.118978075683,-0.0505210384727\n",
      "299826767,3147713998677840015,299848776,News,\"Hauptversammlung: Standortbestimmung für Rapid\",Alexander Huber,574,-0.00307426741347,0.00557295093313,0.00169024511706,-0.000696842325851,0.00229441816919,0.000349134556018,-0.002019014908,0.00252581317909,0.000615236116573,0.00224572489969,0.0424923039973,0.00983688607812,0.0494144596159,0.00495202513412,-0.00325802061707,-0.0105106746778,0.0170645602047,-0.0371509939432,0.118978075683,-0.0505210384727\n",
      "\n",
      "==> features/train.csv-00003-of-00004 <==\n",
      "299853016,6597910477942388453,299865757,News,\"ÖVP und FPÖ wollen zurück zu alten Noten\",None,574,-0.00607592752203,0.00217528478242,0.00144824618474,-0.00160153268371,-0.00188675150275,-0.000632539915387,0.00381029210985,-0.00214082910679,-0.00258546159603,0.00640401616693,-5.64447035119e-13,3.0999890697e-13,1.90860876908e-13,6.17965043286e-13,-1.72767751711e-13,6.43886582507e-13,3.06376966982e-13,-9.22291350279e-14,-2.48576712599e-13,-8.8365670033e-13\n",
      "298279465,689091576699447944,299865757,News,\"ÖVP und FPÖ wollen zurück zu alten Noten\",None,574,3.50430054823e-05,0.00149336177856,-0.00212904182263,3.86200736102e-05,-0.000102000092738,-0.000301287305774,0.000507643271703,-0.00256853131577,0.000466994126327,0.000300052808598,-5.64447035119e-13,3.0999890697e-13,1.90860876908e-13,6.17965043286e-13,-1.72767751711e-13,6.43886582507e-13,3.06376966982e-13,-9.22291350279e-14,-2.48576712599e-13,-8.8365670033e-13\n",
      "299865757,6993015325648036337,299865757,News,\"ÖVP und FPÖ wollen zurück zu alten Noten\",None,574,0.00337698310614,2.49207896559e-05,-0.00269788387232,-0.00445433845744,-0.00294296420179,0.00487069506198,-0.00434142304584,-0.00508004473522,-0.000820307992399,-0.00324016646482,-5.64447035119e-13,3.0999890697e-13,1.90860876908e-13,6.17965043286e-13,-1.72767751711e-13,6.43886582507e-13,3.06376966982e-13,-9.22291350279e-14,-2.48576712599e-13,-8.8365670033e-13\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/envs/py2env/lib/python2.7/site-packages/simplejson/encoder.py:296: DeprecationWarning: Interpreting naive datetime as local 2018-10-09 05:16:20.893523. Please add timezone info to timestamps.\n",
      "  chunks = self.iterencode(o, _one_shot=True)\n"
     ]
    }
   ],
   "source": [
    "!head -3 features/*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "<h2> Create vocabularies using Dataflow </h2>\n",
    "\n",
    "Let's use Cloud Dataflow to read in the BigQuery data, do some preprocessing, and write it out as CSV files.\n",
    "\n",
    "Now we'll create our vocabulary files for our categorical features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "query_vocabularies = \"\"\"\n",
    "SELECT\n",
    "    CAST((SELECT MAX(IF(index = index_value, value, NULL)) FROM UNNEST(hits.customDimensions)) AS STRING) AS grouped_by\n",
    "FROM `cloud-training-demos.GA360_test.ga_sessions_sample`,\n",
    "    UNNEST(hits) AS hits\n",
    "WHERE\n",
    "    # only include hits on pages\n",
    "    hits.type = \"PAGE\"\n",
    "    AND (SELECT MAX(IF(index = index_value, value, NULL)) FROM UNNEST(hits.customDimensions)) IS NOT NULL\n",
    "GROUP BY\n",
    "    grouped_by\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "import apache_beam as beam\n",
    "import datetime, os\n",
    "\n",
    "def to_txt(rowdict):\n",
    "    # Pull columns from BQ and create a line\n",
    "\n",
    "    # Write out rows for each input row for grouped by column in rowdict\n",
    "    return \"{}\".format(rowdict[\"grouped_by\"].encode(\"utf-8\"))\n",
    "  \n",
    "def preprocess(in_test_mode):\n",
    "    import shutil, os, subprocess\n",
    "    job_name = \"preprocess-hybrid-recommendation-vocab-lists\" + \"-\" + datetime.datetime.now().strftime(\"%y%m%d-%H%M%S\")\n",
    "\n",
    "    if in_test_mode:\n",
    "        print(\"Launching local job ... hang on\")\n",
    "        OUTPUT_DIR = \"./preproc/vocabs\"\n",
    "        shutil.rmtree(OUTPUT_DIR, ignore_errors=True)\n",
    "        os.makedirs(OUTPUT_DIR)\n",
    "    else:\n",
    "        print(\"Launching Dataflow job {} ... hang on\".format(job_name))\n",
    "        OUTPUT_DIR = \"gs://{0}/hybrid_recommendation/preproc/vocabs/\".format(BUCKET)\n",
    "        try:\n",
    "            subprocess.check_call(\"gsutil -m rm -r {}\".format(OUTPUT_DIR).split())\n",
    "        except:\n",
    "            pass\n",
    "\n",
    "    options = {\n",
    "        \"staging_location\": os.path.join(OUTPUT_DIR, \"tmp\", \"staging\"),\n",
    "        \"temp_location\": os.path.join(OUTPUT_DIR, \"tmp\"),\n",
    "        \"job_name\": job_name,\n",
    "        \"project\": PROJECT,\n",
    "        \"teardown_policy\": \"TEARDOWN_ALWAYS\",\n",
    "        \"no_save_main_session\": True,\n",
    "        \"max_num_workers\": 6\n",
    "    }\n",
    "    opts = beam.pipeline.PipelineOptions(flags = [], **options)\n",
    "    if in_test_mode:\n",
    "        RUNNER = \"DirectRunner\"\n",
    "    else:\n",
    "        RUNNER = \"DataflowRunner\"\n",
    "\n",
    "    p = beam.Pipeline(RUNNER, options = opts)\n",
    "  \n",
    "    def vocab_list(index, name):\n",
    "        query = query_vocabularies.replace(\"index_value\", \"{}\".format(index))\n",
    "\n",
    "        (p \n",
    "         | \"{}_read\".format(name) >> beam.io.Read(beam.io.BigQuerySource(query = query, use_standard_sql = True))\n",
    "         | \"{}_txt\".format(name) >> beam.Map(to_txt)\n",
    "         | \"{}_out\".format(name) >> beam.io.Write(beam.io.WriteToText(os.path.join(OUTPUT_DIR, \"{0}_vocab.txt\".format(name))))\n",
    "        )\n",
    "\n",
    "    # Call vocab_list function for each\n",
    "    vocab_list(10, \"content_id\") # content_id\n",
    "    vocab_list(7, \"category\") # category\n",
    "    vocab_list(2, \"author\") # author\n",
    "  \n",
    "    job = p.run()\n",
    "    if in_test_mode:\n",
    "        job.wait_until_finish()\n",
    "        print(\"Done!\")\n",
    "    \n",
    "preprocess(in_test_mode = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Also get vocab counts from the length of the vocabularies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "import apache_beam as beam\n",
    "import datetime, os\n",
    "\n",
    "def count_to_txt(rowdict):\n",
    "    # Pull columns from BQ and create a line\n",
    "\n",
    "    # Write out count\n",
    "    return \"{}\".format(rowdict[\"count_number\"])\n",
    "  \n",
    "def mean_to_txt(rowdict):\n",
    "    # Pull columns from BQ and create a line\n",
    "\n",
    "    # Write out mean\n",
    "    return \"{}\".format(rowdict[\"mean_value\"])\n",
    "  \n",
    "def preprocess(in_test_mode):\n",
    "    import shutil, os, subprocess\n",
    "    job_name = \"preprocess-hybrid-recommendation-vocab-counts\" + \"-\" + datetime.datetime.now().strftime(\"%y%m%d-%H%M%S\")\n",
    "\n",
    "    if in_test_mode:\n",
    "        print(\"Launching local job ... hang on\")\n",
    "        OUTPUT_DIR = \"./preproc/vocab_counts\"\n",
    "        shutil.rmtree(OUTPUT_DIR, ignore_errors=True)\n",
    "        os.makedirs(OUTPUT_DIR)\n",
    "    else:\n",
    "        print(\"Launching Dataflow job {} ... hang on\".format(job_name))\n",
    "        OUTPUT_DIR = \"gs://{0}/hybrid_recommendation/preproc/vocab_counts/\".format(BUCKET)\n",
    "        try:\n",
    "            subprocess.check_call(\"gsutil -m rm -r {}\".format(OUTPUT_DIR).split())\n",
    "        except:\n",
    "            pass\n",
    "\n",
    "    options = {\n",
    "        \"staging_location\": os.path.join(OUTPUT_DIR, \"tmp\", \"staging\"),\n",
    "        \"temp_location\": os.path.join(OUTPUT_DIR, \"tmp\"),\n",
    "        \"job_name\": job_name,\n",
    "        \"project\": PROJECT,\n",
    "        \"teardown_policy\": \"TEARDOWN_ALWAYS\",\n",
    "        \"no_save_main_session\": True,\n",
    "        \"max_num_workers\": 6\n",
    "    }\n",
    "    opts = beam.pipeline.PipelineOptions(flags = [], **options)\n",
    "    if in_test_mode:\n",
    "        RUNNER = \"DirectRunner\"\n",
    "    else:\n",
    "        RUNNER = \"DataflowRunner\"\n",
    "\n",
    "    p = beam.Pipeline(RUNNER, options = opts)\n",
    "  \n",
    "    def vocab_count(index, column_name):\n",
    "        query = \"\"\"\n",
    "        SELECT\n",
    "          COUNT(*) AS count_number\n",
    "        FROM ({})\n",
    "        \"\"\".format(query_vocabularies.replace(\"index_value\", \"{}\".format(index)))\n",
    "\n",
    "        (p \n",
    "         | \"{}_read\".format(column_name) >> beam.io.Read(beam.io.BigQuerySource(query = query, use_standard_sql = True))\n",
    "         | \"{}_txt\".format(column_name) >> beam.Map(count_to_txt)\n",
    "         | \"{}_out\".format(column_name) >> beam.io.Write(beam.io.WriteToText(os.path.join(OUTPUT_DIR, \"{0}_vocab_count.txt\".format(column_name))))\n",
    "        )\n",
    "    \n",
    "    def global_column_mean(column_name):\n",
    "        query = \"\"\"\n",
    "        SELECT\n",
    "          AVG(CAST({1} AS FLOAT64)) AS mean_value\n",
    "        FROM ({0})\n",
    "        \"\"\".format(query_hybrid_dataset, column_name)\n",
    "\n",
    "        (p \n",
    "         | \"{}_read\".format(column_name) >> beam.io.Read(beam.io.BigQuerySource(query = query, use_standard_sql = True))\n",
    "         | \"{}_txt\".format(column_name) >> beam.Map(mean_to_txt)\n",
    "         | \"{}_out\".format(column_name) >> beam.io.Write(beam.io.WriteToText(os.path.join(OUTPUT_DIR, \"{0}_mean.txt\".format(column_name))))\n",
    "        )\n",
    "    \n",
    "    # Call vocab_count function for each column we want the vocabulary count for\n",
    "    vocab_count(10, \"content_id\") # content_id\n",
    "    vocab_count(7, \"category\") # category\n",
    "    vocab_count(2, \"author\") # author\n",
    "\n",
    "    # Call global_column_mean function for each column we want the mean for\n",
    "    global_column_mean(\"months_since_epoch\") # months_since_epoch\n",
    "  \n",
    "    job = p.run()\n",
    "    if in_test_mode:\n",
    "        job.wait_until_finish()\n",
    "        print(\"Done!\")\n",
    "    \n",
    "preprocess(in_test_mode = False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "deletable": true,
    "editable": true
   },
   "source": [
    "Let's check our files to make sure everything went as expected"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "rm -rf vocabs\n",
    "mkdir vocabs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "!gsutil -m cp -r gs://{BUCKET}/hybrid_recommendation/preproc/vocabs/*.txt* vocabs/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "==> vocabs/author_vocab.txt-00000-of-00001 <==\n",
      "Wolfgang Atzenhofer\n",
      "Stefan Hofer\n",
      "Bernhard Gaul, Christian Böhmer\n",
      "\n",
      "==> vocabs/category_vocab.txt-00000-of-00001 <==\n",
      "News\n",
      "Lifestyle\n",
      "Stars & Kultur\n",
      "\n",
      "==> vocabs/content_id_vocab.txt-00000-of-00001 <==\n",
      "299792293\n",
      "299965853\n",
      "299800661\n"
     ]
    }
   ],
   "source": [
    "!head -3 vocabs/*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "rm -rf vocab_counts\n",
    "mkdir vocab_counts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [],
   "source": [
    "!gsutil -m cp -r gs://{BUCKET}/hybrid_recommendation/preproc/vocab_counts/*.txt* vocab_counts/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true,
    "jupyter": {
     "outputs_hidden": false
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "==> vocab_counts/author_vocab_count.txt-00000-of-00001 <==\n",
      "1103\n",
      "\n",
      "==> vocab_counts/category_vocab_count.txt-00000-of-00001 <==\n",
      "3\n",
      "\n",
      "==> vocab_counts/content_id_vocab_count.txt-00000-of-00001 <==\n",
      "15634\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/envs/py2env/lib/python2.7/site-packages/simplejson/encoder.py:296: DeprecationWarning: Interpreting naive datetime as local 2018-10-08 04:49:28.672721. Please add timezone info to timestamps.\n",
      "  chunks = self.iterencode(o, _one_shot=True)\n"
     ]
    }
   ],
   "source": [
    "!head -3 vocab_counts/*"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}