old/spark-hw1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Checklist:\n",
    "# AWS emr-5.29.0\n",
    "# MASTER r5d.8xlarge 1x, no EBS\n",
    "# CORE r5d.8xlarge 4x, no EBS\n",
    "# Custom bootstrap action: s3://ydatazian/bootstrap.sh\n",
    "# Allow ssh in master node security group"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import tqdm.notebook as tqdm\n",
    "import numpy as np\n",
    "import scipy\n",
    "import sklearn\n",
    "import matplotlib.pyplot as plt"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SparkSession\n",
    "\n",
    "https://spark.apache.org/docs/2.4.4/api/python/pyspark.html\n",
    "\n",
    "https://spark.apache.org/docs/2.4.4/api/python/pyspark.sql.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "NameNode: http://ec2-54-152-212-69.compute-1.amazonaws.com:50070\n",
      "YARN: http://ec2-54-152-212-69.compute-1.amazonaws.com:8088\n",
      "Spark UI: http://ec2-54-152-212-69.compute-1.amazonaws.com:20888/proxy/application_1589032052307_0007\n"
     ]
    }
   ],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "import spark_utils\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "sc = SparkContext(\"yarn\", \"My App\", conf=spark_utils.get_spark_conf())\n",
    "se = SparkSession(sc)\n",
    "spark_utils.print_ui_links()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ephemeral local HDFS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Filesystem                                  Size     Used  Available  Use%\r\n",
      "hdfs://ip-172-31-26-238.ec2.internal:8020  4.3 T  444.5 M      4.3 T    0%\r\n"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -df -h"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Found 3 items\r\n",
      "drwxrwxrwt   - hdfs hadoop          0 2020-05-09 13:47 /tmp\r\n",
      "drwxr-xr-x   - hdfs hadoop          0 2020-05-09 13:47 /user\r\n",
      "drwxr-xr-x   - hdfs hadoop          0 2020-05-09 13:47 /var\r\n"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -ls /"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Persistent S3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://www.kaggle.com/c/outbrain-click-prediction/data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2020-05-09 13:43:02       1732 bootstrap.sh\r\n",
      "2020-05-07 21:18:30  176843889 clicks_test.parquet\r\n",
      "2020-05-07 21:18:30  495815517 clicks_train.parquet\r\n",
      "2020-05-07 21:18:30   34267065 documents_categories.parquet\r\n",
      "2020-05-07 21:18:30  206455957 documents_entities.parquet\r\n",
      "2020-05-07 21:18:30   23859965 documents_meta.parquet\r\n",
      "2020-05-07 21:18:31  187410196 documents_topics.parquet\r\n",
      "2020-05-07 21:18:32  734643471 events.parquet\r\n",
      "2020-05-07 21:18:35 50764611872 page_views.parquet\r\n",
      "2020-05-07 21:18:35  248421413 page_views_sample.parquet\r\n",
      "2020-05-07 21:18:36    5116927 promoted_content.parquet\r\n",
      "2020-05-07 21:18:36  273136709 sample_submission.csv\r\n"
     ]
    }
   ],
   "source": [
    "! aws s3 ls s3://ydatazian"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Read directly from S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "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": [
    "df = se.read.parquet(\"s3://ydatazian/page_views.parquet\")\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Register all tables for sql queries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "37356aeda87240e7b587c7fb9b099f88",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(FloatProgress(value=0.0, max=10.0), HTML(value='')))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    }
   ],
   "source": [
    "from IPython.display import display\n",
    "tables = [\"clicks_test\", \"clicks_train\", \n",
    "          \"documents_categories\", \"documents_entities\", \"documents_meta\", \"documents_topics\", \n",
    "          \"events\", \"page_views\", \"page_views_sample\", \"promoted_content\"]\n",
    "for name in tqdm.tqdm(tables):\n",
    "    df = se.read.parquet(\"s3://ydatazian/{}.parquet\".format(name))\n",
    "    df.registerTempTable(name)\n",
    "    print(name)\n",
    "    display(df.limit(3).toPandas())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Simple SQL query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 5.68 ms, sys: 3.9 ms, total: 9.58 ms\n",
      "Wall time: 1min 16s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df = se.sql(\n",
    "\"\"\"\n",
    "select \n",
    "    uuid, \n",
    "    count(1) as cnt\n",
    "from page_views\n",
    "group by uuid\n",
    "\"\"\")\n",
    "df.write.parquet(\"/count_per_user.parquet\", mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+---+\n",
      "|          uuid|cnt|\n",
      "+--------------+---+\n",
      "|bbb1b49336e7da|  8|\n",
      "|fe383397d00fc5|  1|\n",
      "|e90c2bc5ce5cd5| 23|\n",
      "|75db87012be51e| 43|\n",
      "|66ba64d2ce45e0|112|\n",
      "+--------------+---+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 1.4 ms, sys: 962 µs, total: 2.36 ms\n",
      "Wall time: 426 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "se.read.parquet(\"/count_per_user.parquet\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Found 4 items\n",
      "drwxr-xr-x   - hadoop hadoop          0 2020-05-09 13:58 /count_per_user.parquet\n",
      "drwxrwxrwt   - hdfs   hadoop          0 2020-05-09 13:47 /tmp\n",
      "drwxr-xr-x   - hdfs   hadoop          0 2020-05-09 13:47 /user\n",
      "drwxr-xr-x   - hdfs   hadoop          0 2020-05-09 13:47 /var\n"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -ls /"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The same in Python (10x slower)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DataFrame[uuid: string, document_id: string, timestamp: string, platform: string, geo_location: string, traffic_source: string]\n"
     ]
    }
   ],
   "source": [
    "df = se.table(\"page_views\")\n",
    "print(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MapPartitionsRDD[41] at javaToPython at NativeMethodAccessorImpl.java:0\n",
      "[Row(uuid='1fd5f051fba643', document_id='120', timestamp='31905835', platform='1', geo_location='RS', traffic_source='2')]\n",
      "1fd5f051fba643\n"
     ]
    }
   ],
   "source": [
    "rdd = df.rdd\n",
    "print(rdd)\n",
    "print(rdd.take(1))  # sent to driver (our Jupyter kernel)\n",
    "print(rdd.take(1)[0].uuid)  # Python object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rm: `/count_per_user.pickle': No such file or directory\n",
      "CPU times: user 84.7 ms, sys: 47.4 ms, total: 132 ms\n",
      "Wall time: 14min 39s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "! hdfs dfs -rm -r /count_per_user.pickle\n",
    "se.table(\"page_views\").rdd\\\n",
    "    .map(lambda x: (x.uuid, 1))\\\n",
    "    .reduceByKey(lambda x, y: x + y)\\\n",
    "    .saveAsPickleFile(\"/count_per_user.pickle\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 4.58 ms, sys: 0 ns, total: 4.58 ms\n",
      "Wall time: 273 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[('3c58d1e5492dca', 11),\n",
       " ('a12ae780183cdb', 30),\n",
       " ('1da02cb8987a6b', 3),\n",
       " ('69bf3a4dc14efa', 7),\n",
       " ('94c581628b3278', 6)]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "sc.pickleFile(\"/count_per_user.pickle\").take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# and we can convert it back to Spark DataFrame\n",
    "from pyspark.sql import Row\n",
    "rdd_of_rows = sc.pickleFile(\"/count_per_user.pickle\")\\\n",
    "    .map(lambda x: Row(uuid=x[0], cnt=x[1]))\n",
    "df = se.createDataFrame(rdd_of_rows)\n",
    "df.write.parquet(\"/count_per_user_converted.parquet\", mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cnt: long (nullable = true)\n",
      " |-- uuid: string (nullable = true)\n",
      "\n",
      "+---+--------------+\n",
      "|cnt|          uuid|\n",
      "+---+--------------+\n",
      "|  3|ec1e1d69ca180d|\n",
      "|  5|24727a458b42ec|\n",
      "|  6|ce86c4017fe64e|\n",
      "| 11|836c96b4247a66|\n",
      "|  1|c3476e60057d80|\n",
      "+---+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = se.read.parquet(\"/count_per_user_converted.parquet\")\n",
    "df.printSchema()\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10.4 G  /count_per_user.parquet\r\n",
      "19.1 G  /count_per_user.pickle\r\n",
      "10.4 G  /count_per_user_converted.parquet\r\n",
      "0  /tmp\r\n",
      "220.5 M  /user\r\n",
      "315  /var\r\n"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -du -s -h \"/*\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Collect data on the driver"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 14.6 s, sys: 2.57 s, total: 17.2 s\n",
      "Wall time: 1min 27s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "counts = se.read.parquet(\"/count_per_user_converted.parquet\").rdd\\\n",
    "    .map(lambda x: x.cnt)\\\n",
    "    .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "image/png": "iVBORw0KGgoAAAANSUhEUgAAAWoAAAEDCAYAAAAcI05xAAAABHNCSVQICAgIfAhkiAAAAAlwSFlzAAALEgAACxIB0t1+/AAAADh0RVh0U29mdHdhcmUAbWF0cGxvdGxpYiB2ZXJzaW9uMy4yLjEsIGh0dHA6Ly9tYXRwbG90bGliLm9yZy+j8jraAAANsUlEQVR4nO3dfYxldX3H8ffHXYQKFlQmhILtYmuwhISHTFCKMRak5cHgPzSB9ME2JJsm1mJjQiBt2tq/MGkVmrSkW0Qaa0FBtAZT1CLGmljsLFAKLBRUykOBHar49Ect8u0f98wyu8wyd3HOnS8771cy2XvvOTv7nV/OvvfsmXN3U1VIkvp6xXoPIEl6cYZakpoz1JLUnKGWpOYMtSQ1Z6glqbnRQp3kmiQ7k9wzxb4/m+S2JHcmuTvJOWPNJUkvN2OeUV8LnDXlvn8EfLKqTgIuAP56rKEk6eVmtFBX1VeAby9/LcnPJ7klyfYk/5LkTUu7Az89PD4U+O+x5pKkl5vNM/71tgG/W1UPJnkzkzPn04E/Bb6Q5L3AwcA7ZjyXJLU1s1AnOQT4JeCGJEsvHzj8eCFwbVX9RZJTgY8lOb6qnpvVfJLU1SzPqF8BPFNVJ66w7SKG69lV9bUkBwGHAztnOJ8ktTSz2/Oq6nvAt5L8GkAmThg2PwKcMbz+i8BBwOKsZpOkzjLWv56X5Drg7UzOjJ8C/gT4EnAVcCRwAHB9Vf1ZkuOAvwUOYfKNxUuq6gujDCZJLzOjhVqStDZ8Z6IkNTfKNxMPP/zw2rJlyxifWpL2S9u3b3+6quZW2rZqqJMcC3xi2UtvAP64qq7Y28/ZsmULCwsL+zyoJG1USf5rb9tWDXVVPQCcOHyiTcDjwKfXbDpJ0ova12vUZwDfqKq9ll+StLb2NdQXANettCHJ1iQLSRYWF70FWpLWytShTvJK4DzghpW2V9W2qpqvqvm5uRWvh0uSXoJ9OaM+G7ijqp4aaxhJ0gvtS6gvZC+XPSRJ45kq1EkOBs4Ebhp3HEnSnqZ6w0tV/RB43cizSJJW4FvIJam5Wf8PL6vacunndj1++PJz13ESSerBM2pJas5QS1JzhlqSmjPUktScoZak5gy1JDVnqCWpOUMtSc0ZaklqzlBLUnOGWpKaM9SS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWrOUEtSc4ZakpqbKtRJDktyY5L7k+xIcurYg0mSJqb9z22vBG6pqvOTvBJ41YgzSZKWWTXUSQ4F3gb8NkBV/Qj40bhjSZKWTHPp4xhgEfhokjuTXJ3k4JHnkiQNpgn1ZuBk4KqqOgn4IXDpnjsl2ZpkIcnC4uLiGo8pSRvXNKF+DHisqm4fnt/IJNy7qaptVTVfVfNzc3NrOaMkbWirhrqqngQeTXLs8NIZwH2jTiVJ2mXauz7eC3x8uOPjm8DvjDeSJGm5qUJdVXcB8yPPIklage9MlKTmDLUkNWeoJak5Qy1JzRlqSWrOUEtSc4Zakpoz1JLUnKGWpOYMtSQ1Z6glqTlDLUnNGWpJas5QS1JzhlqSmjPUktScoZak5gy1JDVnqCWpOUMtSc0ZaklqzlBLUnObp9kpycPA94EfA89W1fyYQ0mSnjdVqAe/XFVPjzaJJGlFXvqQpOamDXUBX0iyPcnWlXZIsjXJQpKFxcXFtZtQkja4aUP91qo6GTgbeE+St+25Q1Vtq6r5qpqfm5tb0yElaSObKtRV9fjw407g08ApYw4lSXreqqFOcnCSVy89Bn4FuGfswSRJE9Pc9XEE8OkkS/v/Q1XdMupUkqRdVg11VX0TOGEGs0iSVuDteZLUnKGWpOYMtSQ1Z6glqTlDLUnNGWpJas5QS1JzhlqSmjPUktScoZak5gy1JDVnqCWpOUMtSc0ZaklqzlBLUnOGWpKaM9SS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWpu6lAn2ZTkziQ3jzmQJGl3+3JGfTGwY6xBJEkrmyrUSY4GzgWuHnccSdKepj2jvgK4BHhubzsk2ZpkIcnC4uLimgwnSZoi1EneCeysqu0vtl9Vbauq+aqan5ubW7MBJWmjm+aM+jTgvCQPA9cDpyf5+1GnkiTtsmqoq+qyqjq6qrYAFwBfqqrfGH0ySRLgfdSS1N7mfdm5qr4MfHmUSSRJK/KMWpKaM9SS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWrOUEtSc4Zakpoz1JLUnKGWpOYMtSQ1Z6glqTlDLUnNGWpJas5QS1JzhlqSmjPUktScoZak5gy1JDVnqCWpuVVDneSgJF9P8u9J7k3ygVkMJkma2DzFPv8LnF5VP0hyAPDVJP9UVf868mySJKYIdVUV8IPh6QHDR405lCTpeVNdo06yKcldwE7gi1V1+wr7bE2ykGRhcXFxreeUpA1rqlBX1Y+r6kTgaOCUJMevsM+2qpqvqvm5ubm1nlOSNqx9uuujqp4BbgPOGmccSdKeprnrYy7JYcPjnwLOBO4fezBJ0sQ0d30cCfxdkk1Mwv7Jqrp53LEkSUumuevjbuCkGcwiSVqB70yUpOYMtSQ1Z6glqTlDLUnNGWpJas5QS1JzhlqSmjPUktScoZak5gy1JDVnqCWpOUMtSc0ZaklqzlBLUnOGWpKaM9SS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWrOUEtSc6uGOsnrk9yW5L4k9ya5eBaDSZImNk+xz7PA+6vqjiSvBrYn+WJV3TfybJIkpjijrqonquqO4fH3gR3AUWMPJkma2Kdr1Em2ACcBt6+wbWuShSQLi4uLazOdJGn6UCc5BPgU8L6q+t6e26tqW1XNV9X83NzcWs4oSRvaVKFOcgCTSH+8qm4adyRJ0nLT3PUR4CPAjqr60PgjSZKWm+aM+jTgN4HTk9w1fJwz8lySpMGqt+dV1VeBzGAWSdIKfGeiJDVnqCWpOUMtSc0ZaklqzlBLUnOGWpKaM9SS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWrOUEtSc4Zakpoz1JLUnKGWpOYMtSQ1Z6glqTlDLUnNGWpJas5QS1Jzq4Y6yTVJdia5ZxYDSZJ2N80Z9bXAWSPPIUnai1VDXVVfAb49g1kkSStYs2vUSbYmWUiysLi4uFafVpI2vDULdVVtq6r5qpqfm5tbq08rSRued31IUnOGWpKam+b2vOuArwHHJnksyUXjjyVJWrJ5tR2q6sJZDCJJWpmXPiSpOUMtSc0ZaklqzlBLUnOGWpKaM9SS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWrOUEtSc4Zakpoz1JLUnKGWpOYMtSQ1Z6glqTlDLUnNrfpfca2nLZd+btfjhy8/dx0nkaT14xm1JDVnqCWpOUMtSc1NdY06yVnAlcAm4OqqunzUqVbg9WpJG9WqZ9RJNgF/BZwNHAdcmOS4sQeTJE1Mc0Z9CvBQVX0TIMn1wLuA+8Yc7MUsP7tezjNtSfujaUJ9FPDosuePAW/ec6ckW4Gtw9MfJHngJcxzOPD0S/h5kxk++FJ/5svOT7ROG4RrNB3XaXWzWqOf29uGNbuPuqq2Adt+ks+RZKGq5tdopP2W67Q612g6rtPqOqzRNHd9PA68ftnzo4fXJEkzME2o/w14Y5JjkrwSuAD47LhjSZKWrHrpo6qeTfJ7wOeZ3J53TVXdO9I8P9Glkw3EdVqdazQd12l1675Gqar1nkGS9CJ8Z6IkNWeoJam5NqFOclaSB5I8lOTS9Z5nvSR5fZLbktyX5N4kFw+vvzbJF5M8OPz4muH1JPnLYd3uTnLy+n4Fs5VkU5I7k9w8PD8mye3Denxi+AY4SQ4cnj80bN+ynnPPSpLDktyY5P4kO5Kc6rH0Qkn+YPj9dk+S65Ic1OlYahFq36a+m2eB91fVccBbgPcMa3EpcGtVvRG4dXgOkzV74/CxFbhq9iOvq4uBHcuefxD4cFX9AvAd4KLh9YuA7wyvf3jYbyO4Erilqt4EnMBkrTyWlklyFPD7wHxVHc/kpokL6HQsVdW6fwCnAp9f9vwy4LL1nqvDB/CPwJnAA8CRw2tHAg8Mj/8GuHDZ/rv2298/mNzTfytwOnAzECbvINs8bN91XDG5a+nU4fHmYb+s99cw8vocCnxrz6/TY+kF67T07uvXDsfGzcCvdjqWWpxRs/Lb1I9ap1naGP5KdRJwO3BEVT0xbHoSOGJ4vJHX7grgEuC54fnrgGeq6tnh+fK12LVOw/bvDvvvz44BFoGPDpeHrk5yMB5Lu6mqx4E/Bx4BnmBybGyn0bHUJdTaQ5JDgE8B76uq7y3fVpM/yjf0fZVJ3gnsrKrt6z1LY5uBk4Grquok4Ic8f5kD8FgCGK7Rv4vJH2w/AxwMnLWuQ+2hS6h9m/oySQ5gEumPV9VNw8tPJTly2H4ksHN4faOu3WnAeUkeBq5ncvnjSuCwJEtv5Fq+FrvWadh+KPA/sxx4HTwGPFZVtw/Pb2QSbo+l3b0D+FZVLVbV/wE3MTm+2hxLXULt29QHSQJ8BNhRVR9atumzwLuHx+9mcu166fXfGr5j/xbgu8v+WrvfqqrLquroqtrC5Hj5UlX9OnAbcP6w257rtLR+5w/779dnklX1JPBokmOHl85g8s8Teyzt7hHgLUleNfz+W1qnPsfSel/IX3ZB/xzgP4FvAH+43vOs4zq8lclfRe8G7ho+zmFyDexW4EHgn4HXDvuHyR0z3wD+g8l3rtf965jxmr0duHl4/Abg68BDwA3AgcPrBw3PHxq2v2G9557R2pwILAzH02eA13gsrbhOHwDuB+4BPgYc2OlY8i3kktRcl0sfkqS9MNSS1JyhlqTmDLUkNWeoJak5Qy1JzRlqSWru/wGdZ3OYl26IiQAAAABJRU5ErkJggg==\n",
      "text/plain": [
       "<Figure size 432x288 with 1 Axes>"
      ]
     },
     "metadata": {
      "needs_background": "light"
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "_ = plt.hist(counts, bins=100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SQL + Python hybrid"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "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": [
    "se.table(\"page_views\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- uuid: string (nullable = true)\n",
      " |-- views: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- document_id: string (nullable = true)\n",
      " |    |    |-- timestamp: string (nullable = true)\n",
      " |    |    |-- platform: string (nullable = true)\n",
      " |    |    |-- geo_location: string (nullable = true)\n",
      " |    |    |-- traffic_source: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = se.sql(\n",
    "\"\"\"\n",
    "select \n",
    "    uuid, \n",
    "    collect_list(struct(document_id, timestamp, platform, geo_location, traffic_source)) as views\n",
    "from page_views\n",
    "group by uuid\n",
    "\"\"\")\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deleted /platforms_per_user.pickle\n",
      "CPU times: user 40.8 ms, sys: 29.9 ms, total: 70.8 ms\n",
      "Wall time: 10min 27s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "! hdfs dfs -rm -r /platforms_per_user.pickle\n",
    "from collections import defaultdict\n",
    "\n",
    "def mapper(x):\n",
    "    d = defaultdict(int)\n",
    "    for row in x.views:\n",
    "        d[row.platform] += 1\n",
    "    return (x.uuid, d)\n",
    "\n",
    "df.rdd.map(mapper).saveAsPickleFile(\"/platforms_per_user.pickle\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The same in Python (5x slower)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rm: `/platforms_per_user2.pickle': No such file or directory\n",
      "CPU times: user 125 ms, sys: 86.1 ms, total: 211 ms\n",
      "Wall time: 52min 8s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "! hdfs dfs -rm -r /platforms_per_user2.pickle\n",
    "\n",
    "def mapper(x):\n",
    "    d = defaultdict(int)\n",
    "    for row in x[1]:\n",
    "        d[row.platform] += 1\n",
    "    return (x[0], d)\n",
    "\n",
    "(\n",
    "    se\n",
    "    .table(\"page_views\")\n",
    "    .rdd\n",
    "    .groupBy(lambda x: x.uuid)\n",
    "    .map(mapper)\n",
    "    .saveAsPickleFile(\"/platforms_per_user2.pickle\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Homework 1: Spark EDA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. Find 10 most visited document_ids in page_views log"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# YOUR CODE HERE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "2. Find 10 most visited topic_ids in page_views log (use documents_topics table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# YOUR CODE HERE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "3. Find out how many users have at least two different traffic_sources in their page_views log"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# YOUR CODE HERE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "4. Find out the average uuid (user) CTR (number of clicks in all of her display_ids / (number of impressions ... + 10)). Plot the distribution of individual CTRs as a histogram (bins=100)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# YOUR CODE HERE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "5. Find out how many page views has an average user before encountering a display_id (use timestamp columns). Plot the distribution of individual values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# YOUR CODE HERE"
   ]
  }
 ],
 "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}