nar_module/scripts/dataproc_preprocessing/nar_preprocessing_addressa_01_dataproc.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setup (only required for the first run on the Spark cluster)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Collecting pandas\n",
      "\u001b[?25l  Downloading https://files.pythonhosted.org/packages/e6/de/a0d3defd8f338eaf53ef716e40ef6d6c277c35d50e09b586e170169cdf0d/pandas-0.24.1-cp36-cp36m-manylinux1_x86_64.whl (10.1MB)\n",
      "\u001b[K    100% |████████████████████████████████| 10.1MB 3.9MB/s eta 0:00:01\n",
      "\u001b[?25hRequirement already satisfied: numpy>=1.12.0 in /opt/conda/lib/python3.6/site-packages (from pandas) (1.16.2)\n",
      "Requirement already satisfied: pytz>=2011k in /opt/conda/lib/python3.6/site-packages (from pandas) (2018.9)\n",
      "Requirement already satisfied: python-dateutil>=2.5.0 in /opt/conda/lib/python3.6/site-packages (from pandas) (2.8.0)\n",
      "Requirement already satisfied: six>=1.5 in /opt/conda/lib/python3.6/site-packages (from python-dateutil>=2.5.0->pandas) (1.11.0)\n",
      "\u001b[31mpyspark 2.3.2 requires py4j==0.10.7, which is not installed.\u001b[0m\n",
      "Installing collected packages: pandas\n",
      "Successfully installed pandas-0.24.1\n",
      "\u001b[33mYou are using pip version 10.0.1, however version 19.0.3 is available.\n",
      "You should consider upgrading via the 'pip install --upgrade pip' command.\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "!pip install pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Copying gs://news_public_datasets4/adressa/data_transformed/adressa_articles.csv...\n",
      "- [1 files][ 60.3 MiB/ 60.3 MiB]                                                \n",
      "Operation completed over 1 objects/60.3 MiB.                                     \n"
     ]
    }
   ],
   "source": [
    "ROOT_PATH = 'gs://news_public_datasets4/adressa'\n",
    "#TODO: Upload this file (generated by the ACR module training) to GCS before calling spark script\n",
    "!gsutil cp {ROOT_PATH}/data_transformed/adressa_articles.csv ."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Loading dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import json\n",
    "import pandas as pd\n",
    "import pickle\n",
    "import datetime\n",
    "import hashlib\n",
    "import math\n",
    "import matplotlib\n",
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "import pyspark.sql.functions as F\n",
    "import pyspark.sql.types as T\n",
    "from pyspark.sql.functions import pandas_udf\n",
    "from pyspark.sql.functions import PandasUDFType"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Loading articles pre-processed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "articles_original_df = pd.read_csv('adressa_articles.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Index(['adressa-access', 'author_1st', 'category0', 'category1', 'category2',\n",
       "       'concepts', 'created_at_ts', 'entities', 'id', 'keywords', 'locations',\n",
       "       'persons', 'publishtime', 'site', 'text_highlights', 'url',\n",
       "       'id_encoded', 'category0_encoded', 'category1_encoded',\n",
       "       'author_encoded', 'keywords_encoded', 'concepts_encoded',\n",
       "       'entities_encoded', 'locations_encoded', 'persons_encoded'],\n",
       "      dtype='object')"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "articles_original_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "72932"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "articles_original_df['url'].nunique()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "72932"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "valid_articles_urls_to_ids_dict = dict(articles_original_df[['url','id_encoded']].apply(lambda x: (x['url'], x['id_encoded']), axis=1).values)\n",
    "len(valid_articles_urls_to_ids_dict)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Loading user interactions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Loading interaction files: ['gs://news_public_datasets4/adressa/three_month/20170101', 'gs://news_public_datasets4/adressa/three_month/20170102', 'gs://news_public_datasets4/adressa/three_month/20170103', 'gs://news_public_datasets4/adressa/three_month/20170104', 'gs://news_public_datasets4/adressa/three_month/20170105', 'gs://news_public_datasets4/adressa/three_month/20170106', 'gs://news_public_datasets4/adressa/three_month/20170107', 'gs://news_public_datasets4/adressa/three_month/20170108', 'gs://news_public_datasets4/adressa/three_month/20170109', 'gs://news_public_datasets4/adressa/three_month/20170110', 'gs://news_public_datasets4/adressa/three_month/20170111', 'gs://news_public_datasets4/adressa/three_month/20170112', 'gs://news_public_datasets4/adressa/three_month/20170113', 'gs://news_public_datasets4/adressa/three_month/20170114', 'gs://news_public_datasets4/adressa/three_month/20170115', 'gs://news_public_datasets4/adressa/three_month/20170116']\n"
     ]
    }
   ],
   "source": [
    "#INTERACTIONS_PATH = 'gs://news_public_datasets4/adressa/one_week/*'\n",
    "#INTERACTIONS_PATH = 'three_month/20170101'\n",
    "\n",
    "DAYS_TO_LOAD_INTERACTIONS=17\n",
    "interaction_json_files = [os.path.join(ROOT_PATH, 'three_month/201701{:02d}'.format(day)) for day in range(1, DAYS_TO_LOAD_INTERACTIONS)]\n",
    "print('Loading interaction files: {}'.format(interaction_json_files))\n",
    "\n",
    "interactions_df = spark.read \\\n",
    "  .option(\"mode\", \"PERMISSIVE\") \\\n",
    "  .json(interaction_json_files)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- activeTime: long (nullable = true)\n",
      " |-- author: string (nullable = true)\n",
      " |-- canonicalUrl: string (nullable = true)\n",
      " |-- category1: string (nullable = true)\n",
      " |-- city: string (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      " |-- deviceType: string (nullable = true)\n",
      " |-- eventId: long (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- keywords: string (nullable = true)\n",
      " |-- os: string (nullable = true)\n",
      " |-- profile: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- groups: array (nullable = true)\n",
      " |    |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |    |-- count: long (nullable = true)\n",
      " |    |    |    |    |-- group: string (nullable = true)\n",
      " |    |    |    |    |-- weight: double (nullable = true)\n",
      " |    |    |-- item: string (nullable = true)\n",
      " |-- publishtime: string (nullable = true)\n",
      " |-- query: string (nullable = true)\n",
      " |-- referrerHostClass: string (nullable = true)\n",
      " |-- referrerQuery: string (nullable = true)\n",
      " |-- referrerSearchEngine: string (nullable = true)\n",
      " |-- referrerSocialNetwork: string (nullable = true)\n",
      " |-- referrerUrl: string (nullable = true)\n",
      " |-- region: string (nullable = true)\n",
      " |-- sessionStart: boolean (nullable = true)\n",
      " |-- sessionStop: boolean (nullable = true)\n",
      " |-- time: long (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- url: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "interactions_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "21134428"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "interactions_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Retrives article id from its cannonical URL (because sometimes article ids in interactions do no match with articles tables, but cannonical URL do)\n",
    "def get_article_id_encoded_from_url(canonical_url):\n",
    "    if canonical_url in valid_articles_urls_to_ids_dict:\n",
    "        return valid_articles_urls_to_ids_dict[canonical_url]    \n",
    "    return None\n",
    "\n",
    "get_article_id_encoded_from_url_udf = F.udf(get_article_id_encoded_from_url, pyspark.sql.types.IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Filtering only interactions whose url/id is available in the articles table\n",
    "interactions_article_id_encoded_df = interactions_df.withColumn('article_id', get_article_id_encoded_from_url_udf(interactions_df['canonicalUrl']))\n",
    "interactions_filtered_df = interactions_article_id_encoded_df.filter(interactions_article_id_encoded_df['article_id'].isNull() == False).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5483902"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Valid interactions\n",
    "interactions_filtered_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "26873"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Distinct items count\n",
    "interactions_filtered_df.select('article_id').distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1483225202000"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "first_timestamp_ts = interactions_filtered_df.select('time').agg(F.min('time')).collect()[0][0] * 1000\n",
    "first_timestamp_ts"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Analyzing elapsed time since publishing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "#interactions_filtered_df.filter(interactions_filtered_df['time'].isNull()).count()\n",
    "#0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_timestamp_from_date_str(value):\n",
    "    if value is not None:\n",
    "        return int(datetime.datetime.strptime(value, '%Y-%m-%dT%H:%M:%S.%fZ').timestamp())\n",
    "    return None\n",
    "\n",
    "get_timestamp_from_date_str_udf = F.udf(get_timestamp_from_date_str, pyspark.sql.types.IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "interactions_filtered_with_publish_ts_df = interactions_filtered_df.withColumn('publish_ts', get_timestamp_from_date_str_udf(interactions_filtered_df['publishtime']))\n",
    "interactions_filtered_with_publish_ts_df = interactions_filtered_with_publish_ts_df.withColumn('elapsed_min_since_published', ((F.col('time') - F.col('publish_ts')) / 60).cast(pyspark.sql.types.IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "#interactions_filtered_with_publish_ts_df.select('publishtime','publish_ts', 'time', 'elapsed_min_since_published').show(100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 40 ms, sys: 12 ms, total: 52 ms\n",
      "Wall time: 26.5 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[49.0, 106.0, 314.0, 984.0, 4145.0]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "interactions_filtered_with_publish_ts_df.approxQuantile(\"elapsed_min_since_published\", [0.10, 0.25, 0.50, 0.75, 0.90], 0.01)\n",
    "#[49.0, 108.0, 334.0, 1020.0, 4611.0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "45551\n"
     ]
    },
    {
     "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>elapsed_min_since_published</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>count</th>\n",
       "      <td>5.438351e+06</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>mean</th>\n",
       "      <td>7.815292e+04</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>std</th>\n",
       "      <td>5.645131e+05</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>min</th>\n",
       "      <td>-3.151590e+05</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>25%</th>\n",
       "      <td>9.800000e+01</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>50%</th>\n",
       "      <td>2.940000e+02</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>75%</th>\n",
       "      <td>9.400000e+02</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>max</th>\n",
       "      <td>1.144721e+07</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       elapsed_min_since_published\n",
       "count                 5.438351e+06\n",
       "mean                  7.815292e+04\n",
       "std                   5.645131e+05\n",
       "min                  -3.151590e+05\n",
       "25%                   9.800000e+01\n",
       "50%                   2.940000e+02\n",
       "75%                   9.400000e+02\n",
       "max                   1.144721e+07"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "elapsed_min_since_published_df = interactions_filtered_with_publish_ts_df.select('elapsed_min_since_published').toPandas()\n",
    "print(len(elapsed_min_since_published_df[pd.isnull(elapsed_min_since_published_df['elapsed_min_since_published'])]))\n",
    "elapsed_min_since_published_df.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'\\nelapsed_min_since_published\\ncount\\t2.600818e+06\\nmean\\t6.438622e+04\\nstd\\t5.051825e+05\\nmin\\t-3.151590e+05\\n25%\\t9.400000e+01\\n50%\\t2.580000e+02\\n75%\\t8.370000e+02\\nmax\\t8.608278e+06\\n'"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "'''\n",
    "elapsed_min_since_published\n",
    "count\t2.600818e+06\n",
    "mean\t6.438622e+04\n",
    "std\t5.051825e+05\n",
    "min\t-3.151590e+05\n",
    "25%\t9.400000e+01\n",
    "50%\t2.580000e+02\n",
    "75%\t8.370000e+02\n",
    "max\t8.608278e+06\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Analyzing clicks by article distribution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "#clicks_by_article_count_df = interactions_filtered_df.groupBy('article_id').count()\n",
    "#clicks_by_article_count_df.approxQuantile(\"count\", [0.01, 0.10, 0.25, 0.50, 0.75, 0.90, 0.99], 0.01)\n",
    "#[1.0, 1.0, 1.0, 1.0, 2.0, 6.0, 33581.0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Processing categorical features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_categ_features_counts_dataframe(interactions_spark_df,column_name):\n",
    "    df_pandas = interactions_spark_df.groupBy(column_name).count().toPandas().sort_values('count', ascending=False)\n",
    "    return df_pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "PAD_TOKEN = '<PAD>'\n",
    "UNFREQ_TOKEN = '<UNF>'\n",
    "\n",
    "def get_encoder_for_values(values):\n",
    "    encoder_values = [PAD_TOKEN, UNFREQ_TOKEN] + values\n",
    "    encoder_ids = list(range(len(encoder_values)))\n",
    "    encoder_dict = dict(zip(encoder_values, encoder_ids))\n",
    "    return encoder_dict\n",
    "\n",
    "def get_categ_features_encoder_dict(counts_df, min_freq=100):\n",
    "    freq_values = counts_df[counts_df['count'] >= 100][counts_df.columns[0]].values.tolist()\n",
    "    encoder_dict = get_encoder_for_values(freq_values)\n",
    "    return encoder_dict\n",
    "\n",
    "def encode_cat_feature(value, encoder_dict):\n",
    "    if value in encoder_dict:\n",
    "        return encoder_dict[value]\n",
    "    else:\n",
    "        return encoder_dict[UNFREQ_TOKEN]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "178"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "countries_df = get_categ_features_counts_dataframe(interactions_filtered_df, 'country')\n",
    "len(countries_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "70"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "countries_encoder_dict = get_categ_features_encoder_dict(countries_df)\n",
    "len(countries_encoder_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "7002"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cities_df = get_categ_features_counts_dataframe(interactions_filtered_df, 'city')\n",
    "len(cities_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1022"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cities_encoder_dict = get_categ_features_encoder_dict(cities_df)\n",
    "len(cities_encoder_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1353"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "regions_df = get_categ_features_counts_dataframe(interactions_filtered_df, 'region')\n",
    "len(regions_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "237"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "regions_encoder_dict = get_categ_features_encoder_dict(regions_df)\n",
    "len(regions_encoder_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3\n"
     ]
    },
    {
     "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>deviceType</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Mobile</td>\n",
       "      <td>2667039</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Desktop</td>\n",
       "      <td>1716787</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Tablet</td>\n",
       "      <td>1100076</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  deviceType    count\n",
       "0     Mobile  2667039\n",
       "2    Desktop  1716787\n",
       "1     Tablet  1100076"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "devices_df = get_categ_features_counts_dataframe(interactions_filtered_df, 'deviceType')\n",
    "print(len(devices_df))\n",
    "devices_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "devices_encoder_dict = get_categ_features_encoder_dict(devices_df)\n",
    "len(devices_encoder_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "11\n"
     ]
    },
    {
     "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>os</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>iPhone OS</td>\n",
       "      <td>2350549</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>Windows</td>\n",
       "      <td>1460956</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>Android</td>\n",
       "      <td>1395958</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>Macintosh</td>\n",
       "      <td>235810</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>Windows Phone OS</td>\n",
       "      <td>18670</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>Linux</td>\n",
       "      <td>15296</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>Unknown</td>\n",
       "      <td>6309</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>BlackBerry</td>\n",
       "      <td>288</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Symbian OS</td>\n",
       "      <td>42</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>BSD</td>\n",
       "      <td>14</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>SunOS</td>\n",
       "      <td>10</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                  os    count\n",
       "0          iPhone OS  2350549\n",
       "10           Windows  1460956\n",
       "8            Android  1395958\n",
       "3          Macintosh   235810\n",
       "2   Windows Phone OS    18670\n",
       "5              Linux    15296\n",
       "7            Unknown     6309\n",
       "4         BlackBerry      288\n",
       "1         Symbian OS       42\n",
       "9                BSD       14\n",
       "6              SunOS       10"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "os_df = get_categ_features_counts_dataframe(interactions_filtered_df, 'os')\n",
    "print(len(os_df))\n",
    "os_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 36,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "os_encoder_dict = get_categ_features_encoder_dict(os_df)\n",
    "len(os_encoder_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5\n"
     ]
    },
    {
     "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>referrerHostClass</th>\n",
       "      <th>count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>internal</td>\n",
       "      <td>3259105</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>search</td>\n",
       "      <td>776911</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>direct</td>\n",
       "      <td>772549</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>social</td>\n",
       "      <td>421781</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>other</td>\n",
       "      <td>253556</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  referrerHostClass    count\n",
       "2          internal  3259105\n",
       "4            search   776911\n",
       "0            direct   772549\n",
       "3            social   421781\n",
       "1             other   253556"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "referrer_class_df = get_categ_features_counts_dataframe(interactions_filtered_df, 'referrerHostClass')\n",
    "print(len(referrer_class_df))\n",
    "referrer_class_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "7"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "referrer_class_encoder_dict = get_categ_features_encoder_dict(referrer_class_df)\n",
    "len(referrer_class_encoder_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "encoders_dict = {\n",
    "    'city': cities_encoder_dict,\n",
    "    'region': regions_encoder_dict,\n",
    "    'country': countries_encoder_dict,\n",
    "    'os': os_encoder_dict,\n",
    "    'device': devices_encoder_dict,\n",
    "    'referrer_class': referrer_class_encoder_dict\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Processing numeric features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[8.0, 21.0, 47.0, 90.0, 158.0]\n",
      "CPU times: user 36 ms, sys: 16 ms, total: 52 ms\n",
      "Wall time: 38.2 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "active_time_quantiles = interactions_filtered_df.approxQuantile(\"activeTime\", [0.10, 0.25, 0.50, 0.75, 0.90], 0.01)\n",
    "print(active_time_quantiles)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "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>summary</th>\n",
       "      <th>activeTime</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>count</td>\n",
       "      <td>2320994</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>mean</td>\n",
       "      <td>65.00062602488417</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>stddev</td>\n",
       "      <td>69.37057327780722</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>min</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>max</td>\n",
       "      <td>899</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  summary         activeTime\n",
       "0   count            2320994\n",
       "1    mean  65.00062602488417\n",
       "2  stddev  69.37057327780722\n",
       "3     min                  1\n",
       "4     max                899"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "active_time_stats_df = interactions_filtered_df.describe('activeTime').toPandas()\n",
    "active_time_stats_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "active_time_mean = float(active_time_stats_df[active_time_stats_df['summary'] == 'mean']['activeTime'].values[0])\n",
    "active_time_stddev = float(active_time_stats_df[active_time_stats_df['summary'] == 'stddev']['activeTime'].values[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Splitting sessions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'\\nschema = T.StructType([\\n    T.StructField(\"userId\", T.StringType()),\\n    T.StructField(\"min_ts\", T.IntegerType())\\n])\\n\\n@pandas_udf(schema, functionType=PandasUDFType.GROUPED_MAP)\\ndef split_sessions(df):\\n\\n    result_df = df[[\\'userId\\']]\\n    result_df[\\'min_ts\\'] = df[\\'time\\'].min()\\n    \\n    return result\\n\\n%%time\\ntmp = interactions_filtered_df.groupBy(\\'userId\\').apply(split_sessions)\\ntmp.show(100)\\n'"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "'''\n",
    "schema = T.StructType([\n",
    "    T.StructField(\"userId\", T.StringType()),\n",
    "    T.StructField(\"min_ts\", T.IntegerType())\n",
    "])\n",
    "\n",
    "@pandas_udf(schema, functionType=PandasUDFType.GROUPED_MAP)\n",
    "def split_sessions(df):\n",
    "\n",
    "    result_df = df[['userId']]\n",
    "    result_df['min_ts'] = df['time'].min()\n",
    "    \n",
    "    return result\n",
    "\n",
    "%%time\n",
    "tmp = interactions_filtered_df.groupBy('userId').apply(split_sessions)\n",
    "tmp.show(100)\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "def hash_str_to_int(encoded_bytes_text, digits):\n",
    "    return int(str(int(hashlib.md5(encoded_bytes_text).hexdigest()[:8], 16))[:digits])      "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "MAX_SESSION_IDLE_TIME_MS = 30 * 60 * 1000    #30 min\n",
    "\n",
    "def close_session(session):\n",
    "    size = len(session)\n",
    "    \n",
    "    #Creating and artificial session id based on the first click timestamp and a hash of user id\n",
    "    first_click = session[0]\n",
    "    session_id = (int(first_click['timestamp']) * 100) + hash_str_to_int(first_click['user_id'].encode(), 3)\n",
    "    session_hour = int((first_click['timestamp'] - first_timestamp_ts) / (1000 * 60 * 60)) #Converting timestamp to hours since first timestamp\n",
    "    \n",
    "    #Converting to Spark DataFrame Rows, to convert RDD back to DataFrame\n",
    "    clicks = list([T.Row(**click) for click in session])\n",
    "    session_dict = {'session_id': session_id,\n",
    "                    'session_hour': session_hour,\n",
    "                    'session_size': size,\n",
    "                    'session_start': first_click['timestamp'],\n",
    "                    'user_id': first_click['user_id'],\n",
    "                    'clicks': clicks \n",
    "                   }\n",
    "    session_row = T.Row(**session_dict)\n",
    "    \n",
    "    return session_row\n",
    "        \n",
    "def transform_interaction(interaction):        \n",
    "    return {\n",
    "            'article_id': interaction['article_id'],\n",
    "            'url': interaction['canonicalUrl'],\n",
    "            'user_id': interaction['userId'],\n",
    "            'timestamp': interaction['time'] * 1000, #converting to timestamp\n",
    "            'active_time_secs': interaction['activeTime'],\n",
    "            'country': encode_cat_feature(interaction['country'], encoders_dict['country']),\n",
    "            'region': encode_cat_feature(interaction['region'], encoders_dict['region']),\n",
    "            'city': encode_cat_feature(interaction['city'], encoders_dict['city']),\n",
    "            'os': encode_cat_feature(interaction['os'], encoders_dict['os']),\n",
    "            'device': encode_cat_feature(interaction['deviceType'], encoders_dict['device']),\n",
    "            'referrer_class': encode_cat_feature(interaction['referrerHostClass'], encoders_dict['referrer_class']),\n",
    "           }\n",
    "\n",
    "def split_sessions(group):\n",
    "    user, interactions = group\n",
    "    #Ensuring items are sorted by time\n",
    "    interactions_sorted_by_time = sorted(interactions, key=lambda x: x['time'])\n",
    "    #Transforming interactions\n",
    "    interactions_transformed = list(map(transform_interaction, interactions_sorted_by_time))\n",
    "\n",
    "    \n",
    "    sessions = []\n",
    "    session = []        \n",
    "    first_timestamp = interactions_transformed[0]['timestamp']\n",
    "    last_timestamp = first_timestamp    \n",
    "    for interaction in interactions_transformed:\n",
    "        \n",
    "        delta_ms = (interaction['timestamp'] - last_timestamp)\n",
    "        interaction['_elapsed_ms_since_last_click'] = delta_ms \n",
    "\n",
    "        if delta_ms <= MAX_SESSION_IDLE_TIME_MS:    \n",
    "            #Ignoring repeated items in session\n",
    "            if len(list(filter(lambda x: x['article_id'] == interaction['article_id'], session))) == 0:        \n",
    "                session.append(interaction)            \n",
    "        else:\n",
    "            #If session have at least 2 clicks (minimum for next click predicition)\n",
    "            if len(session) >= 2:\n",
    "                session_row = close_session(session)\n",
    "                sessions.append(session_row)                \n",
    "            session = [interaction]\n",
    "\n",
    "        last_timestamp = interaction['timestamp']\n",
    "            \n",
    "    if len(session) >= 2:\n",
    "        session_row = close_session(session)\n",
    "        sessions.append(session_row)\n",
    "        \n",
    "    #if len(sessions) > 1:\n",
    "    #    raise Exception('USER with more than one session: {}'.format(user))\n",
    "    \n",
    "    return list(zip(map(lambda x: x['session_id'], sessions), \n",
    "                    sessions))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"\\n#To debug\\n%%time\\nsessions_rdd = interactions_filtered_df.limit(1000).rdd.map(lambda x: (x['userId'], x)).groupByKey()                     .collect()\\n\\nfor row in sessions_rdd:\\n    print(split_sessions(row))\\n    print()\\n\""
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "'''\n",
    "#To debug\n",
    "%%time\n",
    "sessions_rdd = interactions_filtered_df.limit(1000).rdd.map(lambda x: (x['userId'], x)).groupByKey() \\\n",
    "                    .collect()\n",
    "\n",
    "for row in sessions_rdd:\n",
    "    print(split_sessions(row))\n",
    "    print()\n",
    "'''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 304 ms, sys: 84 ms, total: 388 ms\n",
      "Wall time: 27min 59s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "sessions_rdd = interactions_filtered_df.rdd.map(lambda x: (x['userId'], x)).groupByKey() \\\n",
    "                            .flatMap(split_sessions) \\\n",
    "                            .sortByKey() \\\n",
    "                            .map(lambda x: x[1])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Exporting sessions to JSON lines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [],
   "source": [
    "sessions_sdf = sessions_rdd.toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 56 ms, sys: 12 ms, total: 68 ms\n",
      "Wall time: 3min 49s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "sessions_sdf.write.partitionBy(\"session_hour\").json(os.path.join(ROOT_PATH,\"data_transformed/sessions_processed_by_spark/\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "982210"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sessions_sdf.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "def serialize(filename, obj):\n",
    "    with open(filename, 'wb') as handle:\n",
    "        pickle.dump(obj, handle)#, protocol=pickle.HIGHEST_PROTOCOL)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "NAR_ENCODERS_PATH  = 'nar_encoders_adressa.pickle'\n",
    "serialize(NAR_ENCODERS_PATH, encoders_dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Copying file://nar_encoders_adressa.pickle [Content-Type=application/octet-stream]...\n",
      "/ [1 files][ 26.5 KiB/ 26.5 KiB]                                                \n",
      "Operation completed over 1 objects/26.5 KiB.                                     \n"
     ]
    }
   ],
   "source": [
    "!gsutil cp {NAR_ENCODERS_PATH} {ROOT_PATH}/data_transformed/pickles/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}