Explore&Clean&Label&Downsample_play.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Explore, Clean the whole_play file using spark\n",
    "### Then\n",
    "- clean dirty data, drop outliers and unneccesary variables for churn prediction\n",
    "- filter out the 'robot'/'test account' from the user\n",
    "- give churn label to each valid user\n",
    "- downsampling among valid user\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import pyspark libs and initialze SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "#sparksession is used to create DF, register DF as tables, excute SQL over tables, cache tables\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "from pyspark.sql import Row\n",
    "from pyspark.sql import functions\n",
    "\n",
    "from pyspark.sql.functions import *\n",
    "import pyspark.sql.functions as func\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "from operator import add\n",
    "from operator import ge\n",
    "import sys"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "#import matplotlib.pyplot as plt\n",
    "#import seaborn as sns\n",
    "#%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Next, create pyspark.sql.SparkSession, which is the main entry point for DataFrame and SQL functionality\n",
    "#initialize\n",
    "'''\n",
    "class Builder\n",
    "Builder for SparkSession.\n",
    "\n",
    "appName(name)\n",
    "Sets a name for the application, which will be shown in the Spark web UI.\n",
    "\n",
    "If no application name is set, a randomly generated name will be used.\n",
    "\n",
    "Parameters:\tname – an application name\n",
    "New in version 2.0.\n",
    "\n",
    "config(key=None, value=None, conf=None)\n",
    "Sets a config option. Options set using this method are automatically propagated to both SparkConf and SparkSession‘s own configuration.\n",
    "\n",
    "For an existing SparkConf, use conf parameter.\n",
    "'''\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Python Spark SQL\") \\\n",
    "    .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load in play_file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "#load in date file, textFile will return RDD\n",
    "rdd = sc.textFile(\"D:/MusicFile/all_play_1.log.fn\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "164667319"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#How many rows before modifying the format?\n",
    "rdd.count()\n",
    "#might take almost several mins to run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['264715 \\t 20170301_play.log',\n",
       " '290416 \\t 20170301_play.log',\n",
       " '154422682 \\tar \\t20870993 \\t1 \\t用情 \\t狮子合唱团 \\t22013 \\t332 \\t0 \\t 20170301_play.log',\n",
       " '154421907 \\tip \\t6560858 \\t0 \\t表情不要悲伤 \\t伯贤&D.O.&张艺兴&朴灿烈 \\t96 \\t161 \\t0 \\t 20170301_play.log',\n",
       " \"154422630 \\tar \\t3385963 \\t1 \\tBaby, Don't Cry(人鱼的眼泪) \\tEXO \\t235868 \\t235 \\t0 \\t 20170301_play.log\",\n",
       " '154410267 \\tar \\t6777172 \\t0 \\t3D-环绕音律1(3D Mix) \\tMcTaiM \\t164 \\t237 \\t0 \\t 20170301_play.log',\n",
       " '154407793 \\tar \\t19472465 \\t0 \\t刚好遇见你 \\t曲肖冰 \\t24 \\t201 \\t0 \\t 20170301_play.log',\n",
       " '154422626 \\tar \\t3198036 \\t1 \\t只唱给你听 \\tSpeXial \\t275249 \\t0 \\t0 \\t 20170301_play.log',\n",
       " '154422681 \\tar \\t891952 \\t0 \\t老男孩-(电影《老男孩》主题曲) \\t筷子兄弟 \\t300 \\t300 \\t0 \\t 20170301_play.log',\n",
       " '154408091 \\tar \\t4623962 \\t0 \\t预谋 许佳慧 \\t网络歌手 \\t243 \\t243 \\t0 \\t 20170301_play.log']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#take a look at the format\n",
    "rdd.take(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice the first two rows contains nothing, at with wrong column number.\n",
    "Each row is separated by \"\\t\", with 10 fields, with some column type as float, some as string\n",
    "\n",
    "If use **sc.read.csv(path =..., sep = \"\\t\", schema = schema**, it will cause mislocation(spark ignore missing value and continue reading the next field as the value of this field\n",
    "\n",
    "Also can use Pandas to load in file directly as DataFrame, using **error_bad_lines = False**, it will automatically ignore those mislocation rows, but since the whole_play file is >13G, using Pandas cost lots of time\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Split each row by \"\\t\", if there are 10 fields with correct type, \n",
    "#keep them, if not, skip it, also give the col_name\n",
    "def parseLine(line):\n",
    "    fields = line.split(\"\\t\")\n",
    "    if len(fields) == 10:\n",
    "        try: \n",
    "            uid = float(fields[0])\n",
    "            device = str(fields[1])\n",
    "            song_id = float(fields[2])\n",
    "            song_type = float(fields[3])\n",
    "            song_name = str(fields[4])\n",
    "            singer = str(fields[5])\n",
    "            play_time = float(fields[6])\n",
    "            song_length = float(fields[7])\n",
    "            paid_flag = float(fields[8])\n",
    "            file_name = str(fields[9])\n",
    "            return Row(uid, device, song_id, song_type, song_name, singer, \n",
    "                       play_time, song_length, paid_flag, file_name)\n",
    "        except:\n",
    "            return Row(None)\n",
    "    else:\n",
    "        return Row(None)\n",
    "\n",
    "# Create Row entries that specify column name, to prepare the RDD to convert it to a DataFrame\n",
    "# Always important to filter on field length after splitting, to avoid \"index out of range error\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# schema is like the setting parameters for each column\n",
    "# Spark SQL StructType is the data type representing rows. \n",
    "# A StructType object comprises a list of StructField, which represents a field in a StructType\n",
    "# StructField(\"col_name\", dataType, nullable)\n",
    "\n",
    "\n",
    "schema = StructType([StructField('uid', FloatType(), False),\n",
    "                     StructField('device', StringType(), True),\n",
    "                     StructField('song_id', FloatType(), False),\n",
    "                     StructField('song_type', FloatType(), True),\n",
    "                     StructField('song_name', StringType(), True),\n",
    "                     StructField('singer', StringType(), True),\n",
    "                     StructField('play_time', FloatType(), False),\n",
    "                     StructField('song_length', FloatType(), True),\n",
    "                     StructField('paid_flag', FloatType(), True),\n",
    "                     StructField('file_name', StringType(), True),])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(schema)\n",
    "#only rows satisfying schema will be kept after filtering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "songs = rdd.map(parseLine).filter(lambda x: len(x) == len(schema))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "#cache means temperorily save in the memory\n",
    "song_df = spark.createDataFrame(songs, schema).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "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>uid</th>\n",
       "      <th>device</th>\n",
       "      <th>song_id</th>\n",
       "      <th>song_type</th>\n",
       "      <th>song_name</th>\n",
       "      <th>singer</th>\n",
       "      <th>play_time</th>\n",
       "      <th>song_length</th>\n",
       "      <th>paid_flag</th>\n",
       "      <th>file_name</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>154422688.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>20870992.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>用情</td>\n",
       "      <td>狮子合唱团</td>\n",
       "      <td>22013.0</td>\n",
       "      <td>332.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>154421904.0</td>\n",
       "      <td>ip</td>\n",
       "      <td>6560858.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>表情不要悲伤</td>\n",
       "      <td>伯贤&amp;D.O.&amp;张艺兴&amp;朴灿烈</td>\n",
       "      <td>96.0</td>\n",
       "      <td>161.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>154422624.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>3385963.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>Baby, Don't Cry(人鱼的眼泪)</td>\n",
       "      <td>EXO</td>\n",
       "      <td>235868.0</td>\n",
       "      <td>235.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>154410272.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>6777172.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>3D-环绕音律1(3D Mix)</td>\n",
       "      <td>McTaiM</td>\n",
       "      <td>164.0</td>\n",
       "      <td>237.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>154407792.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>19472464.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>刚好遇见你</td>\n",
       "      <td>曲肖冰</td>\n",
       "      <td>24.0</td>\n",
       "      <td>201.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>154422624.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>3198036.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>只唱给你听</td>\n",
       "      <td>SpeXial</td>\n",
       "      <td>275249.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>154422688.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>891952.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>老男孩-(电影《老男孩》主题曲)</td>\n",
       "      <td>筷子兄弟</td>\n",
       "      <td>300.0</td>\n",
       "      <td>300.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>154408096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>4623962.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>预谋 许佳慧</td>\n",
       "      <td>网络歌手</td>\n",
       "      <td>243.0</td>\n",
       "      <td>243.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>154422576.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>703750.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>爸爸妈妈听我说</td>\n",
       "      <td>儿童歌曲</td>\n",
       "      <td>207.0</td>\n",
       "      <td>207.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>154417312.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>6491500.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>Stereo Love</td>\n",
       "      <td>Edward Maya</td>\n",
       "      <td>56.0</td>\n",
       "      <td>184.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20170301_play.log</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           uid device     song_id  song_type                song_name  \\\n",
       "0  154422688.0    ar   20870992.0        1.0                      用情    \n",
       "1  154421904.0    ip    6560858.0        0.0                  表情不要悲伤    \n",
       "2  154422624.0    ar    3385963.0        1.0  Baby, Don't Cry(人鱼的眼泪)    \n",
       "3  154410272.0    ar    6777172.0        0.0        3D-环绕音律1(3D Mix)    \n",
       "4  154407792.0    ar   19472464.0        0.0                   刚好遇见你    \n",
       "5  154422624.0    ar    3198036.0        1.0                   只唱给你听    \n",
       "6  154422688.0    ar     891952.0        0.0        老男孩-(电影《老男孩》主题曲)    \n",
       "7  154408096.0    ar    4623962.0        0.0                  预谋 许佳慧    \n",
       "8  154422576.0    ar     703750.0        0.0                 爸爸妈妈听我说    \n",
       "9  154417312.0    ar    6491500.0        0.0             Stereo Love    \n",
       "\n",
       "             singer  play_time  song_length  paid_flag           file_name  \n",
       "0            狮子合唱团     22013.0        332.0        0.0   20170301_play.log  \n",
       "1  伯贤&D.O.&张艺兴&朴灿烈        96.0        161.0        0.0   20170301_play.log  \n",
       "2              EXO    235868.0        235.0        0.0   20170301_play.log  \n",
       "3           McTaiM       164.0        237.0        0.0   20170301_play.log  \n",
       "4              曲肖冰        24.0        201.0        0.0   20170301_play.log  \n",
       "5          SpeXial    275249.0          0.0        0.0   20170301_play.log  \n",
       "6             筷子兄弟       300.0        300.0        0.0   20170301_play.log  \n",
       "7             网络歌手       243.0        243.0        0.0   20170301_play.log  \n",
       "8             儿童歌曲       207.0        207.0        0.0   20170301_play.log  \n",
       "9      Edward Maya        56.0        184.0        0.0   20170301_play.log  "
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(song_df.take(10), columns=song_df.columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inspect the play.log"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('uid', 'float'),\n",
       " ('device', 'string'),\n",
       " ('song_id', 'float'),\n",
       " ('song_type', 'float'),\n",
       " ('song_name', 'string'),\n",
       " ('singer', 'string'),\n",
       " ('play_time', 'float'),\n",
       " ('song_length', 'float'),\n",
       " ('paid_flag', 'float'),\n",
       " ('file_name', 'string')]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "song_df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#song_df.describe().show()\n",
    "#Not enough space to cache...."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The file size is too large, so the thing before we actually explore other columns is shrink file size based on the most important columns(uid, device, play_time, song_length, file_name)\n",
    "\n",
    "Drop column(song_type, song_name, singer, paid_flag)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Extracting & Exploring & Cleaning\n",
    "Since we are focusing on user churn analysis and prediction, the information about song, singer are useless right now, so drop those columns to shrink the file size\n",
    "\n",
    "Filter rows with incorrect records\n",
    "i.e.\n",
    "some user has extremely high number of play counts;\n",
    "too long play_time; negative play_time/song_length\n",
    "#### Impute song_length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# as noticed at One_file_explore, there might be negative song_length, play_time, \n",
    "#drop them, impute song_length with play time\n",
    "song_df = song_df.filter(song_df.play_time >=0) \\\n",
    "                .filter(song_df.song_length >0) \\\n",
    "                .dropna(how = 'any', subset = ['play_time']).cache()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+------------------+\n",
      "|summary|           play_time|       song_length|\n",
      "+-------+--------------------+------------------+\n",
      "|  count|           139711287|         139711287|\n",
      "|   mean|  204256.21820355425|308.64432725864117|\n",
      "| stddev|5.3620264507477504E8|167713.64520210458|\n",
      "|    min|                 0.0|               1.0|\n",
      "|    max|        1.4934674E12|      1.34396621E9|\n",
      "+-------+--------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#take a look at play_time and song_length\n",
    "song_df.select('play_time', 'song_length').describe().show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------+\n",
      "|         uid|  count|\n",
      "+------------+-------+\n",
      "|   1685126.0|3537754|\n",
      "| 3.7025504E7|3290924|\n",
      "|   1791497.0|2992986|\n",
      "|    497685.0|2496827|\n",
      "|    736305.0|1741557|\n",
      "|    751824.0|1663395|\n",
      "|   1062806.0|1160654|\n",
      "|         0.0| 759309|\n",
      "|   1749320.0| 414280|\n",
      "| 2.8638488E7| 402177|\n",
      "| 4.6532272E7| 402150|\n",
      "|   1679121.0| 379672|\n",
      "|    637650.0| 211764|\n",
      "| 6.4268008E7| 141533|\n",
      "| 3.2166204E7| 127948|\n",
      "| 1.5594824E8| 115974|\n",
      "| 2.6036032E7|  95701|\n",
      "|1.67982848E8|  82605|\n",
      "|    533817.0|  77965|\n",
      "| 1.6517426E7|  68340|\n",
      "+------------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Check uid>>take so much time to run...\n",
    "uid_count = song_df.groupBy('uid').count().orderBy('count',ascending = False)\n",
    "uid_count.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "According to the groupby result, there are some users' records seem abnormal, either with extremely large play times, or total_play_length. Based on Statistical domain knowledge and business sense, these users might be **test account, robot, or account for special function**, so those can be viewed as outliers, do little help for us to analyze real user's behavior. So we truncate user by how many songs they played totally during this period try**95%**(almost +-2sigma)?.\n",
    "\n",
    "or common assumption: average song_length = 4min, a user play at most 8 hours a day, so the total play count for one day is 150, from March 1st to May 12th(drop March 2nd to March 29th), there are 45 days, so the total play time is 5400. That might be the uppper bound for normal user at MusicBox. \n",
    "\n",
    "**Notice: according to the tutor, 95% might be too conservative, maybe try 98%**\n",
    "\n",
    "#### Detect Outliers and clean them"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "141644"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#How many users during this period?\n",
    "song_df.select('uid').distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 169,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "141644"
      ]
     },
     "execution_count": 169,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "uid_count.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "95 percentile of user play time is 3904.0\n"
     ]
    }
   ],
   "source": [
    "count_ceiling = uid_count.approxQuantile('count', [0.98],0)\n",
    "print('98 percentile of user play time is {:0}'.format(count_ceiling[0]))\n",
    "#try 0.98?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.9928765072999916"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#try <5400\n",
    "count_ceiling_1 = uid_count.filter(uid_count['count'] <= 5400)\n",
    "count_ceiling_1.count()/141644"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### if use 2sigma to drop outlier, the max play_time is 3904, if use our assumption 5400, only 0.08% will be treated as outlier. In this situation, I will follow drop 0.08%."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- uid: float (nullable = false)\n",
      " |-- count: long (nullable = false)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(uid=168392096.0, count=5399),\n",
       " Row(uid=168531584.0, count=5398),\n",
       " Row(uid=168281952.0, count=5394),\n",
       " Row(uid=168373024.0, count=5392),\n",
       " Row(uid=168203056.0, count=5391),\n",
       " Row(uid=168345696.0, count=5388),\n",
       " Row(uid=168607088.0, count=5387),\n",
       " Row(uid=168111184.0, count=5384),\n",
       " Row(uid=168947008.0, count=5382),\n",
       " Row(uid=168318656.0, count=5381)]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "uid_count.printSchema()\n",
    "valid_uid = count_ceiling_1.cache()\n",
    "# .toPandas() removed,  for join df purpose below\n",
    "valid_uid.take(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "ename": "TypeError",
     "evalue": "'Column' object is not callable",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-33-a11234ba8c96>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      1\u001b[0m print(\"number of valid users = {0:0}, \\n number of valid plays = {1:.2e}\"\n\u001b[1;32m----> 2\u001b[1;33m       .format(valid_uid.count(), valid_uid[\"count\"].sum()))\n\u001b[0m",
      "\u001b[1;31mTypeError\u001b[0m: 'Column' object is not callable"
     ]
    }
   ],
   "source": [
    "print(\"number of valid users = {0:0}, \\n number of valid plays = {1:.2e}\"\n",
    "      .format(valid_uid.count(), valid_uid[\"count\"].sum()))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save valid uid\n",
    "valid_uid.repartition(1).write.csv('D:/MusicFile/valid_uid_316', header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "140635"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "valid_uid.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('uid', 'float'), ('count', 'bigint')]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "valid_uid.dtypes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The following valid_uid is shared by my teammates, these file maybe different with mine.\n",
    "My valid_uid is 134563, hers is 135426. But that is OK"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "uid    float64\n",
      "dtype: object\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>uid</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>13600000.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>16800000.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>22000000.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>23200000.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>23900000.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          uid\n",
       "0  13600000.0\n",
       "1  16800000.0\n",
       "2  22000000.0\n",
       "3  23200000.0\n",
       "4  23900000.0"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "valid_uid_teammates = pd.read_csv(\"D:/MusicFile/valid_uid_zw.csv\")\n",
    "print(valid_uid_teammates.dtypes)\n",
    "valid_uid_teammates.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "135426"
      ]
     },
     "execution_count": 158,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#how many valid_user of my teammate's?\n",
    "valid_uid_teammates.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Join valid uid, with song_df of necessary columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "valid_uid = pd.read_csv(\"D:/MusicFile/valid_uid/valid_uid_new.csv\")\n",
    "sv = StructType([StructField('uid', FloatType(), False),\n",
    "                 StructField('count', IntegerType(), False)])\n",
    "valid_uid = spark.createDataFrame(valid_uid, sv).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "song_valid_df = song_df.join(valid_uid, on = 'uid', how = 'inner') \\\n",
    "                       .select('uid', 'device', 'song_id', 'song_type','play_time', 'song_length', 'file_name') \\\n",
    "                    .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------+---------+---------+---------+-----------+--------------------+\n",
      "|        uid|device|  song_id|song_type|play_time|song_length|           file_name|\n",
      "+-----------+------+---------+---------+---------+-----------+--------------------+\n",
      "|1.3586118E7|    ar| 157799.0|      0.0|     31.0|      258.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|9561562.0|      0.0|      6.0|      253.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 237843.0|      0.0|      4.0|      210.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|9561562.0|      0.0|      0.0|      253.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|9561562.0|      0.0|      4.0|      253.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 157671.0|      0.0|      1.0|      203.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 157671.0|      0.0|      0.0|      203.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 157671.0|      0.0|      1.0|      203.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 157719.0|      0.0|    237.0|      237.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 237843.0|      0.0|    210.0|      210.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|9561562.0|      0.0|     44.0|      253.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 157799.0|      0.0|     25.0|      258.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 157719.0|      0.0|    236.0|      237.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 237843.0|      0.0|    209.0|      210.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|9561562.0|      0.0|     15.0|      253.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 590383.0|      0.0|     33.0|      192.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|6546492.0|      0.0|      5.0|      227.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar| 590383.0|      0.0|      1.0|      192.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|6546492.0|      0.0|      0.0|      227.0| 20170330_3_play.log|\n",
      "|1.3586118E7|    ar|6546492.0|      0.0|      0.0|      227.0| 20170330_3_play.log|\n",
      "+-----------+------+---------+---------+---------+-----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "song_valid_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "137975112"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "song_valid_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "140635"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "song_valid_df.select('uid').distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filtered whole_play.log by valid uid, we can now see there are 137975112 play records now. Compared to 164667319, almost **16%** is dropped. Which is acceptable, since we drop 5% users, and those users are responsible for 16% play records.\n",
    "Now, extract the date information so that we can analysis the trends and patterns of user behavior.\n",
    "#### Clean file name, get time feature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "song_valid_df2 = song_valid_df.withColumn(\"device\", trim(song_valid_df.device)) \\\n",
    "                               .withColumn('date_str', trim(song_valid_df.file_name.substr(1,9))) \\\n",
    "                               .withColumn('date_string', regexp_replace('date_str', '20170339', '20170329')) \\\n",
    "                               .withColumn(\"unix_date\", unix_timestamp('date_string', 'yyyyMMdd')) \\\n",
    "                               .withColumn(\"date\", from_unixtime('unix_date').cast(DateType())) \\\n",
    "                               .drop('date_str') \\\n",
    "                               .drop('date_string') \\\n",
    "                               .drop('unix_date')\n",
    "                               \n",
    "#.withColumn means adding columns\n",
    "#trim() truncate blanks, which is a sql function\n",
    "#change string to date: detect incorrect record, replace it. then "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#way to deal with date variable\n",
    "add_date = song_valid_df.withColumn('date_str', trim(song_valid_df.file_name).substr(1,9)) \\ \n",
    "                        .withColumn('date_string', regexp_replace('date_str', '20170339', '20170329')) \\\n",
    "                        .withColumn('date', from_unixtime(unix_timestamp('date_string', 'yyyyMMdd')) \\\n",
    "                                    .alias(\"date\").cast(DateType()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Although we filtered incorrect format rows when we parseLine the entire file, each fields also contains blank at the head and end, for example, device, and file_name. So truncate them first in order to detect them more accurately.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "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>uid</th>\n",
       "      <th>device</th>\n",
       "      <th>song_id</th>\n",
       "      <th>song_type</th>\n",
       "      <th>play_time</th>\n",
       "      <th>song_length</th>\n",
       "      <th>file_name</th>\n",
       "      <th>date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157799.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>31.0</td>\n",
       "      <td>258.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>9561562.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>6.0</td>\n",
       "      <td>253.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>237843.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>210.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>9561562.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>253.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>9561562.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>253.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157671.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>203.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157671.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>203.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157671.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>203.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157719.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>237.0</td>\n",
       "      <td>237.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>237843.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>210.0</td>\n",
       "      <td>210.0</td>\n",
       "      <td>20170330_3_play.log</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          uid device    song_id  song_type  play_time  song_length  \\\n",
       "0  13586118.0     ar   157799.0        0.0       31.0        258.0   \n",
       "1  13586118.0     ar  9561562.0        0.0        6.0        253.0   \n",
       "2  13586118.0     ar   237843.0        0.0        4.0        210.0   \n",
       "3  13586118.0     ar  9561562.0        0.0        0.0        253.0   \n",
       "4  13586118.0     ar  9561562.0        0.0        4.0        253.0   \n",
       "5  13586118.0     ar   157671.0        0.0        1.0        203.0   \n",
       "6  13586118.0     ar   157671.0        0.0        0.0        203.0   \n",
       "7  13586118.0     ar   157671.0        0.0        1.0        203.0   \n",
       "8  13586118.0     ar   157719.0        0.0      237.0        237.0   \n",
       "9  13586118.0     ar   237843.0        0.0      210.0        210.0   \n",
       "\n",
       "              file_name        date  \n",
       "0   20170330_3_play.log  2017-03-30  \n",
       "1   20170330_3_play.log  2017-03-30  \n",
       "2   20170330_3_play.log  2017-03-30  \n",
       "3   20170330_3_play.log  2017-03-30  \n",
       "4   20170330_3_play.log  2017-03-30  \n",
       "5   20170330_3_play.log  2017-03-30  \n",
       "6   20170330_3_play.log  2017-03-30  \n",
       "7   20170330_3_play.log  2017-03-30  \n",
       "8   20170330_3_play.log  2017-03-30  \n",
       "9   20170330_3_play.log  2017-03-30  "
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(song_valid_df2.take(10), columns = song_valid_df2.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "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>uid</th>\n",
       "      <th>device</th>\n",
       "      <th>song_id</th>\n",
       "      <th>song_type</th>\n",
       "      <th>play_time</th>\n",
       "      <th>song_length</th>\n",
       "      <th>date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157799.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>31.0</td>\n",
       "      <td>258.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>9561562.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>6.0</td>\n",
       "      <td>253.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>237843.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>210.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>9561562.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>253.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>9561562.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>253.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157671.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>203.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157671.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>203.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157671.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>203.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>157719.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>237.0</td>\n",
       "      <td>237.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>237843.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>210.0</td>\n",
       "      <td>210.0</td>\n",
       "      <td>2017-03-30</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          uid device    song_id  song_type  play_time  song_length        date\n",
       "0  13586118.0     ar   157799.0        0.0       31.0        258.0  2017-03-30\n",
       "1  13586118.0     ar  9561562.0        0.0        6.0        253.0  2017-03-30\n",
       "2  13586118.0     ar   237843.0        0.0        4.0        210.0  2017-03-30\n",
       "3  13586118.0     ar  9561562.0        0.0        0.0        253.0  2017-03-30\n",
       "4  13586118.0     ar  9561562.0        0.0        4.0        253.0  2017-03-30\n",
       "5  13586118.0     ar   157671.0        0.0        1.0        203.0  2017-03-30\n",
       "6  13586118.0     ar   157671.0        0.0        0.0        203.0  2017-03-30\n",
       "7  13586118.0     ar   157671.0        0.0        1.0        203.0  2017-03-30\n",
       "8  13586118.0     ar   157719.0        0.0      237.0        237.0  2017-03-30\n",
       "9  13586118.0     ar   237843.0        0.0      210.0        210.0  2017-03-30"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "song_cleaned_df = song_valid_df2.drop('file_name').cache()\n",
    "pd.DataFrame(song_cleaned_df.take(10), columns = song_cleaned_df.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "137975112"
      ]
     },
     "execution_count": 154,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "song_cleaned_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Churn Label\n",
    "Using records from 2017-03-01 to 2017-05-12 as train/test dataset, the last two week(2017-04-29~ 2017-05-12) as churn window, which means if user show no activity during the last two weeks, then we label them as Churn user.\n",
    "\n",
    "**Potential idea about how to define churn**\n",
    "1. According to their average play time trend(increasing, decreasing, stable)\n",
    "2. Maybe \"reduce 80% play_time within 2-week\" instead of \"totally non-active\"\n",
    "3. Since we have other information such as search.log, download.log, combine them together to see the churn condition << Inspired by RFM model, we can weighted score play, download, search by quantile, higher score means more active user, lower score means large possibility of churn.\n",
    "\n",
    "Active user is the one who has play record during the last two week, non-record user are labeled as churn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "active_uid = song_cleaned_df.filter(song_cleaned_df.date >= '2017-04-29') \\\n",
    "                           .select(song_cleaned_df.uid.alias('active_uid')) \\\n",
    "                           .distinct()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "89216"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#valid_uid.count() 146035\n",
    "active_uid.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save the active_uid\n",
    "active_uid.repartition(1).write.csv(\"D:/MusicFile/active_uid\", header = True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Add label"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "uid_label = active_uid.join(valid_uid, valid_uid.uid == active_uid.active_uid, \"left_outer\")\n",
    "uid_label = uid_label.withColumn('Churn', uid_label.active_uid.isNull().astype(IntegerType())).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "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>uid</th>\n",
       "      <th>count</th>\n",
       "      <th>active_uid</th>\n",
       "      <th>Churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>445</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>16844004.0</td>\n",
       "      <td>2</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>22030996.0</td>\n",
       "      <td>186</td>\n",
       "      <td>22030996.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>23232528.0</td>\n",
       "      <td>1</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>23885908.0</td>\n",
       "      <td>42</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          uid  count  active_uid  Churn\n",
       "0  13586118.0    445         NaN      1\n",
       "1  16844004.0      2         NaN      1\n",
       "2  22030996.0    186  22030996.0      0\n",
       "3  23232528.0      1         NaN      1\n",
       "4  23885908.0     42         NaN      1"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(uid_label.take(5), columns = uid_label.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [],
   "source": [
    "#drop count and active_uid\n",
    "uid_label = uid_label.drop('count').drop('active_uid')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save the output, so that later when continuing work, just read in this csv\n",
    "uid_label.select('uid', 'Churn').repartition(1).write.csv(\"D:/MusicFile/uid_label_new\", header = True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Downsampling\n",
    "Aim: shrink file size, balance churn/active precentage\n",
    "Method: randomly select 5% from uid(with churn label), according to the original churn/active rate: 5/8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "uid_label = pd.read_csv(\"D:/MusicFile/uid_label_new.csv\")\n",
    "#convert into spark dataframe\n",
    "sv = StructType([StructField('uid', FloatType(), False),\n",
    "                 StructField('churn', IntegerType(), False)])\n",
    "uid_label = spark.createDataFrame(uid_label, sv).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('uid', 'float'), ('churn', 'int')]\n"
     ]
    }
   ],
   "source": [
    "print(uid_label.dtypes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "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>uid</th>\n",
       "      <th>churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>13586118.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>16844004.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>22030996.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>23232528.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>23885908.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>32366964.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>32962054.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>36249280.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>39197032.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>39567468.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          uid  churn\n",
       "0  13586118.0      1\n",
       "1  16844004.0      1\n",
       "2  22030996.0      0\n",
       "3  23232528.0      1\n",
       "4  23885908.0      1\n",
       "5  32366964.0      1\n",
       "6  32962054.0      0\n",
       "7  36249280.0      1\n",
       "8  39197032.0      1\n",
       "9  39567468.0      0"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(uid_label.take(10), columns = uid_label.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|Churn|count|\n",
      "+-----+-----+\n",
      "|    1|51419|\n",
      "|    0|89216|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "uid_label.groupBy('Churn').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Active/Churn: 1.735078472938019\n"
     ]
    }
   ],
   "source": [
    "print(\"Active/Churn:\", 89216/51419)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are 8/5 active user/churn user, which is not that balance, applying downsampling technique to balance positive/negative and reduce file size.\n",
    "![Image of Downsampling](\"D:/MusicFile/downsampling.PNG)\n",
    "\n",
    "Now, since we are the first time, downsample to around 5%"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+\n",
      "|Churn|count|\n",
      "+-----+-----+\n",
      "|    1| 4358|\n",
      "|    0| 4504|\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sampled_uid = uid_label.sampleBy('Churn', fractions = {1:0.086, 0:0.05})\n",
    "sampled_uid.groupBy('Churn').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 198,
   "metadata": {},
   "outputs": [],
   "source": [
    "sampled_uid = sampled_uid.select('uid', \"Churn\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Summary of the amount change of user\n",
    "**original user** is 141644 <p>↓  \n",
    "**valid user** is 140635 with **active user** 89216, **churn user** 51419<p>↓  \n",
    "**downsampled user** is 8862, with **active** 4504, **churn** 4358"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "sampled_uid = pd.read_csv(\"D:/MusicFile/sampled_uid_8862.csv\")\n",
    "su = StructType([StructField('uid', FloatType(), False)])\n",
    "sampled_uid = spark.createDataFrame(sampled_uid, su).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_uid_label = sampled_uid.join(uid_label, sampled_uid.uid == uid_label.uid, how = 'left') \\\n",
    "                              .drop(sampled_uid.uid) \\\n",
    "                              .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "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>uid</th>\n",
       "      <th>churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>154422848.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>154489312.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>154502640.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>154504752.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>154570608.0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           uid  churn\n",
       "0  154422848.0      1\n",
       "1  154489312.0      1\n",
       "2  154502640.0      1\n",
       "3  154504752.0      1\n",
       "4  154570608.0      1"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pd.DataFrame(sample_uid_label.take(5), columns = sample_uid_label.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('uid', 'float'), ('churn', 'int')]"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_uid_label.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8862"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_uid_label.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.06301418565790877"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(4358+4504)/140635\n",
    "#1.735*0.05"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "#save sample_uid_label\n",
    "sample_uid_label.repartition(1).write.csv('D:/MusicFile/sample_uid_label', header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we sampled 8862 user as around 50% from churn, 50% from active, which is 6% of the original user pool. The following analysis and predictive model building will be based on these sampled user.\n",
    "\n",
    "## Extract all records(play, search, download) of sampled user\n",
    "Using join to filter play.log of sampled user from the original cleaned play_df, then check how those sampled user acted during the entire period\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 205,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Timedelta('72 days 00:00:00')"
      ]
     },
     "execution_count": 205,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import datediff, to_date, lit\n",
    "date_diff = pd.to_datetime('2017-05-12') - pd.to_datetime(\"2017-03-01\")\n",
    "date_diff"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|      date|\n",
      "+----------+\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "|2017-03-01|\n",
      "+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#check the date.dtype\n",
    "#song_valid_df2.schema\n",
    "#song_valid_df2.show(10)\n",
    "song_valid_df2.select(\"date\").show(10)\n",
    "#why column is not callable?? Pyspark is not like python, using select\n",
    "#song_valid_df2.date.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "#from operator import ge\n",
    "#s_date = \"2017-03-01\"\n",
    "#e_date = \"2017-04-29\"\n",
    "#date_end = to_date(date).cast(TimestampType())\n",
    "#use data before snapshot_time as training set,\n",
    "#data after snapshot_time as test set \n",
    "sample_play = song_cleaned_df.filter(song_cleaned_df[\"date\"] < \"2017-04-29\").join(sample_uid_label, sample_uid_label.uid == song_cleaned_df.uid, how = 'inner') \\\n",
    "                            .drop(sample_uid_label.uid) \\\n",
    "                            .cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4955455"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#sample_play involves log data before 2017-04-29\n",
    "sample_play.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [],
   "source": [
    "#in entire period(3-01 to 5-12)\n",
    "sampled_play = song_cleaned_df.join(sampled_uid, sampled_uid.uid == song_cleaned_df.uid, how = 'inner') \\\n",
    "                            .drop(sampled_uid.uid) \\\n",
    "                            .cache()\n",
    "#done 3/7"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6118322"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#including 2017--12, there are 180000 more records\n",
    "sampled_play.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#test data just need valid_uid, and records after 4.29 can all be used as test data\n",
    "test_play = song_cleaned_df.filter(song_cleaned_df['date']>= '2017-04-29').join(sample_uid_label, sample_uid_label.uid == song_cleaned_df.uid, how = 'inner') \\\n",
    "                        .drop(sample_uid_label.uid)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_play.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "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>uid</th>\n",
       "      <th>device</th>\n",
       "      <th>song_id</th>\n",
       "      <th>play_time</th>\n",
       "      <th>song_length</th>\n",
       "      <th>date</th>\n",
       "      <th>churn</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>203139.0</td>\n",
       "      <td>204.0</td>\n",
       "      <td>270.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>133946.0</td>\n",
       "      <td>165.0</td>\n",
       "      <td>286.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>380389.0</td>\n",
       "      <td>40.0</td>\n",
       "      <td>233.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>23534240.0</td>\n",
       "      <td>115.0</td>\n",
       "      <td>378.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>6672344.0</td>\n",
       "      <td>69.0</td>\n",
       "      <td>249.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>6769715.0</td>\n",
       "      <td>123.0</td>\n",
       "      <td>124.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>6989315.0</td>\n",
       "      <td>80.0</td>\n",
       "      <td>84.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>7175883.0</td>\n",
       "      <td>138.0</td>\n",
       "      <td>139.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>7353175.0</td>\n",
       "      <td>6.0</td>\n",
       "      <td>123.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>49423096.0</td>\n",
       "      <td>ar</td>\n",
       "      <td>921642.0</td>\n",
       "      <td>5.0</td>\n",
       "      <td>243.0</td>\n",
       "      <td>2017-04-02</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          uid device     song_id  play_time  song_length        date  churn\n",
       "0  49423096.0     ar    203139.0      204.0        270.0  2017-04-02      0\n",
       "1  49423096.0     ar    133946.0      165.0        286.0  2017-04-02      0\n",
       "2  49423096.0     ar    380389.0       40.0        233.0  2017-04-02      0\n",
       "3  49423096.0     ar  23534240.0      115.0        378.0  2017-04-02      0\n",
       "4  49423096.0     ar   6672344.0       69.0        249.0  2017-04-02      0\n",
       "5  49423096.0     ar   6769715.0      123.0        124.0  2017-04-02      0\n",
       "6  49423096.0     ar   6989315.0       80.0         84.0  2017-04-02      0\n",
       "7  49423096.0     ar   7175883.0      138.0        139.0  2017-04-02      0\n",
       "8  49423096.0     ar   7353175.0        6.0        123.0  2017-04-02      0\n",
       "9  49423096.0     ar    921642.0        5.0        243.0  2017-04-02      0"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_play = sample_play.drop('song_type')\n",
    "pd.DataFrame(sample_play.take(10), columns = sample_play.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 203,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+--------------------+------------------+-----------------+--------------------+-------------------+\n",
      "|summary| device|             song_id|         play_time|      song_length|                 uid|              Churn|\n",
      "+-------+-------+--------------------+------------------+-----------------+--------------------+-------------------+\n",
      "|  count|2852352|             2852352|           2852352|          2852352|             2852352|            2852352|\n",
      "|   mean|   null|1.022984193691201E14|1596.9657751442098|270.6672415878897|1.6314567413960478E8|  0.276720755362592|\n",
      "| stddev|   null| 2.83027760288859E16|30606.474290019458|306.6561443148601|  1.63940429026185E7|0.44737730058985425|\n",
      "|    min|     ar|                -1.0|               0.0|              1.0|             12333.0|                  0|\n",
      "|    max|     ip|        1.8446744E19|       2.8639388E7|          23978.0|        1.69258736E8|                  1|\n",
      "+-------+-------+--------------------+------------------+-----------------+--------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "##RUN IT DURING SLEEP!!! COST FOREVER\n",
    "#check again of the dataset to make sure every column has no missing value, outliters\n",
    "sampled_play.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## sampled_play including 5-12, sample_play till 4-29\n",
    "### save the sampled_play as csv, so when building prediction model, it can be used directly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "sampled_play.repartition(1).write.csv(\"D:/MusicFile/sampled_play\", header = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_play.repartition(1).write.csv(\"D:/MusicFile/sample_play_new316\", header = True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Some queries about how user behavior changed along with the date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 204,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+\n",
      "|      date| count|\n",
      "+----------+------+\n",
      "|2017-04-24| 22299|\n",
      "|2017-05-05| 30571|\n",
      "|2017-05-09| 31077|\n",
      "|2017-05-04| 31510|\n",
      "|2017-05-11| 32236|\n",
      "|2017-04-14| 32495|\n",
      "|2017-05-12| 32625|\n",
      "|2017-05-10| 32811|\n",
      "|2017-04-11| 33206|\n",
      "|2017-05-07| 33623|\n",
      "|2017-05-08| 33835|\n",
      "|2017-05-02| 34280|\n",
      "|2017-05-06| 34608|\n",
      "|2017-05-03| 35531|\n",
      "|2017-05-01| 36560|\n",
      "|2017-04-29| 37319|\n",
      "|2017-04-30| 38588|\n",
      "|2017-04-20| 39306|\n",
      "|2017-04-28| 39871|\n",
      "|2017-04-26| 40011|\n",
      "|2017-04-25| 40048|\n",
      "|2017-04-18| 40951|\n",
      "|2017-04-27| 41910|\n",
      "|2017-04-21| 42095|\n",
      "|2017-04-22| 42301|\n",
      "|2017-03-29| 42732|\n",
      "|2017-04-23| 43752|\n",
      "|2017-04-19| 43823|\n",
      "|2017-03-09| 44774|\n",
      "|2017-04-17| 46903|\n",
      "|2017-03-08| 47363|\n",
      "|2017-03-07| 47799|\n",
      "|2017-04-16| 49121|\n",
      "|2017-04-15| 50028|\n",
      "|2017-04-13| 50572|\n",
      "|2017-04-03| 53748|\n",
      "|2017-03-06| 53969|\n",
      "|2017-04-12| 54118|\n",
      "|2017-04-10| 54313|\n",
      "|2017-04-09| 57961|\n",
      "|2017-03-05| 58414|\n",
      "|2017-04-07| 63077|\n",
      "|2017-03-04| 63323|\n",
      "|2017-04-05| 63606|\n",
      "|2017-04-06| 64061|\n",
      "|2017-04-08| 64975|\n",
      "|2017-03-03| 67131|\n",
      "|2017-04-04| 84990|\n",
      "|2017-03-02| 96004|\n",
      "|2017-04-02| 97482|\n",
      "|2017-03-30|104881|\n",
      "|2017-04-01|110660|\n",
      "|2017-03-01|140920|\n",
      "|2017-03-31|142185|\n",
      "+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#top 20 dates with largest play_records?\n",
    "sampled_play.groupBy(\"date\").count().orderBy('count').show(80)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 206,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+\n",
      "|      date| count|\n",
      "+----------+------+\n",
      "|2017-03-01|140920|\n",
      "|2017-03-02| 96004|\n",
      "|2017-03-03| 67131|\n",
      "|2017-03-04| 63323|\n",
      "|2017-03-05| 58414|\n",
      "|2017-03-06| 53969|\n",
      "|2017-03-07| 47799|\n",
      "|2017-03-08| 47363|\n",
      "|2017-03-09| 44774|\n",
      "|2017-03-29| 42732|\n",
      "|2017-03-30|104881|\n",
      "|2017-03-31|142185|\n",
      "|2017-04-01|110660|\n",
      "|2017-04-02| 97482|\n",
      "|2017-04-03| 53748|\n",
      "|2017-04-04| 84990|\n",
      "|2017-04-05| 63606|\n",
      "|2017-04-06| 64061|\n",
      "|2017-04-07| 63077|\n",
      "|2017-04-08| 64975|\n",
      "|2017-04-09| 57961|\n",
      "|2017-04-10| 54313|\n",
      "|2017-04-11| 33206|\n",
      "|2017-04-12| 54118|\n",
      "|2017-04-13| 50572|\n",
      "|2017-04-14| 32495|\n",
      "|2017-04-15| 50028|\n",
      "|2017-04-16| 49121|\n",
      "|2017-04-17| 46903|\n",
      "|2017-04-18| 40951|\n",
      "|2017-04-19| 43823|\n",
      "|2017-04-20| 39306|\n",
      "|2017-04-21| 42095|\n",
      "|2017-04-22| 42301|\n",
      "|2017-04-23| 43752|\n",
      "|2017-04-24| 22299|\n",
      "|2017-04-25| 40048|\n",
      "|2017-04-26| 40011|\n",
      "|2017-04-27| 41910|\n",
      "|2017-04-28| 39871|\n",
      "|2017-04-29| 37319|\n",
      "|2017-04-30| 38588|\n",
      "|2017-05-01| 36560|\n",
      "|2017-05-02| 34280|\n",
      "|2017-05-03| 35531|\n",
      "|2017-05-04| 31510|\n",
      "|2017-05-05| 30571|\n",
      "|2017-05-06| 34608|\n",
      "|2017-05-07| 33623|\n",
      "|2017-05-08| 33835|\n",
      "|2017-05-09| 31077|\n",
      "|2017-05-10| 32811|\n",
      "|2017-05-11| 32236|\n",
      "|2017-05-12| 32625|\n",
      "+----------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#from 2017-03-01 to 2017-04-29, what's the play trend?\n",
    "sampled_play.groupBy(\"date\").count().orderBy('date').show(80)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 228,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------------+--------------------+--------------+--------------+\n",
      "|Churn|    avg(play_time)|      sum(play_time)|min(play_time)|max(play_time)|\n",
      "+-----+------------------+--------------------+--------------+--------------+\n",
      "|    1| 2838.889162910153|2.2407494107307982E9|           0.0|   2.8639388E7|\n",
      "|    0|1121.8159896179477| 2.314359111933338E9|           0.0|   2.2684604E7|\n",
      "+-----+------------------+--------------------+--------------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "churn_playtime = sampled_play.groupBy(sampled_play.Churn) \\\n",
    "             .agg(func.avg('play_time'), func.sum(\"play_time\"), func.min(\"play_time\"),func.max(\"play_time\"))\n",
    "churn_playtime.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 235,
   "metadata": {},
   "outputs": [
    {
     "ename": "AttributeError",
     "evalue": "'DataFrame' object has no attribute 'bar'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-235-e096c666eba2>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[0max\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mfig\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0madd_subplot\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m111\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      3\u001b[0m \u001b[0mchurn_playtime\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcolumns\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 4\u001b[1;33m \u001b[0mchurn_playtime\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mbar\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mx\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;34m\"Churn\"\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0my\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;33m[\u001b[0m\u001b[1;34m'avg(play_time)'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'sum(play_time)'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'min(play_time)'\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;34m'max(play_time)'\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32mC:\\spark\\spark-2.2.1-bin-hadoop2.7_1\\python\\pyspark\\sql\\dataframe.py\u001b[0m in \u001b[0;36m__getattr__\u001b[1;34m(self, name)\u001b[0m\n\u001b[0;32m   1018\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mname\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mcolumns\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1019\u001b[0m             raise AttributeError(\n\u001b[1;32m-> 1020\u001b[1;33m                 \"'%s' object has no attribute '%s'\" % (self.__class__.__name__, name))\n\u001b[0m\u001b[0;32m   1021\u001b[0m         \u001b[0mjc\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jdf\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mapply\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1022\u001b[0m         \u001b[1;32mreturn\u001b[0m \u001b[0mColumn\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mjc\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAttributeError\u001b[0m: 'DataFrame' object has no attribute 'bar'"
     ]
    },
    {
     "data": {
      "image/png": "iVBORw0KGgoAAAANSUhEUgAAAXwAAAD8CAYAAAB0IB+mAAAABHNCSVQICAgIfAhkiAAAAAlwSFlzAAALEgAACxIB0t1+/AAAADl0RVh0U29mdHdhcmUAbWF0cGxvdGxpYiB2ZXJzaW9uIDIuMS4yLCBodHRwOi8vbWF0cGxvdGxpYi5vcmcvNQv5yAAADYBJREFUeJzt3HGI33d9x/Hny8ROprWO5QRJou1YuhrKoO7oOoRZ0Y20fyT/FEmguEppwK0OZhE6HCr1rylDELJptolT0Fr9Qw+J5A9X6RAjudJZmpTALTpzROhZu/5TtGZ774/fT++4XHLf3v3uLt77+YDA7/v7fX6/e+fD3TO/fH/3+6WqkCRtf6/a6gEkSZvD4EtSEwZfkpow+JLUhMGXpCYMviQ1sWrwk3wuyXNJnrnC7Uny6SRzSZ5O8rbJjylJWq8hz/A/Dxy4yu13AfvGf44C/7T+sSRJk7Zq8KvqCeBnV1lyCPhCjZwC3pDkTZMaUJI0GTsn8Bi7gQtLjufH1/1k+cIkRxn9L4DXvva1f3TLLbdM4MtLUh9PPvnkT6tqai33nUTws8J1K35eQ1UdB44DTE9P1+zs7AS+vCT1keS/13rfSfyWzjywd8nxHuDiBB5XkjRBkwj+DPDe8W/r3AG8WFWXnc6RJG2tVU/pJPkycCewK8k88FHg1QBV9RngBHA3MAe8BLxvo4aVJK3dqsGvqiOr3F7AX01sIknShvCdtpLUhMGXpCYMviQ1YfAlqQmDL0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDVh8CWpCYMvSU0YfElqwuBLUhMGX5KaMPiS1ITBl6QmDL4kNWHwJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDUxKPhJDiQ5l2QuycMr3P7mJI8neSrJ00nunvyokqT1WDX4SXYAx4C7gP3AkST7ly37O+CxqroNOAz846QHlSStz5Bn+LcDc1V1vqpeBh4FDi1bU8Drx5dvAC5ObkRJ0iQMCf5u4MKS4/nxdUt9DLg3yTxwAvjASg+U5GiS2SSzCwsLaxhXkrRWQ4KfFa6rZcdHgM9X1R7gbuCLSS577Ko6XlXTVTU9NTX1yqeVJK3ZkODPA3uXHO/h8lM29wOPAVTV94DXALsmMaAkaTKGBP80sC/JTUmuY/Si7MyyNT8G3gWQ5K2Mgu85G0m6hqwa/Kq6BDwInASeZfTbOGeSPJLk4HjZQ8ADSX4AfBm4r6qWn/aRJG2hnUMWVdUJRi/GLr3uI0sunwXePtnRJEmT5DttJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDVh8CWpCYMvSU0YfElqwuBLUhMGX5KaMPiS1ITBl6QmDL4kNWHwJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNDAp+kgNJziWZS/LwFda8J8nZJGeSfGmyY0qS1mvnaguS7ACOAX8GzAOnk8xU1dkla/YBfwu8vapeSPLGjRpYkrQ2Q57h3w7MVdX5qnoZeBQ4tGzNA8CxqnoBoKqem+yYkqT1GhL83cCFJcfz4+uWuhm4Ocl3k5xKcmClB0pyNMlsktmFhYW1TSxJWpMhwc8K19Wy453APuBO4AjwL0necNmdqo5X1XRVTU9NTb3SWSVJ6zAk+PPA3iXHe4CLK6z5RlX9sqp+CJxj9A+AJOkaMST4p4F9SW5Kch1wGJhZtubrwDsBkuxidIrn/CQHlSStz6rBr6pLwIPASeBZ4LGqOpPkkSQHx8tOAs8nOQs8Dnyoqp7fqKElSa9cqpafjt8c09PTNTs7uyVfW5J+UyV5sqqm13Jf32krSU0YfElqwuBLUhMGX5KaMPiS1ITBl6QmDL4kNWHwJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDVh8CWpCYMvSU0YfElqwuBLUhMGX5KaMPiS1ITBl6QmDL4kNWHwJakJgy9JTRh8SWrC4EtSE4OCn+RAknNJ5pI8fJV19ySpJNOTG1GSNAmrBj/JDuAYcBewHziSZP8K664H/hr4/qSHlCSt35Bn+LcDc1V1vqpeBh4FDq2w7uPAJ4CfT3A+SdKEDAn+buDCkuP58XW/luQ2YG9VffNqD5TkaJLZJLMLCwuveFhJ0toNCX5WuK5+fWPyKuBTwEOrPVBVHa+q6aqanpqaGj6lJGndhgR/Hti75HgPcHHJ8fXArcB3kvwIuAOY8YVbSbq2DAn+aWBfkpuSXAccBmZ+dWNVvVhVu6rqxqq6ETgFHKyq2Q2ZWJK0JqsGv6ouAQ8CJ4Fngceq6kySR5Ic3OgBJUmTsXPIoqo6AZxYdt1HrrD2zvWPJUmaNN9pK0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDVh8CWpCYMvSU0YfElqwuBLUhMGX5KaMPiS1ITBl6QmDL4kNWHwJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDVh8CWpCYMvSU0YfElqYlDwkxxIci7JXJKHV7j9g0nOJnk6ybeTvGXyo0qS1mPV4CfZARwD7gL2A0eS7F+27Clguqr+EPga8IlJDypJWp8hz/BvB+aq6nxVvQw8ChxauqCqHq+ql8aHp4A9kx1TkrReQ4K/G7iw5Hh+fN2V3A98a6UbkhxNMptkdmFhYfiUkqR1GxL8rHBdrbgwuReYBj650u1VdbyqpqtqempqaviUkqR12zlgzTywd8nxHuDi8kVJ3g18GHhHVf1iMuNJkiZlyDP808C+JDcluQ44DMwsXZDkNuCzwMGqem7yY0qS1mvV4FfVJeBB4CTwLPBYVZ1J8kiSg+NlnwReB3w1yX8mmbnCw0mStsiQUzpU1QngxLLrPrLk8rsnPJckacJ8p60kNWHwJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNGHxJasLgS1ITBl+SmjD4ktSEwZekJgy+JDVh8CWpCYMvSU0YfElqwuBLUhMGX5KaMPiS1ITBl6QmDL4kNWHwJakJgy9JTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ1YfAlqQmDL0lNDAp+kgNJziWZS/LwCrf/VpKvjG//fpIbJz2oJGl9Vg1+kh3AMeAuYD9wJMn+ZcvuB16oqt8HPgX8/aQHlSStz5Bn+LcDc1V1vqpeBh4FDi1bcwj4t/HlrwHvSpLJjSlJWq+dA9bsBi4sOZ4H/vhKa6rqUpIXgd8Ffrp0UZKjwNHx4S+SPLOWobehXSzbq8bci0XuxSL3YtEfrPWOQ4K/0jP1WsMaquo4cBwgyWxVTQ/4+tuee7HIvVjkXixyLxYlmV3rfYec0pkH9i453gNcvNKaJDuBG4CfrXUoSdLkDQn+aWBfkpuSXAccBmaWrZkB/mJ8+R7g36vqsmf4kqSts+opnfE5+QeBk8AO4HNVdSbJI8BsVc0A/wp8Mckco2f2hwd87ePrmHu7cS8WuReL3ItF7sWiNe9FfCIuST34TltJasLgS1ITGx58P5Zh0YC9+GCSs0meTvLtJG/Zijk3w2p7sWTdPUkqybb9lbwhe5HkPePvjTNJvrTZM26WAT8jb07yeJKnxj8nd2/FnBstyeeSPHel9ypl5NPjfXo6ydsGPXBVbdgfRi/y/hfwe8B1wA+A/cvW/CXwmfHlw8BXNnKmrfozcC/eCfz2+PL7O+/FeN31wBPAKWB6q+fewu+LfcBTwO+Mj9+41XNv4V4cB94/vrwf+NFWz71Be/GnwNuAZ65w+93Atxi9B+oO4PtDHnejn+H7sQyLVt2Lqnq8ql4aH55i9J6H7WjI9wXAx4FPAD/fzOE22ZC9eAA4VlUvAFTVc5s842YZshcFvH58+QYuf0/QtlBVT3D19zIdAr5QI6eANyR502qPu9HBX+ljGXZfaU1VXQJ+9bEM282QvVjqfkb/gm9Hq+5FktuAvVX1zc0cbAsM+b64Gbg5yXeTnEpyYNOm21xD9uJjwL1J5oETwAc2Z7RrzivtCTDsoxXWY2Ify7ANDP57JrkXmAbesaETbZ2r7kWSVzH61NX7NmugLTTk+2Ino9M6dzL6X99/JLm1qv5ng2fbbEP24gjw+ar6hyR/wuj9P7dW1f9t/HjXlDV1c6Of4fuxDIuG7AVJ3g18GDhYVb/YpNk222p7cT1wK/CdJD9idI5yZpu+cDv0Z+QbVfXLqvohcI7RPwDbzZC9uB94DKCqvge8htEHq3UzqCfLbXTw/ViGRavuxfg0xmcZxX67nqeFVfaiql6sql1VdWNV3cjo9YyDVbXmD426hg35Gfk6oxf0SbKL0Sme85s65eYYshc/Bt4FkOStjIK/sKlTXhtmgPeOf1vnDuDFqvrJanfa0FM6tXEfy/AbZ+BefBJ4HfDV8evWP66qg1s29AYZuBctDNyLk8CfJzkL/C/woap6fuum3hgD9+Ih4J+T/A2jUxj3bccniEm+zOgU3q7x6xUfBV4NUFWfYfT6xd3AHPAS8L5Bj7sN90qStALfaStJTRh8SWrC4EtSEwZfkpow+JLUhMGXpCYMviQ18f+GmWq6NWLIwgAAAABJRU5ErkJggg==\n",
      "text/plain": [
       "<matplotlib.figure.Figure at 0xd584630>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "fig = plt.figure()\n",
    "ax = fig.add_subplot(111)\n",
    "churn_playtime.columns\n",
    "churn_playtime.bar(x = \"Churn\", y = ['avg(play_time)', 'sum(play_time)', 'min(play_time)', 'max(play_time)'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 222,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------------+--------------------+----------------+----------------+\n",
      "|Churn|  avg(song_length)|    sum(song_length)|min(song_length)|max(song_length)|\n",
      "+-----+------------------+--------------------+----------------+----------------+\n",
      "|    1|269.10882532989916|2.1240894137701607E8|             1.0|         19732.0|\n",
      "|    0| 271.2634789709998| 5.596293065006843E8|             1.0|         23978.0|\n",
      "+-----+------------------+--------------------+----------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "churn_songlength = sampled_play.groupBy(sampled_play.Churn) \\\n",
    "                               .agg(func.avg('song_length'), func.sum(\"song_length\"), func.min(\"song_length\"),func.max(\"song_length\"))\n",
    "churn_songlength.show()    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Clearly we can see there are large difference between churn/active user about play_time and song_length. Active user spend twice average play time than churn user, "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------+------+\n",
      "|churn_device|     ar|    ip|\n",
      "+------------+-------+------+\n",
      "|           1|1099765|220193|\n",
      "|           0|3928343|870021|\n",
      "+------------+-------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sampled_play.crosstab('churn', 'device').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "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": 2
}