ClickstreamEventsAnalysis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h1 align=\"center\">\n",
    "  <a href=\"https://escale.com.br/\"><img src=\"https://www.kaszek.com/wp-content/uploads/2018/04/logo-escale-black.png\" alt=\"Markdownify\" width=\"200\"></a>\n",
    "  <br>\n",
    "  Clickstream Events Analysis\n",
    "</h1>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h4 align=\"center\">An analytical approach to clickstream and sessions data using <a href=\"https://spark.apache.org/docs/latest/api/python/pyspark.html\" target=\"_blank\">Pyspark</a>.</h4>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Author\n",
    "\n",
    "2020 - Eduardo Trevisani"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Developed from case supplied by Escale, available on this [online document](https://escaletech.github.io/dataplatform/data-engineer-test)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Requirements\n",
    "---\n",
    "***Python >= 3.6***\n",
    "\n",
    "* PySpark >= 2.4.3 \n",
    "  - Using environment variable `export PYSPARK_PYTHON=${replace_for_your_python3_home_location}`\n",
    "* requests\n",
    "* shutils"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initial settings: clickstream data download\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import requests\n",
    "import shutil"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "files= [\"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00000.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00001.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00002.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00003.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00004.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00005.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00006.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00007.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00008.json.gz\",\n",
    "        \"https://d3l36jjwr70u5l.cloudfront.net/data-engineer-test/part-00009.json.gz\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Replace by a convenient file system path to store the downloaded files\n",
    "FILES_FOLDER_PATH = '/tmp/clickstream_data/'\n",
    "\n",
    "os.makedirs(FILES_FOLDER_PATH, exist_ok=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "for url in files:\n",
    "    filename = f'{FILES_FOLDER_PATH}{url.split(\"/\")[-1]}'\n",
    "\n",
    "    r = requests.get(url, stream=True)\n",
    "    \n",
    "    if r.status_code == 200:\n",
    "        with open(filename, 'wb') as f:\n",
    "            r.raw.decode_content = True  # just in case transport encoding was applied\n",
    "            shutil.copyfileobj(r.raw, f)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initial settings: Spark initialization\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### References\n",
    "* https://www.knowru.com/blog/2-tunings-you-should-make-spark-applications-running-emr/\n",
    "* https://stackoverflow.com/questions/39868263/spark-load-data-and-add-filename-as-dataframe-column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import input_file_name, split, reverse\n",
    "from operator import add\n",
    "\n",
    "# Using Kryo Serialize to enable a high performance on data serialization\n",
    "spark = SparkSession.builder \\\n",
    "                   .master(\"local\")\\\n",
    "                   .appName(\"ClickstreamEventsAnalysis\")\\\n",
    "                   .config(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\")\\\n",
    "                   .getOrCreate()\n",
    "\n",
    "df = spark.read.json(f\"{FILES_FOLDER_PATH}/*.json.gz\")\n",
    "df = df.withColumn(\"filename\", reverse(split(input_file_name(), '/'))[0])\n",
    "df = df.repartition(\"anonymous_id\").sortWithinPartitions(\"anonymous_id\", \"device_sent_timestamp\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 1: unique sessions analysis\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### References\n",
    "* https://spark.apache.org/docs/latest/api/python/pyspark.html?#pyspark.RDD.aggregateByKey\n",
    "* https://spark.apache.org/docs/latest/api/python/pyspark.html?#pyspark.RDD.reduceByKey\n",
    "* https://sparkbyexamples.com/pyspark/pyspark-reducebykey-usage-with-examples/\n",
    "* https://medium.com/@mukeshkumar_46704/apache-spark-rdd-api-using-pyspark-91a1f17507c\n",
    "* https://medium.com/@yesilliali/apache-spark-understanding-zerovalue-in-aggregatebykey-function-3d7df62567ae\n",
    "* https://b.tdhopper.com/blog/pysparks-aggregatebykey-method/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "kv_pairs = df.rdd.map(lambda x: (x.anonymous_id, (x[1:])))\n",
    "\n",
    "def seqFunc(x, y):\n",
    "    is_new_session = 1 if int(y[2]) >= x[0] + 1800000 else 0\n",
    "    return (y[2], y[8], x[2] + is_new_session)\n",
    "\n",
    "def combFunc(x, y):\n",
    "    last_timestamp = x[0] if x[0] > y[0] else y[0]\n",
    "    return (last_timestamp, x[1], x[2] + y[2])\n",
    "\n",
    "agg_kv_pairs = kv_pairs.aggregateByKey((0,'',0), seqFunc, combFunc)\n",
    "agg_filename_pairs = agg_kv_pairs.map(lambda x: (x[1][1], (x[1][2])))\n",
    "agg_unique_sessions = agg_filename_pairs.reduceByKey(add)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "             filename  unique_sessions\n",
      "7  part-00000.json.gz         10234861\n",
      "4  part-00001.json.gz         10230776\n",
      "0  part-00002.json.gz         10227177\n",
      "1  part-00003.json.gz         10233329\n",
      "3  part-00004.json.gz         10233306\n",
      "8  part-00005.json.gz         10235200\n",
      "2  part-00006.json.gz         10232658\n",
      "5  part-00007.json.gz         10236335\n",
      "6  part-00008.json.gz         10228150\n",
      "9  part-00009.json.gz         10229042\n"
     ]
    }
   ],
   "source": [
    "# Converting Spark dataframe to Pandas dataframe\n",
    "# Estimated time to completition: 35 min (local execution with dual-core i5 processor)\n",
    "\n",
    "pd_step_1 = agg_unique_sessions.toDF([\"filename\",\"unique_sessions\"])\\\n",
    "                               .toPandas()\\\n",
    "                               .sort_values('filename')\n",
    "print(pd_step_1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'part-00000.json.gz': 10234861, 'part-00001.json.gz': 10230776, 'part-00002.json.gz': 10227177, 'part-00003.json.gz': 10233329, 'part-00004.json.gz': 10233306, 'part-00005.json.gz': 10235200, 'part-00006.json.gz': 10232658, 'part-00007.json.gz': 10236335, 'part-00008.json.gz': 10228150, 'part-00009.json.gz': 10229042}\n"
     ]
    }
   ],
   "source": [
    "# Converting Pandas dataframe to JSON string\n",
    "result_step_1 = pd_step_1.sort_values('filename')\\\n",
    "                         .set_index('filename')\\\n",
    "                         .to_dict()['unique_sessions']\n",
    "print(result_step_1)"
   ]
  }
 ],
 "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}