dataset-preprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Outbrain click prediction data processing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "I downloaded the data files from https://www.kaggle.com/c/outbrain-click-prediction/data?select=page_views.csv.zip,\n",
    "\n",
    "and uploaded to our HDFS. using a pySpark kernel."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "first, let's load each CSV into a dataframe:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "events_df = spark.read.csv(\"/user/ykarni/kaggle/events.csv\", header=True)\n",
    "events_df = events_df.withColumn('timestamp', events_df['timestamp'].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['display_id', 'uuid', 'document_id', 'timestamp', 'platform', 'geo_location']"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "events_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------+-----------+---------+--------+------------+\n",
      "|display_id|          uuid|document_id|timestamp|platform|geo_location|\n",
      "+----------+--------------+-----------+---------+--------+------------+\n",
      "|         1|cb8c55702adb93|     379743|       61|       3|   US>SC>519|\n",
      "|         2|79a85fa78311b9|    1794259|       81|       2|   US>CA>807|\n",
      "|         3|822932ce3d8757|    1179111|      182|       2|   US>MI>505|\n",
      "|         4|85281d0a49f7ac|    1777797|      234|       2|   US>WV>564|\n",
      "|         5|8d0daef4bf5b56|     252458|      338|       2|       SG>00|\n",
      "+----------+--------------+-----------+---------+--------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "events_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "document_categories_df = spark.read.csv(\"/user/ykarni/kaggle/documents_categories.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['document_id', 'category_id', 'confidence_level']"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "document_categories_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "97"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "document_categories_df.select(\"category_id\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "promoted_content_df = spark.read.csv(\"/user/ykarni/kaggle/promoted_content.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ad_id', 'document_id', 'campaign_id', 'advertiser_id']"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "promoted_content_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "documents_meta_df = spark.read.csv(\"/user/ykarni/kaggle/documents_meta.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['document_id', 'source_id', 'publisher_id', 'publish_time']"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "documents_meta_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [],
   "source": [
    "clicks_train_df = spark.read.csv(\"/user/ykarni/kaggle/clicks_train.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['display_id', 'ad_id', 'clicked']"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clicks_train_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "87141731"
      ]
     },
     "execution_count": 51,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clicks_train_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "clicks_test_df = spark.read.csv(\"/user/ykarni/kaggle/clicks_test.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['display_id', 'ad_id']"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clicks_test_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "32225162"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "clicks_test_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "page_views_df = spark.read.csv(\"/user/ykarni/kaggle/page_views.csv\", header=True)\n",
    "page_views_df = page_views_df.withColumn('timestamp', page_views_df['timestamp'].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['uuid',\n",
       " 'document_id',\n",
       " 'timestamp',\n",
       " 'platform',\n",
       " 'geo_location',\n",
       " 'traffic_source']"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "page_views_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----------+---------+--------+------------+--------------+\n",
      "|          uuid|document_id|timestamp|platform|geo_location|traffic_source|\n",
      "+--------------+-----------+---------+--------+------------+--------------+\n",
      "|1fd5f051fba643|        120| 31905835|       1|          RS|             2|\n",
      "|8557aa9004be3b|        120| 32053104|       1|       VN>44|             2|\n",
      "|c351b277a358f0|        120| 54013023|       1|       KR>12|             1|\n",
      "|8205775c5387f9|        120| 44196592|       1|       IN>16|             2|\n",
      "|9cb0ccd8458371|        120| 65817371|       1|   US>CA>807|             2|\n",
      "+--------------+-----------+---------+--------+------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "page_views_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col, when"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "convert the 0/1 format to the FW label format (1/-1):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "clicks_train_df = clicks_train_df.withColumn(\"label\", when(col(\"clicked\") == \"0\",\"-1\").otherwise(1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "next, we construct a \"document-categories\" mapping:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [],
   "source": [
    "doc_categories_grouped = document_categories_df.filter(\"confidence_level > 0.9\") \\\n",
    "    .groupBy(\"document_id\") \\\n",
    "    .agg(f.concat_ws(\" \", f.collect_list(document_categories_df[\"category_id\"]))) \\\n",
    "    .withColumnRenamed(\"concat_ws( , collect_list(category_id))\", \"categories\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['document_id', 'categories']"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "doc_categories_grouped.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "gather available data for the ads in our inventory:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "ads = promoted_content_df.join(documents_meta_df, \"document_id\") \\\n",
    "    .join(doc_categories_grouped, \"document_id\", how='leftouter')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "ads = ads.withColumnRenamed(existing=\"document_id\", new=\"ad_document_id\") \\\n",
    "   .withColumnRenamed(\"source_id\", \"ad_source_id\") \\\n",
    "   .withColumnRenamed(\"publisher_id\", \"ad_publisher_id\") \\\n",
    "   .withColumnRenamed(\"publish_time\", \"ad_publish_time\") \\\n",
    "   .withColumnRenamed(\"categories\", \"ad_categories\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ad_document_id',\n",
       " 'ad_id',\n",
       " 'campaign_id',\n",
       " 'advertiser_id',\n",
       " 'ad_source_id',\n",
       " 'ad_publisher_id',\n",
       " 'ad_publish_time',\n",
       " 'ad_categories']"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ads.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "we'll use the page views data, to understand what interests specific users.\n",
    "\n",
    "first step - collect just the user id and document id pairs.\n",
    "\n",
    "second step - join with the document categories dataframe to get the user categories:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_docs_df = page_views_df.select(\"uuid\", \"document_id\").distinct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_categories = user_docs_df.join(document_categories_df.filter(\"confidence_level > 0.9\"), \"document_id\", how=\"leftouter\") \\\n",
    "    .select(\"uuid\", \"category_id\") \\\n",
    "    .distinct() \\\n",
    "    .groupBy(\"uuid\") \\\n",
    "    .agg(f.concat_ws(\" \", f.collect_list(document_categories_df[\"category_id\"]))) \\\n",
    "    .withColumnRenamed(\"concat_ws( , collect_list(category_id))\", \"user_categories\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['uuid', 'user_categories']"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "user_categories.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "we want to count user page views for every event, but need to make sure we don't just aggregate on the entire dataset - for every event in time T, we want to count all of the page views of the user occurring before time T:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "windowval = Window.partitionBy('uuid').orderBy('timestamp').rangeBetween(Window.unboundedPreceding, 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [],
   "source": [
    "page_views_df = page_views_df.withColumn('previous_pageviews_count', f.count('uuid').over(windowval))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "finally, we join everything into a single dataframe with all relevant data concerning each event where an ad was presented to a user:\n",
    "* the context data (which document the user was reading, the section and publisher, the content categories)\n",
    "* the ad information - which advertiser and campaign - and who is the content publisher and what section the content belongs to etc.\n",
    "* the user context - user id, where the user came from (search, social, internal traffic), the geolocation of the request etc.\n",
    "* the supervision - click/no click"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "labelled_events_full = clicks_train_df.join(events_df, \"display_id\") \\\n",
    "    .join(ads, \"ad_id\") \\\n",
    "    .join(documents_meta_df, \"document_id\") \\\n",
    "    .join(doc_categories_grouped, \"document_id\", how='leftouter') \\\n",
    "    .join(page_views_df, [\"uuid\", \"document_id\", \"timestamp\", \"platform\", \"geo_location\"], how=\"leftouter\") \\\n",
    "    .join(user_categories, \"uuid\", how=\"leftouter\") \\\n",
    "    .fillna(subset=['previous_pageviews_count'], value=0) \\\n",
    "    .na.fill(\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['uuid',\n",
       " 'document_id',\n",
       " 'timestamp',\n",
       " 'platform',\n",
       " 'geo_location',\n",
       " 'ad_id',\n",
       " 'display_id',\n",
       " 'clicked',\n",
       " 'label',\n",
       " 'ad_document_id',\n",
       " 'campaign_id',\n",
       " 'advertiser_id',\n",
       " 'ad_source_id',\n",
       " 'ad_publisher_id',\n",
       " 'ad_publish_time',\n",
       " 'ad_categories',\n",
       " 'source_id',\n",
       " 'publisher_id',\n",
       " 'publish_time',\n",
       " 'categories',\n",
       " 'traffic_source',\n",
       " 'previous_pageviews_count',\n",
       " 'user_categories']"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "labelled_events_full.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import concat, col, lit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "labelled_events_full.select( \\\n",
    "    concat(col(\"label\"), lit(\" \")), \\\n",
    "    concat(lit(\"A \"), col(\"uuid\"), lit(\" \")), \\\n",
    "    concat(lit(\"B \"), col(\"platform\"), lit(\" \")), \\\n",
    "    concat(lit(\"C \"), col(\"geo_location\"), lit(\" \")), \\\n",
    "    concat(lit(\"D \"), col(\"traffic_source\"), lit(\" \")), \\\n",
    "    concat(lit(\"E \"), col(\"document_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"F \"), col(\"source_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"G \"), col(\"publisher_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"H \"), col(\"categories\"), lit(\" \")), \\\n",
    "    concat(lit(\"I \"), col(\"ad_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"J \"), col(\"campaign_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"K \"), col(\"advertiser_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"L \"), col(\"ad_document_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"M \"), col(\"ad_source_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"N \"), col(\"ad_publisher_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"O \"), col(\"ad_categories\"), lit(\" \")), \\\n",
    "    concat(lit(\"P \"), col(\"user_categories\"), lit(\" \")), \\\n",
    "    concat(lit(\"Q \"), col(\"previous_pageviews_count\"), lit(\" \")),\n",
    "    concat(lit(\"R \"), col(\"display_id\"), lit(\" \"))) \\\n",
    "    .write.csv(\"/user/ykarni/kaggle/train\", header=False, sep=\"|\", quote=\"\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "and similarly we generate the test set for the ob kaggle submission:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_set_full = clicks_test_df.join(events_df, \"display_id\") \\\n",
    "    .join(ads, \"ad_id\") \\\n",
    "    .join(documents_meta_df, \"document_id\") \\\n",
    "    .join(doc_categories_grouped, \"document_id\", how='leftouter') \\\n",
    "    .join(page_views_df, [\"uuid\", \"document_id\", \"timestamp\", \"platform\", \"geo_location\"], how=\"leftouter\") \\\n",
    "    .join(user_categories, \"uuid\", how=\"leftouter\") \\\n",
    "    .fillna(subset=['previous_pageviews_count'], value=0) \\\n",
    "    .na.fill(\" \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['uuid',\n",
       " 'document_id',\n",
       " 'timestamp',\n",
       " 'platform',\n",
       " 'geo_location',\n",
       " 'ad_id',\n",
       " 'display_id',\n",
       " 'ad_document_id',\n",
       " 'campaign_id',\n",
       " 'advertiser_id',\n",
       " 'ad_source_id',\n",
       " 'ad_publisher_id',\n",
       " 'ad_publish_time',\n",
       " 'ad_categories',\n",
       " 'source_id',\n",
       " 'publisher_id',\n",
       " 'publish_time',\n",
       " 'categories',\n",
       " 'traffic_source',\n",
       " 'previous_pageviews_count',\n",
       " 'user_categories']"
      ]
     },
     "execution_count": 76,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_set_full.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_set_full.select(\\\n",
    "    concat(lit(\"A \"), col(\"uuid\"), lit(\" \")), \\\n",
    "    concat(lit(\"B \"), col(\"platform\"), lit(\" \")), \\\n",
    "    concat(lit(\"C \"), col(\"geo_location\"), lit(\" \")), \\\n",
    "    concat(lit(\"D \"), col(\"traffic_source\"), lit(\" \")), \\\n",
    "    concat(lit(\"E \"), col(\"document_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"F \"), col(\"source_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"G \"), col(\"publisher_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"H \"), col(\"categories\"), lit(\" \")), \\\n",
    "    concat(lit(\"I \"), col(\"ad_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"J \"), col(\"campaign_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"K \"), col(\"advertiser_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"L \"), col(\"ad_document_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"M \"), col(\"ad_source_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"N \"), col(\"ad_publisher_id\"), lit(\" \")), \\\n",
    "    concat(lit(\"O \"), col(\"ad_categories\"), lit(\" \")), \\\n",
    "    concat(lit(\"P \"), col(\"user_categories\"), lit(\" \")), \\\n",
    "    concat(lit(\"Q \"), col(\"previous_pageviews_count\"), lit(\" \")),\n",
    "    concat(lit(\"R \"), col(\"display_id\"), lit(\" \"))) \\\n",
    "    .write.csv(\"/user/ykarni/kaggle/test\", header=False, sep=\"|\", quote=\"\", mode=\"overwrite\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "once done, run the post_process_dataset.sh script to join the results to a single file, perform last minute cleanups, and do the train-dev split -\n",
    "CONGRATULATIONS, you're ready to go."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "pySpark (Spark 2.3.1)",
   "language": "python",
   "name": "pyspark"
  },
  "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.15+"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}