7_feature_generation_recommendation_spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\n",
      "      ____              __\n",
      "     / __/__  ___ _____/ /__\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 2.3.1\n",
      "      /_/\n",
      "\n",
      "Using Python version 3.6.5 (default, Mar 29 2018 13:14:23)\n",
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "# os.environ[\"SPARK_HOME\"]\n",
    "exec(open(os.path.join(os.environ[\"SPARK_HOME\"], 'python/pyspark/shell.py')).read())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://192.168.0.4:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>pyspark-shell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=pyspark-shell>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load data into Spark DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "import pyspark.sql.functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: string, event: string, song_id: string, date: string]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# df is A Spark DataFrame\n",
    "df = spark.read.csv('../data/event_ds.csv',header=True).cache()\n",
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----+--------+----------+\n",
      "|      uid|event| song_id|      date|\n",
      "+---------+-----+--------+----------+\n",
      "|168535490|    P| 6616004|2017-03-30|\n",
      "|168551042|    P|  505355|2017-03-30|\n",
      "|168540348|    P|   77260|2017-03-30|\n",
      "|168530895|    P|       0|2017-03-30|\n",
      "|168530479|    P| 1676759|2017-03-30|\n",
      "|168551607|    P| 5517668|2017-03-30|\n",
      "|168548101|    P| 6913185|2017-03-30|\n",
      "|168542680|    P|       0|2017-03-30|\n",
      "|168550237|    P| 6511070|2017-03-30|\n",
      "|168519439|    P|       0|2017-03-30|\n",
      "|168548942|    P| 4112522|2017-03-30|\n",
      "|168551495|    P|22797719|2017-03-30|\n",
      "|168551383|    P| 7149583|2017-03-30|\n",
      "|168550382|    P| 3971726|2017-03-30|\n",
      "|168548945|    P|20429929|2017-03-30|\n",
      "|168550086|    P| 7172624|2017-03-30|\n",
      "|168540348|    P| 6437996|2017-03-30|\n",
      "|168551522|    P|  150918|2017-03-30|\n",
      "|168541857|    P|  295529|2017-03-30|\n",
      "|168546393|    P|  242615|2017-03-30|\n",
      "+---------+-----+--------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# show the first 20 rows of the Spark DataFrame\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## Data Exploration"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Features Type Casting and Verification"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: int, event: string, song_id: int, date: date]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# create new or overwrite original field with withColumn: \n",
    "# for \"date\" column, convert \"string\" type to \"date\" type \n",
    "# for \"id\" column, convert \"string\" type to \"int\" type\n",
    "df = df.withColumn('date', F.col('date').cast('date'))\n",
    "for col in ['uid', 'song_id']:\n",
    "    df = df.withColumn(col, F.col(col).cast('int'))\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that song_id has a **special** negative value and there are missing values in __event__ column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+-------+-----------------+\n",
      "|summary|                 uid|  event|          song_id|\n",
      "+-------+--------------------+-------+-----------------+\n",
      "|  count|             2194568|2194568|          2049856|\n",
      "|   mean|1.6645771046223176E8|   null| 5915507.37680842|\n",
      "| stddev|1.6287908907350207E7|   null|6653398.679705742|\n",
      "|    min|             2626270|      D|               -1|\n",
      "|    max|           168941043|      S|         24024145|\n",
      "+-------+--------------------+-------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# show the descriptive statistics of the DataFrame\n",
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Missing Values\n",
    "Only Nulls in **song_id** feature."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----+-------+----+\n",
      "|uid|event|song_id|date|\n",
      "+---+-----+-------+----+\n",
      "|  0|    0| 144712|   0|\n",
      "+---+-----+-------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Count the number of Nulls in every feature\n",
    "df.select(*[F.sum(F.col(c).isNull().cast(\"int\")).alias(c) for c in df.columns]).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+-----+-------+----------+\n",
      "|      uid|event|song_id|      date|\n",
      "+---------+-----+-------+----------+\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "|168564110|    P|   null|2017-03-30|\n",
      "+---------+-----+-------+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Check the data with missing values\n",
    "df.filter(F.col('song_id').isNull()).show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Nulls Pattern Inspection\n",
    "The song ids of the \"search\" event are all missing. A tiny proportion of the \"play\" event song ids is missing. None of song ids is absent in \"download\" event. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-------+\n",
      "|event|  count|\n",
      "+-----+-------+\n",
      "|    D| 119950|\n",
      "|    S| 144337|\n",
      "|    P|1930281|\n",
      "+-----+-------+\n",
      "\n",
      "+-----+------+\n",
      "|event| count|\n",
      "+-----+------+\n",
      "|    S|144337|\n",
      "|    P|   375|\n",
      "+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# group by aggregation\n",
    "df.groupBy('event').count().show()\n",
    "df.filter(F.col('song_id').isNull()).groupBy('event').count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Song ids and User ids"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2194568"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# simple count rows\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(9757, 157320)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# count number of distinct user ids and song ids\n",
    "df.select('uid').distinct().count(), df.select('song_id').distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "597572"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('uid', 'song_id').distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+\n",
      "|start_date|  end_date|\n",
      "+----------+----------+\n",
      "|2017-03-30|2017-05-12|\n",
      "+----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(F.min(F.col('date')).alias('start_date'), F.max(F.col('date')).alias('end_date')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "date_count = df.groupBy('date').count().toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(array([736419., 736426., 736433., 736440., 736447., 736454., 736461.]),\n",
       " <a list of 7 Text xticklabel objects>)"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "data": {
      "image/png": "iVBORw0KGgoAAAANSUhEUgAAAY0AAAEsCAYAAAA/5++aAAAABHNCSVQICAgIfAhkiAAAAAlwSFlzAAALEgAACxIB0t1+/AAAADl0RVh0U29mdHdhcmUAbWF0cGxvdGxpYiB2ZXJzaW9uIDIuMi4yLCBodHRwOi8vbWF0cGxvdGxpYi5vcmcvhp/UCwAAHnhJREFUeJzt3Xu0XGWd5vHvYwKIIoTLEZkEDaNRQR2DRGQtZ9bQ4EDAHoM9gPCHRBs76oCtjm0TezkDLeJgTys2jrKGXqQJLgUitpJpA3QGsdUZuYSLYIjAEdBEboEEhEahgWf+qDdaHOqcek/l5Oza8HzWqnV2/fbt9yY59WRfqkq2iYiIqPGiphuIiIj2SGhERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERUS2hERES1hEZERFSb2XQDU22PPfbw3Llzm24jIqJVrr/++gdtj/Rb7nkXGnPnzmXNmjVNtxER0SqSflGzXE5PRUREtYRGRERU6xsakl4s6VpJP5G0VtJflvr5ku6SdFN5zC91STpb0qikmyW9pWtbiyXdUR6Lu+oHSLqlrHO2JJX6bpJWl+VXS9p16v8IIiKiVs2RxhPAIbbfDMwHFko6qMz7pO355XFTqR0BzCuPJcA50AkA4FTgbcCBwKldIXBOWXbLegtLfSlwpe15wJXleURENKRvaLjjsfJ0u/KY6Es4FgEXlPWuBmZJ2gs4HFhte5PtzcBqOgG0F7Cz7R+78+UeFwBHdW1reZle3lWPiIgGVF3TkDRD0k3AA3Re+K8ps84op6DOkrRDqc0G1netvqHUJqpv6FEH2NP2vQDl58vH6W+JpDWS1mzcuLFmSBERMYCq0LD9tO35wBzgQElvBD4FvB54K7AbcEpZXL02MUC9mu1zbS+wvWBkpO9txhERMaBJ3T1l+2Hg+8BC2/eWU1BPAH9H5zoFdI4U9u5abQ5wT5/6nB51gPvL6SvKzwcm029EREytmrunRiTNKtM7Au8Aftb1Yi461xp+WlZZCZxQ7qI6CHiknFq6AjhM0q7lAvhhwBVl3qOSDirbOgG4tGtbW+6yWtxVj4iIBtS8I3wvYLmkGXRCZoXtf5D0PUkjdE4v3QR8qCy/CjgSGAUeB94PYHuTpNOB68pyn7G9qUx/GDgf2BG4rDwAzgRWSDoR+CVwzKADrTF36Xd71u8+853bcrcREa3RNzRs3wzs36N+yDjLGzhpnHnLgGU96muAN/aoPwQc2q/HiIiYHnlHeEREVEtoREREtYRGRERUS2hERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERUS2hERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERUS2hERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERU6xsakl4s6VpJP5G0VtJflvo+kq6RdIekiyVtX+o7lOejZf7crm19qtRvk3R4V31hqY1KWtpV77mPiIhoRs2RxhPAIbbfDMwHFko6CPg8cJbtecBm4MSy/InAZtuvAc4qyyFpP+A44A3AQuCrkmZImgF8BTgC2A84vizLBPuIiIgG9A0NdzxWnm5XHgYOAS4p9eXAUWV6UXlOmX+oJJX6RbafsH0XMAocWB6jtu+0/SRwEbCorDPePiIiogFV1zTKEcFNwAPAauDnwMO2nyqLbABml+nZwHqAMv8RYPfu+ph1xqvvPsE+IiKiAVWhYftp2/OBOXSODPbttVj5qXHmTVX9OSQtkbRG0pqNGzf2WiQiIqbApO6esv0w8H3gIGCWpJll1hzgnjK9AdgboMzfBdjUXR+zznj1ByfYx9i+zrW9wPaCkZGRyQwpIiImoebuqRFJs8r0jsA7gHXAVcDRZbHFwKVlemV5Tpn/Pdsu9ePK3VX7APOAa4HrgHnlTqnt6VwsX1nWGW8fERHRgJn9F2EvYHm5y+lFwArb/yDpVuAiSZ8FbgTOK8ufB3xN0iidI4zjAGyvlbQCuBV4CjjJ9tMAkk4GrgBmAMtsry3bOmWcfURERAP6hobtm4H9e9TvpHN9Y2z9t8Ax42zrDOCMHvVVwKrafURERDPyjvCIiKiW0IiIiGoJjYiIqJbQiIiIagmNiIioltCIiIhqCY2IiKiW0IiIiGoJjYiIqJbQiIiIagmNiIioltCIiIhqCY2IiKiW0IiIiGoJjYiIqJbQiIiIagmNiIioltCIiIhqCY2IiKiW0IiIiGoJjYiIqNY3NCTtLekqSeskrZX00VI/TdKvJN1UHkd2rfMpSaOSbpN0eFd9YamNSlraVd9H0jWS7pB0saTtS32H8ny0zJ87lYOPiIjJqTnSeAr4hO19gYOAkyTtV+adZXt+eawCKPOOA94ALAS+KmmGpBnAV4AjgP2A47u28/myrXnAZuDEUj8R2Gz7NcBZZbmIiGhI39Cwfa/tG8r0o8A6YPYEqywCLrL9hO27gFHgwPIYtX2n7SeBi4BFkgQcAlxS1l8OHNW1reVl+hLg0LJ8REQ0YFLXNMrpof2Ba0rpZEk3S1omaddSmw2s71ptQ6mNV98deNj2U2Pqz9pWmf9IWT4iIhpQHRqSdgK+BXzM9q+Bc4BXA/OBe4EvbFm0x+oeoD7Rtsb2tkTSGklrNm7cOOE4IiJicFWhIWk7OoHxddt/D2D7fttP234G+Fs6p5+gc6Swd9fqc4B7Jqg/CMySNHNM/VnbKvN3ATaN7c/2ubYX2F4wMjJSM6SIiBhAzd1TAs4D1tn+Yld9r67F3g38tEyvBI4rdz7tA8wDrgWuA+aVO6W2p3OxfKVtA1cBR5f1FwOXdm1rcZk+GvheWT4iIhows/8ivB14L3CLpJtK7S/o3P00n87poruBDwLYXitpBXArnTuvTrL9NICkk4ErgBnAMttry/ZOAS6S9FngRjohRfn5NUmjdI4wjtuKsUZExFbqGxq2f0TvawurJljnDOCMHvVVvdazfSe/P73VXf8tcEy/HiMiYnrkHeEREVEtoREREdUSGhERUS2hERER1RIaERFRLaERERHVEhoREVEtoREREdUSGhERUS2hERER1RIaERFRLaERERHVEhoREVEtoREREdUSGhERUS2hERER1RIaERFRLaERERHVEhoREVEtoREREdUSGhERUa1vaEjaW9JVktZJWivpo6W+m6TVku4oP3ctdUk6W9KopJslvaVrW4vL8ndIWtxVP0DSLWWdsyVpon1EREQzao40ngI+YXtf4CDgJEn7AUuBK23PA64szwGOAOaVxxLgHOgEAHAq8DbgQODUrhA4pyy7Zb2FpT7ePiIiogF9Q8P2vbZvKNOPAuuA2cAiYHlZbDlwVJleBFzgjquBWZL2Ag4HVtveZHszsBpYWObtbPvHtg1cMGZbvfYRERENmNQ1DUlzgf2Ba4A9bd8LnWABXl4Wmw2s71ptQ6lNVN/Qo84E+4iIiAZUh4aknYBvAR+z/euJFu1R8wD1apKWSFojac3GjRsns2pERExCVWhI2o5OYHzd9t+X8v3l1BLl5wOlvgHYu2v1OcA9fepzetQn2sez2D7X9gLbC0ZGRmqGFBERA6i5e0rAecA621/smrUS2HIH1GLg0q76CeUuqoOAR8qppSuAwyTtWi6AHwZcUeY9Kumgsq8Txmyr1z4iIqIBMyuWeTvwXuAWSTeV2l8AZwIrJJ0I/BI4psxbBRwJjAKPA+8HsL1J0unAdWW5z9jeVKY/DJwP7AhcVh5MsI+IiGhA39Cw/SN6X3cAOLTH8gZOGmdby4BlPeprgDf2qD/Uax8REdGMvCM8IiKqJTQiIqJaQiMiIqolNCIiolpCIyIiqiU0IiKiWkIjIiKqJTQiIqJaQiMiIqolNCIiolpCIyIiqiU0IiKiWkIjIiKqJTQiIqJaQiMiIqolNCIiolpCIyIiqiU0IiKiWkIjIiKqJTQiIqJaQiMiIqr1DQ1JyyQ9IOmnXbXTJP1K0k3lcWTXvE9JGpV0m6TDu+oLS21U0tKu+j6SrpF0h6SLJW1f6juU56Nl/typGnRERAym5kjjfGBhj/pZtueXxyoASfsBxwFvKOt8VdIMSTOArwBHAPsBx5dlAT5ftjUP2AycWOonApttvwY4qywXEREN6hsatn8AbKrc3iLgIttP2L4LGAUOLI9R23fafhK4CFgkScAhwCVl/eXAUV3bWl6mLwEOLctHRERDtuaaxsmSbi6nr3YttdnA+q5lNpTaePXdgYdtPzWm/qxtlfmPlOUjIqIhg4bGOcCrgfnAvcAXSr3XkYAHqE+0reeQtETSGklrNm7cOFHfERGxFQYKDdv3237a9jPA39I5/QSdI4W9uxadA9wzQf1BYJakmWPqz9pWmb8L45wms32u7QW2F4yMjAwypIiIqDBQaEjaq+vpu4Etd1atBI4rdz7tA8wDrgWuA+aVO6W2p3OxfKVtA1cBR5f1FwOXdm1rcZk+GvheWT4iIhoys98Cki4EDgb2kLQBOBU4WNJ8OqeL7gY+CGB7raQVwK3AU8BJtp8u2zkZuAKYASyzvbbs4hTgIkmfBW4Eziv184CvSRqlc4Rx3FaPdivMXfrdnvW7z3znNHcSEdGcvqFh+/ge5fN61LYsfwZwRo/6KmBVj/qd/P70Vnf9t8Ax/fqLiIjpk3eER0REtYRGRERUS2hERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERUS2hERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERUS2hERES1hEZERFRLaERERLWERkREVEtoREREtYRGRERUS2hERES1vqEhaZmkByT9tKu2m6TVku4oP3ctdUk6W9KopJslvaVrncVl+TskLe6qHyDplrLO2ZI00T4iIqI5NUca5wMLx9SWAlfangdcWZ4DHAHMK48lwDnQCQDgVOBtwIHAqV0hcE5Zdst6C/vsIyIiGtI3NGz/ANg0prwIWF6mlwNHddUvcMfVwCxJewGHA6ttb7K9GVgNLCzzdrb9Y9sGLhizrV77iIiIhgx6TWNP2/cClJ8vL/XZwPqu5TaU2kT1DT3qE+0jIiIaMtUXwtWj5gHqk9uptETSGklrNm7cONnVIyKi0qChcX85tUT5+UCpbwD27lpuDnBPn/qcHvWJ9vEcts+1vcD2gpGRkQGHFBER/QwaGiuBLXdALQYu7aqfUO6iOgh4pJxaugI4TNKu5QL4YcAVZd6jkg4qd02dMGZbvfYRERENmdlvAUkXAgcDe0jaQOcuqDOBFZJOBH4JHFMWXwUcCYwCjwPvB7C9SdLpwHVluc/Y3nJx/cN07tDaEbisPJhgH0Nn7tLv9qzffeY7p7mTiIhtq29o2D5+nFmH9ljWwEnjbGcZsKxHfQ3wxh71h3rtIyIimpN3hEdERLWERkREVEtoREREtYRGRERUS2hERES1vndPxdbJ7bgR8XySI42IiKiW0IiIiGoJjYiIqJbQiIiIagmNiIioltCIiIhqCY2IiKiW92kMqV7v78h7OyKiaTnSiIiIagmNiIioltCIiIhqCY2IiKiW0IiIiGoJjYiIqJZbbl9AchtvRGytrQoNSXcDjwJPA0/ZXiBpN+BiYC5wN3Cs7c2SBPwNcCTwOPA+2zeU7SwGPl02+1nby0v9AOB8YEdgFfBR296anp8PJnrxTzBExLY0FUcaf2D7wa7nS4ErbZ8paWl5fgpwBDCvPN4GnAO8rYTMqcACwMD1klba3lyWWQJcTSc0FgKXTUHPMUbCJiJqbItrGouA5WV6OXBUV/0Cd1wNzJK0F3A4sNr2phIUq4GFZd7Otn9cji4u6NpWREQ0YGuPNAz8oyQD/8v2ucCetu8FsH2vpJeXZWcD67vW3VBqE9U39KjHNMtRSERssbWh8Xbb95RgWC3pZxMsqx41D1B/7oalJXROY/HKV75y4o5jSiVQIl5Ytur0lO17ys8HgG8DBwL3l1NLlJ8PlMU3AHt3rT4HuKdPfU6Peq8+zrW9wPaCkZGRrRlSRERMYOAjDUkvBV5k+9EyfRjwGWAlsBg4s/y8tKyyEjhZ0kV0LoQ/Uk5fXQF8TtKuZbnDgE/Z3iTpUUkHAdcAJwBfHrTfaIdeRy7Q/+hl0PWitxxBxni25vTUnsC3O3fSMhP4hu3LJV0HrJB0IvBL4Jiy/Co6t9uO0rnl9v0AJRxOB64ry33G9qYy/WF+f8vtZeTOqeeN6XxRSqBETJ2BQ8P2ncCbe9QfAg7tUTdw0jjbWgYs61FfA7xx0B4jImJq5R3h8YKW02ERk5PQiGiBhFQMi4RGxDim+4U6wRBtkNCImGITvfgPQxBtyxBq+11XCe7+EhoRLdeWF+q2f9BmAqUjoRHxAjXoC3Ve4JvX5PgSGhHRSsN059ugpySH6VRmrYRGtMaw/hJFvJAkNCJiaOU/CsMn3xEeERHVEhoREVEtoREREdUSGhERUS2hERER1RIaERFRLaERERHVEhoREVEtoREREdUSGhERUS2hERER1fLZUw3K5+pERNsM/ZGGpIWSbpM0Kmlp0/1ERLyQDXVoSJoBfAU4AtgPOF7Sfs12FRHxwjXUoQEcCIzavtP2k8BFwKKGe4qIeMEa9tCYDazver6h1CIiogGy3XQP45J0DHC47Q+U5+8FDrT9kTHLLQGWlKevA27bxq3tATy4jfexrbV9DG3vH9o/hrb3DxlDt1fZHum30LDfPbUB2Lvr+RzgnrEL2T4XOHe6mpK0xvaC6drfttD2MbS9f2j/GNreP2QMgxj201PXAfMk7SNpe+A4YGXDPUVEvGAN9ZGG7acknQxcAcwAltle23BbEREvWEMdGgC2VwGrmu5jjGk7FbYNtX0Mbe8f2j+GtvcPGcOkDfWF8IiIGC7Dfk0jIiKGSEIjIiKqJTQiIqJaQiNaSdJbmu7hhU7SHk33ENMvobEVJF3WdA/9SLpB0qclvbrpXgYl6S1jHgcAKyXt/3wID0m3NN1DP5KOkHSXpB+VP/e1wDWSNkg6tOn+akj6467pOZKulPSwpP8n6bVN9lZL0pskXS1pvaRzJe3aNe/a6ehh6G+5bdoEL0oC5k9nLwPaFZgFXCXpPuBC4GLbz3ln/RBbA1wNPNFV2x34ImDgkCaamgxJfzTeLOAV09nLgP47cCSdf0v/B3in7asl7Qt8HWhDeJ8MLCvTXwRWAP+BzoegngO0IfzOAU6j8/vwAeBHkt5l++fAdtPRQEKjv+uAf6Lzyz3WrGnuZRCbbf8Z8GeS/h1wPHCDpHXAheUjWIbdscBHgP9R3reDpLts/0GzbU3KxXReXHvd4/7iae5lEM/YXgcg6XHbVwPYXiepjWcsXmv72DL9bUn/rdFu6u1k+/Iy/deSrgcuL5/LNy3vn0ho9LcO+KDtO8bOkLS+x/JDy/YPgR9K+gid/2G9hxa8ucn2JZIuB06X9H7gE0zTL8gUuhn4a9s/HTtD0jsa6GeyHpb0QWBnYLOkj9P5n/o7gMca7azeHEln0/kP4Iik7Wz/S5k3Lf9LnwKStIvtRwBsXyXpPwHfAnabjgYSGv2dxvjXfj4yTn2Y3D62YPtp4PLyaAXbjwEflzQfWA7s1HBLk/Ux4NfjzHv3dDYyoMXAp4FngMPoHLFeAfwC+JMG+5qMT3ZNr6Hzb2izpFfQns+0+zywL53TUwDYvrlcV/qv09FA3hEerSNJwMtsj/ciHBHbSBvPRU4rdRwr6ZgyfaiksyX957acy5V0uKRzJK2UdGmZXth0X4Nyx68BWnQuesvfw4mS5o6p/3HvNYbL86D/mZI+KOkySTdL+kmZ/pCktpyeGpekaTnVnCONPiR9FXg5sD2d0ws7AP+bzp0k99v+aIPt9SXpS8BrgQvofD8JdL6X5ATgjmHvvx9Jv7T9yqb76EfS54B/C9wA/EfgS7a/XObdYHuo7z5qe/8Aki4EHqZzerP7d2ExsJvt9zTVWy1J4123EPAT23O2eQ8JjYlJusX2m8r/RO4D9rL9pKSZwI2239RwixOSdLvt59yDXk7x3G57XgNtTYqk8U5DCdjR9tBfmyvvxdi/fNz/LOAbwG22Py7pRtv7N9zihNreP4Ck22y/bpx5PX9Pho2kp+lcR+q+m9Pl+Wzb22/rHlpxeqVhTwGUuyyus/1kef4U8HSTjVX6raQDe9TfCvx2upsZ0MPAPNs7j3m8DLi36eYqzSz/ZrD9MJ3/re8s6Zt0jmKHXdv7h85F72O6TytLepGk9wCbG+xrMu4EDra9T9fjX9veB7h/OhpIaPR3n6SdAGz/7jpAuePiyca6qvc+4MuSbpX0j+WxDvhymdcGFwCvGmfeN6azka3wc0n/fssT20/bPpHO99nv21xb1dreP3S++fNo4H5Jt0u6nc4L7R+VeW3wJTpv2O3lr6ajgZyeGpCklwIvtf1A073UKCE3m85h7Abb9zXc0guKpB0BbP+mx7zZtn81/V3Va3v/Y0nanc7r34NN99I2OdIYgKTTbP9zWwIDwPZ9tq+3vQb4UNP9bC1JpzXdw2TY/s3YF9wtY2jDC27b+x/L9kO2H5yuO462pekeQ0JjMO9quoGt1Pb+IWMYBm3vH2BB0w1MgWkdQ0JjML0+h6pN2t4/ZAzDoO39A7TmbMEEpnUMuaYxAEkvsv1M030Mqu39Q8YwDNrefwwmRxp9SHqJpD+X9ElJL5b0PuA7kv5qy11VLfSzphuYAhlD81rVv6R/0zW9nTrfM7NS0uckvaTJ3moNwxhypNGHpBXAemBH4HV0PvV2BZ371F9h+70NtteXpEf5/SfCbjmd8BLgcTqfyLFzI41NQsbQvLb3D89+57qkL9D5Tpa/A44Cdrd9QpP91RiGMQz9O2mHwGttH1veQX0v8A7blvRD4CcN91bjfGAX4JO274fffRfFPo12NTnnkzE07Xza3T88+xrMocBbbf+LpB/Qjt9lGIIxJDQqlaBY5XJoVp4P/WGa7Y+o8/WoF0r6DvA/adl3UWQMzWt7/8Uukt5N57T8DuVTHlrzu1w0PoZc0+hvTdc7wru/Y/jVwKONdTUJtq+n82U50PkWwjZ8U9yzZAzNa3v/dHp+F/CHwNWS9oTfvfG1LW/ya3wMuaaxFSTJLfsDlLQXnQ+eW9V0L4PKGJrX9v5jcDk9VUHS6+l8+fxsOofk9wArXb4zedj16r+cj25F/5AxDIO29w/t/12G5seQ01N9SDoFuIjOBahrgevK9IWSljbZW4229w8ZwzBoe/+QMUxZDy07uzLtyidhvsG//wL6LfXtgbUe8u+jaHv/kDEMg7b3DxnDVMmRRn/PAP+qR32vMm/Ytb1/yBiGQdv7h4xhSuSaRn8fA66UdAedN/kBvBJ4DXByY13Va3v/kDEMg7b3DxnDlMjpqQrqfNPXgXR9HwWdb/Frwzf3tb5/yBiGQdv7h4xhSvaf0Jg8SUtst/Zz+NveP2QMw6Dt/UPGMIhc0xhM27/EqO39Q8YwDNreP2QMk5bQGEzbv0eg7f1DxjAM2t4/ZAyT31lOT02epDm2NzTdx6Da3j9kDMOg7f1DxjCIHGn0IelPJe3dXWvTP7K29w8ZwzBoe/+QMUxZDznSmJikR4B/Bn4OXAh80/bGZruq1/b+IWMYBm3vHzKGqZIjjf7uBOYApwMHALdKulzSYkkva7a1Km3vHzKGYdD2/iFjmBI50uhDXd+UVZ5vBxwBHE/nC5lGGmuuQtv7h4xhGLS9f8gYpqyHhMbEJN1oe/9x5u1o+zfT3dNktL1/yBiGQdv7h4xhynpIaExM0mtt3950H4Nqe/+QMQyDtvcPGcOU9ZDQGJyknWw/1nQfg2p7/5AxDIO29w8Zw2TkQvjWubXpBrZS2/uHjGEYtL1/yBiq5VNu+5D0X8abBew0nb0Mou39Q8YwDNreP2QMUyVHGv19DtgVeNmYx06048+v7f1DxjAM2t4/ZAxTIkca/d0AfMf29WNnSPpAA/1MVtv7h4xhGLS9f8gYpkQuhPch6XXApl7vupS0p+37G2irWtv7h4xhGLS9f8gYpqyHhEZERNRqy3m8xkjaRdKZkn4m6aHyWFdqs5rur5+29w8ZwzBoe/+QMUyVhEZ/K4DNwMG2d7e9O/AHpfbNRjur0/b+IWMYBm3vHzKGKZHTU31Ius326yY7b1i0vX/IGIZB2/uHjGGq5Eijv19I+nNJe24pSNpT0inA+gb7qtX2/iFjGAZt7x8yhimR0OjvPcDuwD9J2ixpE/B9YDfg2CYbq9T2/iFjGAZt7x8yhimR01MVJL2ezmfYX9392S6SFtq+vLnO6rS9f8gYhkHb+4eMYUrYzmOCB/CnwG3Ad4C7gUVd825our/ne/8Zw3A82t5/xjB1j7wjvL8/AQ6w/ZikucAlkuba/hs6n/cy7NreP2QMw6Dt/UPGMCUSGv3NcDkEtH23pIPp/EW9inb8Q2t7/5AxDIO29w8Zw5TIhfD+7pM0f8uT8hf2h8AewJsa66pe2/uHjGEYtL1/yBimRC6E9yFpDvCU7ft6zHu77f/bQFvV2t4/ZAzDoO39Q8YwZT0kNCIiolZOT0VERLWERkREVEtoREREtYRGRERUS2hERES1/w85Y7rwEVbGggAAAABJRU5ErkJggg==\n",
      "text/plain": [
       "<Figure size 432x288 with 1 Axes>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# We use matplotlib for plotting\n",
    "import matplotlib.pyplot as plt\n",
    "# This statement allow to display plot without asking to\n",
    "%matplotlib inline\n",
    "plt.bar(date_count.date.tolist(), date_count['count'])\n",
    "plt.xticks(rotation='vertical')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "---\n",
    "## Generation of Implicit Rating"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Frequecy "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "recommendation window: 2017-03-30 ~ 2017-05-12 \n",
      "days: 44\n"
     ]
    }
   ],
   "source": [
    "import datetime\n",
    "from dateutil import parser\n",
    "\n",
    "reco_window_size= 44 \n",
    "reco_window_end_date = parser.parse('2017-05-12').date()\n",
    "reco_window_start_date = reco_window_end_date - datetime.timedelta(reco_window_size - 1)\n",
    "print('recommendation window:', reco_window_start_date, '~', \n",
    "      reco_window_end_date, '\\ndays:', reco_window_size)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def frequency_reco_generation_time_windows(df,event,time_window_list,snapshot_date):\n",
    "    \"\"\"\n",
    "    generate frequency features for one event type and a list of time windows\n",
    "    \"\"\"\n",
    "    df_feature = df \\\n",
    "        .filter(F.col('event')==event) \\\n",
    "        .groupBy('uid', 'song_id') \\\n",
    "        .agg(*[F.sum(F.when((F.col('date')>=snapshot_date-datetime.timedelta(time_window-1)) \\\n",
    "                            & (F.col('date')<=snapshot_date),1).otherwise(0))\\\n",
    "               .alias('freq_'+event+'_last_'+str(time_window)) \\\n",
    "                for time_window in time_window_list]\n",
    "            )# *[] opens list and make them comma separated\n",
    "    return df_feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# generate frequency features for all event_list, time_window_list\n",
    "event_list = ['P','D']\n",
    "time_window_list = [1,3,7,14,30,44]\n",
    "snapshot_date = reco_window_end_date\n",
    "\n",
    "df_reco_list = []\n",
    "for event in event_list:\n",
    "    df_reco_list.append(frequency_reco_generation_time_windows(df,event,\n",
    "                                                                     time_window_list,snapshot_date))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+-------------+-------------+-------------+--------------+--------------+--------------+\n",
      "|      uid| song_id|freq_P_last_1|freq_P_last_3|freq_P_last_7|freq_P_last_14|freq_P_last_30|freq_P_last_44|\n",
      "+---------+--------+-------------+-------------+-------------+--------------+--------------+--------------+\n",
      "|168548942| 4589801|            0|            0|            0|             0|             0|             1|\n",
      "|168542101|  849435|            0|            0|            0|             0|             0|             1|\n",
      "|168551694|23489729|            0|            0|            0|             0|             0|             1|\n",
      "|168551694| 1222356|            0|            0|            0|             0|             0|             1|\n",
      "|168552867| 1094256|            0|            0|            0|             0|             0|             1|\n",
      "|168536197| 6762377|            0|            0|            0|             0|             0|             4|\n",
      "|168553253|  891952|            0|            0|            0|             0|             0|             1|\n",
      "|168553426| 6165034|            0|            0|            0|             0|             0|             1|\n",
      "|168531662| 9950164|            0|            0|            0|             0|             0|             4|\n",
      "|168553520|  327577|            0|            0|            0|             0|             0|             1|\n",
      "|168550815| 7077575|            0|            0|            0|             0|             0|             3|\n",
      "|168548138| 3973072|            0|            0|            0|             0|             0|             1|\n",
      "|168548942| 1161336|            0|            0|            0|             6|            23|            26|\n",
      "|168548211| 1238180|            0|            0|            0|             0|             0|             2|\n",
      "|168554809|23491656|            0|            0|            0|             0|             0|             1|\n",
      "|168555719| 5825379|            0|            0|            0|             0|             0|             2|\n",
      "|168541614| 4229383|            0|            0|            0|             0|             0|             1|\n",
      "|168555652|       0|            0|            0|            0|             0|             0|            71|\n",
      "|168553343|19535291|            0|            0|            1|             2|             2|            10|\n",
      "|168556816| 9848594|            0|            0|            0|             0|             0|             1|\n",
      "+---------+--------+-------------+-------------+-------------+--------------+--------------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_reco_list[0].show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recency"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import timedelta\n",
    "def recency_reco_generation_time_windows(df,event,snapshot_date):\n",
    "    \"\"\"\n",
    "    generate recency features for one event type, so totally 3 possible features\n",
    "    \"\"\"\n",
    "    df_feature = df \\\n",
    "        .filter(F.col('event')==event) \\\n",
    "        .groupBy('uid', 'song_id') \\\n",
    "        .agg(F.datediff(F.lit(snapshot_date), F.max(F.col('date'))).alias('rec_' + event) \n",
    "            ) \n",
    "    \n",
    "    return df_feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[DataFrame[uid: int, song_id: int, freq_P_last_1: bigint, freq_P_last_3: bigint, freq_P_last_7: bigint, freq_P_last_14: bigint, freq_P_last_30: bigint, freq_P_last_44: bigint],\n",
       " DataFrame[uid: int, song_id: int, freq_D_last_1: bigint, freq_D_last_3: bigint, freq_D_last_7: bigint, freq_D_last_14: bigint, freq_D_last_30: bigint, freq_D_last_44: bigint],\n",
       " DataFrame[uid: int, song_id: int, rec_P: int],\n",
       " DataFrame[uid: int, song_id: int, rec_D: int]]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "event_list = ['P','D']\n",
    "snapshot_date = reco_window_end_date\n",
    "\n",
    "for event in event_list:\n",
    "    df_reco_list.append(recency_reco_generation_time_windows(df,event,\n",
    "                                                                     snapshot_date))\n",
    "df_reco_list"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "### Play/Song Length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+------+-------+----------+---------+-----------+\n",
      "|      uid|device|song_id|      date|play_time|song_length|\n",
      "+---------+------+-------+----------+---------+-----------+\n",
      "|168535490|    ar|6616004|2017-03-30|      283|        283|\n",
      "|168551042|    ar| 505355|2017-03-30|      106|        277|\n",
      "|168540348|    ar|  77260|2017-03-30|    64528|          0|\n",
      "|168530895|    ar|      0|2017-03-30|      264|        265|\n",
      "|168530479|    ar|1676759|2017-03-30|      338|        338|\n",
      "|168551607|    ar|5517668|2017-03-30|      231|        245|\n",
      "|168548101|    ip|6913185|2017-03-30|       40|        198|\n",
      "|168542680|    ar|      0|2017-03-30|      284|        285|\n",
      "|168550237|    ar|6511070|2017-03-30|      144|        237|\n",
      "|168519439|    ar|      0|2017-03-30|      227|        228|\n",
      "+---------+------+-------+----------+---------+-----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_play = spark.read.csv('../data/play_ds.csv',header=True)\n",
    "df_play.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# df_play_feature_window.show(10)\n",
    "df_profile = df_play.drop('device').distinct() \\\n",
    "    .withColumn('uid', F.col('uid').cast('int')) \\\n",
    "    .withColumn('song_id', F.col('song_id').cast('int')) \\\n",
    "    .withColumn('play_time', F.col('play_time').cast(\"float\")) \\\n",
    "    .withColumn('song_length', F.col('song_length').cast('float'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Play Time and Song Length Cleansing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------+----------+---------+-----------+\n",
      "|      uid| song_id|      date|play_time|song_length|\n",
      "+---------+--------+----------+---------+-----------+\n",
      "|168550004|       0|2017-03-30|      0.0|      180.0|\n",
      "|168530027|  461314|2017-03-30|    277.0|      277.0|\n",
      "|168548945|20274234|2017-03-30|      1.0|      307.0|\n",
      "|168553584| 7149583|2017-03-30|    214.0|      215.0|\n",
      "|168553426|   64840|2017-03-30|     27.0|      273.0|\n",
      "|168551535|22873373|2017-03-30|      0.0|        0.0|\n",
      "|168550549|11592276|2017-03-30|      1.0|      285.0|\n",
      "|168543638| 3338912|2017-03-30|      0.0|       29.0|\n",
      "|168548514|  145454|2017-03-30|      3.0|      324.0|\n",
      "|168553532| 4078943|2017-03-30|     22.0|      283.0|\n",
      "+---------+--------+----------+---------+-----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_profile = df_profile.filter((F.col('play_time') >= 0) & (F.col('song_length') >= 0) \\\n",
    "                              & (F.col('play_time') <= F.col('song_length')))\n",
    "# df_profile.describe().show()\n",
    "df_profile.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "# feature: play_time/song_length\n",
    "def relative_freq_reco_generation_time_windows(df,time_window_list,snapshot_date):\n",
    "    \"\"\"\n",
    "    generate counts of songs play 80% of their song length (using play_ds data) for different time window\n",
    "    \"\"\"\n",
    "    df_feature = df \\\n",
    "        .groupBy('uid', 'song_id') \\\n",
    "        .agg(*[F.sum(F.when((F.col('date')>=snapshot_date-datetime.timedelta(time_window-1)) \\\n",
    "                            & (F.col('date')<=snapshot_date) \\\n",
    "                            & (F.col('play_time')/F.col('song_length')>=0.8),1).otherwise(0))\\\n",
    "               .alias('rela_freq_last_'+str(time_window)) \\\n",
    "                for time_window in time_window_list]\n",
    "            )# *[] opens list and make them comma separated\n",
    "    return df_feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: int, song_id: int, rela_freq_last_1: bigint, rela_freq_last_3: bigint, rela_freq_last_7: bigint, rela_freq_last_14: bigint, rela_freq_last_30: bigint, rela_freq_last_44: bigint]"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "time_window_list = [1,3,7,14,30,44]\n",
    "snapshot_date = reco_window_end_date\n",
    "df_rela_freq_reco = relative_freq_reco_generation_time_windows(df_profile, time_window_list, snapshot_date)\n",
    "df_rela_freq_reco"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "### Join all feature datasets together"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "def join_reco_data(df_reco_list):\n",
    "    df_master = df_reco_list[0]\n",
    "    df_reco_list.pop(0)\n",
    "    for df_reco in df_reco_list:\n",
    "        df_master = df_master.join(df_reco,on=['uid','song_id'],how='left')\n",
    "        #df_master.persist() # uncomment if number of joins is too many\n",
    "    return df_master    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[uid: int, song_id: int, freq_P_last_1: bigint, freq_P_last_3: bigint, freq_P_last_7: bigint, freq_P_last_14: bigint, freq_P_last_30: bigint, freq_P_last_44: bigint, freq_D_last_1: bigint, freq_D_last_3: bigint, freq_D_last_7: bigint, freq_D_last_14: bigint, freq_D_last_30: bigint, freq_D_last_44: bigint, rec_P: int, rec_D: int, rela_freq_last_1: bigint, rela_freq_last_3: bigint, rela_freq_last_7: bigint, rela_freq_last_14: bigint, rela_freq_last_30: bigint, rela_freq_last_44: bigint]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_reco_list_copy = df_reco_list.copy()\n",
    "df_reco_list_copy.append(df_rela_freq_reco)\n",
    "df_reco_final = join_reco_data(df_reco_list_copy)\n",
    "df_reco_final"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_reco_final.fillna(0).toPandas().to_csv('../data/df_reco_final.csv',index=False)"
   ]
  },
  {
   "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}