Sparkify_IBM.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Waiting for a Spark session to start...\n",
      "Spark Initialization Done! ApplicationId = app-20200424010717-0000\n",
      "KERNEL_ID = 70043a78-429c-4226-a0a3-4ca6d41334e9\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(artist='Martin Orford', auth='Logged In', firstName='Joseph', gender='M', itemInSession=20, lastName='Morales', length=597.55057, level='free', location='Corpus Christi, TX', method='PUT', page='NextSong', registration=1532063507000, sessionId=292, song='Grand Designs', status=200, ts=1538352011000, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='293'),\n",
       " Row(artist=\"John Brown's Body\", auth='Logged In', firstName='Sawyer', gender='M', itemInSession=74, lastName='Larson', length=380.21179, level='free', location='Houston-The Woodlands-Sugar Land, TX', method='PUT', page='NextSong', registration=1538069638000, sessionId=97, song='Bulls', status=200, ts=1538352025000, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='98'),\n",
       " Row(artist='Afroman', auth='Logged In', firstName='Maverick', gender='M', itemInSession=184, lastName='Santiago', length=202.37016, level='paid', location='Orlando-Kissimmee-Sanford, FL', method='PUT', page='NextSong', registration=1535953455000, sessionId=178, song='Because I Got High', status=200, ts=1538352118000, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='179'),\n",
       " Row(artist=None, auth='Logged In', firstName='Maverick', gender='M', itemInSession=185, lastName='Santiago', length=None, level='paid', location='Orlando-Kissimmee-Sanford, FL', method='PUT', page='Logout', registration=1535953455000, sessionId=178, song=None, status=307, ts=1538352119000, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"', userId='179'),\n",
       " Row(artist='Lily Allen', auth='Logged In', firstName='Gianna', gender='F', itemInSession=22, lastName='Campos', length=194.53342, level='paid', location='Mobile, AL', method='PUT', page='NextSong', registration=1535931018000, sessionId=245, song='Smile (Radio Edit)', status=200, ts=1538352124000, userAgent='Mozilla/5.0 (Windows NT 6.1; WOW64; rv:31.0) Gecko/20100101 Firefox/31.0', userId='246')]"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "import ibmos2spark\n",
    "# @hidden_cell\n",
    "credentials = {\n",
    "    'endpoint': 'https://s3.eu-geo.objectstorage.service.networklayer.com',\n",
    "    'service_id': 'iam-ServiceId-484d9a43-05cb-4321-94b1-11af412e073e',\n",
    "    'iam_service_endpoint': 'https://iam.eu-gb.bluemix.net/oidc/token',\n",
    "    'api_key': 'sLplkCVqbxNnm9CVPbfmHX5XntM-I672AXTIaN2DTg0w'\n",
    "}\n",
    "\n",
    "configuration_name = 'os_758721e72b18404c941eec0bda75f42b_configs'\n",
    "cos = ibmos2spark.CloudObjectStorage(sc, credentials, configuration_name, 'bluemix_cos')\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName('Sparkify').getOrCreate()\n",
    "# Since JSON data can be semi-structured and contain additional metadata, it is possible that you might face issues with the DataFrame layout.\n",
    "# Please read the documentation of 'SparkSession.read()' to learn more about the possibilities to adjust the data loading.\n",
    "# PySpark documentation: http://spark.apache.org/docs/2.0.2/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.json\n",
    "\n",
    "df = spark.read.json(cos.url('medium-sparkify-event-data.json', 'sparkify-donotdelete-pr-mmgfjln3udz4qy'))\n",
    "df.take(5)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import avg, desc, min, max, split, udf\n",
    "from pyspark.sql.types import IntegerType\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
    "from pyspark.ml.feature import CountVectorizer, StandardScaler, StringIndexer, VectorAssembler, OneHotEncoder\n",
    "from pyspark.ml.regression import LinearRegression\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- artist: string (nullable = true)\n",
      " |-- auth: string (nullable = true)\n",
      " |-- firstName: string (nullable = true)\n",
      " |-- gender: string (nullable = true)\n",
      " |-- itemInSession: long (nullable = true)\n",
      " |-- lastName: string (nullable = true)\n",
      " |-- length: double (nullable = true)\n",
      " |-- level: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- method: string (nullable = true)\n",
      " |-- page: string (nullable = true)\n",
      " |-- registration: long (nullable = true)\n",
      " |-- sessionId: long (nullable = true)\n",
      " |-- song: string (nullable = true)\n",
      " |-- status: long (nullable = true)\n",
      " |-- ts: long (nullable = true)\n",
      " |-- userAgent: string (nullable = true)\n",
      " |-- userId: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(artist='Martin Orford', auth='Logged In', firstName='Joseph', gender='M', itemInSession=20, lastName='Morales', length=597.55057, level='free', location='Corpus Christi, TX', method='PUT', page='NextSong', registration=1532063507000, sessionId=292, song='Grand Designs', status=200, ts=1538352011000, userAgent='\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"', userId='293')"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userId|\n",
      "+------+\n",
      "|   293|\n",
      "|    98|\n",
      "|   179|\n",
      "|   179|\n",
      "|   246|\n",
      "|   163|\n",
      "|      |\n",
      "|      |\n",
      "|      |\n",
      "|      |\n",
      "|   179|\n",
      "|   175|\n",
      "|   100|\n",
      "|   100|\n",
      "|   163|\n",
      "|   246|\n",
      "|   179|\n",
      "|    39|\n",
      "|   163|\n",
      "|   179|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('userId').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.where(\"userId != '' and userId != ' '\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.dropna(how='any', subset=['userId', 'sessionId'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "valid = df.dropDuplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userId|\n",
      "+------+\n",
      "|    10|\n",
      "|   100|\n",
      "|100001|\n",
      "|100002|\n",
      "|100003|\n",
      "|100004|\n",
      "|100005|\n",
      "|100006|\n",
      "|100007|\n",
      "|100008|\n",
      "|100009|\n",
      "|100010|\n",
      "|100011|\n",
      "|100012|\n",
      "|100013|\n",
      "|100014|\n",
      "|100015|\n",
      "|100016|\n",
      "|100017|\n",
      "|100018|\n",
      "+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.groupby('userId').max().select('userId').sort('userId').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|                page|\n",
      "+--------------------+\n",
      "|               About|\n",
      "|          Add Friend|\n",
      "|     Add to Playlist|\n",
      "|              Cancel|\n",
      "|Cancellation Conf...|\n",
      "|           Downgrade|\n",
      "|               Error|\n",
      "|                Help|\n",
      "|                Home|\n",
      "|              Logout|\n",
      "|            NextSong|\n",
      "|         Roll Advert|\n",
      "|       Save Settings|\n",
      "|            Settings|\n",
      "|    Submit Downgrade|\n",
      "|      Submit Upgrade|\n",
      "|         Thumbs Down|\n",
      "|           Thumbs Up|\n",
      "|             Upgrade|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.select('page').dropDuplicates().sort('page').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define Churn\n",
    "\n",
    "Create a column `Churn` to use as the label for your model. I use the `Cancellation Confirmation` events to define my churn, which happen for both paid and free users."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|     page|\n",
      "+---------+\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "|Thumbs Up|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "| NextSong|\n",
      "+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.select('page').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "churn = udf(lambda x: 1 if x == 'Cancellation Confirmation' else 0, IntegerType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "valid = valid.withColumn('churn', churn(valid.page))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "99"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "valid.where(\"churn == 1\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+------+\n",
      "|churn|gender| count|\n",
      "+-----+------+------+\n",
      "|    1|     F|    45|\n",
      "|    0|     M|302558|\n",
      "|    1|     M|    54|\n",
      "|    0|     F|225348|\n",
      "+-----+------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "valid.groupby([\"churn\", 'gender']).count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "gender_idx = StringIndexer(inputCol=\"gender\", outputCol='gender_idx')\n",
    "page_idx = StringIndexer(inputCol=\"page\", outputCol='page_idx')\n",
    "user_idx = StringIndexer(inputCol=\"userAgent\", outputCol='user_idx')\n",
    "gender_encoder = OneHotEncoder(inputCol='gender_idx', outputCol='gender_vec')\n",
    "page_encoder = OneHotEncoder(inputCol='page_idx', outputCol='page_vec')\n",
    "user_encoder = OneHotEncoder(inputCol='user_idx', outputCol='user_vec')\n",
    "assembler = VectorAssembler(inputCols=[\"gender_vec\", \"page_vec\", \"status\"], outputCol=\"features\")\n",
    "idx = StringIndexer(inputCol=\"churn\", outputCol=\"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "model =  LogisticRegression(maxIter=15)\n",
    "pipeline = Pipeline(stages=[gender_idx, page_idx, user_idx, gender_encoder, page_encoder, user_encoder, assembler, idx, model])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "rest, validation = valid.randomSplit([0.9, 0.1], seed=42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "param = ParamGridBuilder() \\\n",
    "    .addGrid(model.regParam,[0.0, 0.1, ]) \\\n",
    "    .addGrid(model.elasticNetParam,[0.0, 0.1, 0.5, 1.0]) \\\n",
    "    .build()\n",
    "\n",
    "\n",
    "cross = CrossValidator(estimator=pipeline,\n",
    "                          estimatorParamMaps=param,\n",
    "                          evaluator=MulticlassClassificationEvaluator(),\n",
    "                          numFolds=3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "fitted = cross.fit(rest)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = fitted.transform(validation)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|prediction|\n",
      "+----------+\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "|       0.0|\n",
      "+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "results.select(\"prediction\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1.0,\n",
       " 1.0,\n",
       " 1.0,\n",
       " 1.0,\n",
       " 0.9997190150118527,\n",
       " 0.9997190150118527,\n",
       " 0.9997190150118527,\n",
       " 0.9997190150118527]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fitted.avgMetrics"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.6 with Spark",
   "language": "python3",
   "name": "python36"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}