<jupyter_start><jupyter_text>1. Sparkify Project WorkspaceThis is done based on the analysis with Pandas. So some detailed steps are ommited here.<jupyter_code># import libraries
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt 
% matplotlib inline
import seaborn as sns

import pyspark
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.sql.functions import udf
from pyspark.sql.types import *
from pyspark.sql import functions as F

import datetime
# create a Spark session
configure = SparkConf().setAppName("sparkify").setMaster("local")
import ibmos2spark
# @hidden_cell
credentials = {
    'endpoint': 'https://s3.eu-geo.objectstorage.service.networklayer.com',
    'service_id': 'iam-ServiceId-d5afc86b-deb0-40c5-94fd-816cf79fab75',
    'iam_service_endpoint': 'https://iam.cloud.ibm.com/oidc/token',
    'api_key': '<KEY>'
}

configuration_name = 'os_8bfcc3c030124734958972b41a36cec1_configs'
cos = ibmos2spark.CloudObjectStorage(sc, credentials, configuration_name, 'bluemix_cos')

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
# Since JSON data can be semi-structured and contain additional metadata, it is possible that you might face issues with the DataFrame layout.
# Please read the documentation of 'SparkSession.read()' to learn more about the possibilities to adjust the data loading.
# PySpark documentation: http://spark.apache.org/docs/2.0.2/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.json<jupyter_output><empty_output><jupyter_text>2. Load and Clean DatasetIn this workspace, the mini-dataset file is `mini_sparkify_event_data.json`. Load and clean the dataset, checking for invalid or missing data - for example, records without userids or sessionids.<jupyter_code>df_total = spark.read.json(cos.url('medium-sparkify-event-data.json', 'sparkify-donotdelete-pr-gkwqpiyvqhelzl'))
df_total.take(1)
# backup dataframe
df_total_backup = df_total<jupyter_output><empty_output><jupyter_text>2.1 Remove undesired columns Following columns are removed:- **`artist`**, **`song`** and **`location`**    The behaviour of a R&B fan and a classic music lover can certainly be different. But considering this variance requirs a    database with labeled artists and songs, which is beyond the scope of this project.     Similarly, user from different locations may also behaviour differenetly related with for example the local personal income level. This feature is also left for investigation in future.      - **`firstName`** and **`lastName`**    They are not related with user behaviour.    - **`itemInSession`**, **`method`**, **`sessionId`**, **`status`**    These parameters are simply technical parameters. They are not related with user behaviour.<jupyter_code>df_total = df_total.drop('artist','song','location','firstName','lastName','itemInSession','method','sessionId','status')
df_total.take(1)<jupyter_output><empty_output><jupyter_text>2.2 Understanding and clean the data   2.2.1 Overview<jupyter_code>df_total.printSchema()
df_total.describe().collect()
print(df_total.where(df_total.gender.isNull()).count())
print(df_total.where(df_total.length.isNull()).count())
print(df_total.where(df_total.registration.isNull()).count())
print(df_total.where(df_total.userAgent.isNull()).count())<jupyter_output>15700
110828
15700
15700<jupyter_text>They are all referring to unregistered users, see the results below.<jupyter_code>df_total[df_total.userId == ''].count()<jupyter_output><empty_output><jupyter_text>Those columns are removed. Statistics of the dataframe with remaining columns is shown below:<jupyter_code>df_total = df_total[df_total.userId != '']<jupyter_output><empty_output><jupyter_text>2.2.3 Clean Column `auth` Unique values of `auth`:<jupyter_code>df_total = df_total.drop('auth')<jupyter_output><empty_output><jupyter_text>2.2.4. Column `gender`<jupyter_code># check against null 
df_total.select('gender').distinct().collect()<jupyter_output><empty_output><jupyter_text>There is no empty value. It will be transformed to numeric value using OneHotEnconder later.  2.2.5 Clean Column `length` When not visiting page 'Next song', the value `length` is null. These nulls are as expected, so they are "right" nulls, and do not require removing or imputing.<jupyter_code>df_total.select('length','page').where(df_total.length.isNull()).groupBy('page').count().show()<jupyter_output>+--------------------+-----+
|                page|count|
+--------------------+-----+
|              Cancel|   99|
|    Submit Downgrade|  117|
|         Thumbs Down| 4911|
|                Home|19089|
|           Downgrade| 3811|
|         Roll Advert| 7773|
|              Logout| 5990|
|       Save Settings|  585|
|Cancellation Conf...|   99|
|               About| 1026|
|            Settings| 2964|
|     Add to Playlist|12349|
|          Add Friend| 8087|
|           Thumbs Up|23826|
|                Help| 2644|
|             Upgrade|  968|
|               Error|  503|
|      Submit Upgrade|  287|
+--------------------+-----+<jupyter_text>2.2.6 Clean Column `level`<jupyter_code># check against null
df_total.where(df_total.level.isNull()).count()<jupyter_output><empty_output><jupyter_text>This column has no nulls now. It is used to generate the UBC.  2.2.7 Clean column `page` Unique values:<jupyter_code>list(df_total.select('page').distinct().toPandas().page.values)<jupyter_output><empty_output><jupyter_text>2.2.8 clean column `registration`, 2.2.9 clean column `ts` I leave these two columns as they are to save computation cost. After all, I only care about the difference.   2.2.10 clean column `userAgent` I devide the values into three groups:- when it contains 'Windows', then it is category w- when it contains 'Linux', then it is category l- the rest are category m (ios, mac etc.)For this purpose, a function is defined:<jupyter_code>function = udf(lambda x: 'w' if 'Windows' in x else 'l' if 'Linux' in x else 'm')
df_total = df_total.withColumn('userAgentNew', function('userAgent'))
df_total.show(1)
df_total = df_total.drop('userAgent')
df_total.select('userAgentNew').distinct().show()<jupyter_output>+------------+
|userAgentNew|
+------------+
|           l|
|           m|
|           w|
+------------+<jupyter_text>2.2.11 clean column `userId` Nothing need to be done.  3. Feature Engineering  3.1 Generate user behaviour capsule (UBC) and new features Churn can be identified by the column `level`, which shows the current level of the user (whether 'paid' or 'free'), and the column `page`, which shows the pages the user visits. If a paid user visits 'Submit Downgrade' or 'Cancellation Confirmation', that means the users has churned. As discussed above, the investigation here focus on the categorize following two groups of users:- 1. paided user and churn the service - 2. paided user and does not churn the serviceFor this purpose, I define the so-called "user behaviour capsule" for this two groups of users as below:- 1. from the first row of level "paid" to the first row of page 'Submit Downgrade' or 'Cancellation Confirmation'- 2. from the first row of level "paid" to the end of the data without churnThe bahaviour of free user and their upgrade activities are not the focus in this project. For the users who have downgrades the service several times, only the first time is considered. To generate UBC, a function is defined below.<jupyter_code># sort the data according to `userId` and `ts`
df_total = df_total.sort(['userId','ts'])

# Generate the list of userIds

userId_list = list(df_total.select('userId').distinct().toPandas().userId.values)


# columns of the old data frame 
all_col = ['gender','length','level','page','registration','ts','userAgentNew','userId']

# columns of the to generate data frame 
new_col = ['userId','PlayedTime','PaidTime','PlayNumber','PosEva','NegEva','Gender','Device','y']

# schema of the to generate data frame 
schema = StructType([
    StructField('userId', StringType(), True),
    StructField('PlayedTime',IntegerType(), True),
    StructField('PaidTime',IntegerType(), True), 
    StructField('PlayNumber',IntegerType(), True),
    StructField('PosEva',IntegerType(), True),
    StructField('NegEva',IntegerType(), True), 
    StructField('Gender',StringType(), True),
    StructField('Device',StringType(),True),
    StructField('y',IntegerType(),True)
])

df_ubc = spark.createDataFrame(data = '', schema = schema ).toDF(*new_col) # the empty dataframe to store the data
df_ubc_list = [] # the empty list used to store data 
# Due to the disconnection setting of Udacity, I have to perform the calculation in two steps

userId_length = len(userId_list)
for userId in userId_list[0:int(userId_length/2)]:
    
    value_temp = []

    stay_user = 0

    df_temp_paid = df_total.select(*all_col).where((df_total.level == 'paid') & (df_total.userId == userId))

    if len(df_temp_paid.head(1)) == 0:
        # print("Free users")
        pass

    else: 
        first_paid_time = df_temp_paid.take(1)[0].__getitem__('ts')

        # NO Free users 
        churn_time = df_temp_paid.select("ts").where((df_temp_paid.level == 'paid') & (df_temp_paid.page == 'Cancellation Confirmation')|(df_temp_paid.level == 'Submit Downgrade'))

        if churn_time.head(1) == []:
            # stay paid user 
            stay_user = 1
            first_churn_time = df_temp_paid.select('ts').groupby().max().collect()[0].__getitem__('max(ts)')


        else:
            # churn paid user 
            first_churn_time = churn_time.take(1)[0].__getitem__('ts')

        df_final = df_temp_paid.select(*all_col).where((df_temp_paid.ts >= first_paid_time) & (df_temp_paid.ts <= first_churn_time))

        # add userId 
        value_temp.append(userId)

        # add new column total played time 
        try:
            value_temp.append(int(df_final.select('length').groupby().sum().collect()[0].__getitem__('sum(length)')))
        except:
            value_temp.append(None)

        # total paid time   
        try:
            value_temp.append(int((df_final.select('ts').groupby().max().collect()[0].__getitem__('max(ts)')\
                                -df_final.select('registration').collect()[0].__getitem__('registration'))/3600/1000))
        except:
            value_temp.append(None)

        # total played number
        try:
            value_temp.append(int(df_final.select('page').where(df_final.page == 'NextSong').groupby().count().collect()[0]\
                                .__getitem__('count')))
        except:
            value_temp.append(None)
            
        # add to list and thumb up
        try: 
            value_temp.append(int(df_final.select('page').where((df_final.page == 'Add to Playlist')\
                                                            |(df_final.page == 'Thumbs Up')).\
                                groupby().count().collect()[0].__getitem__('count')))        
        except: 
            value_temp.append(None)
            
        # negative evaluated
        
        try: 
            value_temp.append(int(df_final.select('page').where((df_final.page == 'Downgrade')\
                                                            |(df_final.page == 'Thumbs Down')\
                                                            |(df_final.page == 'Roll Advert')\
                                                            |(df_final.page == 'Cancel')).\
                                groupby().count().collect()[0].__getitem__('count')))          
        except:
            value_temp.append(None)
            
        # gender 
        value_temp.append(df_final.select('gender').collect()[0].__getitem__('gender'))

        # device 
        value_temp.append(df_final.select('userAgentNew').collect()[0].__getitem__('userAgentNew'))

        value_temp.append(stay_user)
        
        df_ubc_list.append(tuple(value_temp))

userId_length
# save temperory results 

import csv
with open('df_ubc_totallist', 'w') as f: 
      
    # using csv.writer method from CSV package 
    write = csv.writer(f) 
      
    write.writerow(df_ubc_list) 
# run the second half 

for userId in userId_list[int(userId_length/2):-1]:
    
    value_temp = []

    stay_user = 0

    df_temp_paid = df_total.select(*all_col).where((df_total.level == 'paid') & (df_total.userId == userId))

    if len(df_temp_paid.head(1)) == 0:
        # print("Free users")
        pass

    else: 
        first_paid_time = df_temp_paid.take(1)[0].__getitem__('ts')

        # NO Free users 
        churn_time = df_temp_paid.select("ts").where((df_temp_paid.level == 'paid') & (df_temp_paid.page == 'Cancellation Confirmation')|(df_temp_paid.level == 'Submit Downgrade'))

        if churn_time.head(1) == []:
            # stay paid user 
            stay_user = 1
            first_churn_time = df_temp_paid.select('ts').groupby().max().collect()[0].__getitem__('max(ts)')


        else:
            # churn paid user 
            first_churn_time = churn_time.take(1)[0].__getitem__('ts')

        df_final = df_temp_paid.select(*all_col).where((df_temp_paid.ts >= first_paid_time) & (df_temp_paid.ts <= first_churn_time))

        # add userId 
        value_temp.append(userId)

        # add new column total played time 
        try:
            value_temp.append(int(df_final.select('length').groupby().sum().collect()[0].__getitem__('sum(length)')))
        except:
            value_temp.append(None)

        # total paid time   
        try:
            value_temp.append(int((df_final.select('ts').groupby().max().collect()[0].__getitem__('max(ts)')\
                                -df_final.select('registration').collect()[0].__getitem__('registration'))/3600/1000))
        except:
            value_temp.append(None)

        # total played number
        try:
            value_temp.append(int(df_final.select('page').where(df_final.page == 'NextSong').groupby().count().collect()[0]\
                                .__getitem__('count')))
        except:
            value_temp.append(None)
            
        # add to list and thumb up
        try: 
            value_temp.append(int(df_final.select('page').where((df_final.page == 'Add to Playlist')\
                                                            |(df_final.page == 'Thumbs Up')).\
                                groupby().count().collect()[0].__getitem__('count')))        
        except: 
            value_temp.append(None)
            
        # negative evaluated
        
        try: 
            value_temp.append(int(df_final.select('page').where((df_final.page == 'Downgrade')\
                                                            |(df_final.page == 'Thumbs Down')\
                                                            |(df_final.page == 'Roll Advert')\
                                                            |(df_final.page == 'Cancel')).\
                                groupby().count().collect()[0].__getitem__('count')))          
        except:
            value_temp.append(None)
            
        # gender 
        value_temp.append(df_final.select('gender').collect()[0].__getitem__('gender'))

        # device 
        value_temp.append(df_final.select('userAgentNew').collect()[0].__getitem__('userAgentNew'))

        value_temp.append(stay_user)
        
        df_ubc_list.append(tuple(value_temp))
        

with open('df_ubc_totallist', 'w') as f: 
      
    # using csv.writer method from CSV package 
    write = csv.writer(f) 
      
    write.writerow(df_ubc_list)         


# Since JSON data can be semi-structured and contain additional metadata, it is possible that you might face issues with the DataFrame layout.
# Please read the documentation of 'SparkSession.read()' to learn more about the possibilities to adjust the data loading.
# PySpark documentation: http://spark.apache.org/docs/2.0.2/api/python/pyspark.sql.html#pyspark.sql.DataFrameReader.json

df_data_1 = spark.read.json(cos.url('medium-sparkify-event-data.json', 'sparkify-donotdelete-pr-gkwqpiyvqhelzl'))
df_data_1.take(5)

df_ubc.write.format('json').save('sparkify-donotdelete-pr-gkwqpiyvqhelzl')
df_ubc = df_ubc.union(spark.createDataFrame(data = df_ubc_list, schema = schema ).toDF(*new_col))
# save to local 
# df_ubc.write.format('json').save('./final/ubc.json')
df_ubc.show()<jupyter_output>+------+----------+--------+----------+------+------+------+------+---+
|userId|PlayedTime|PaidTime|PlayNumber|PosEva|NegEva|Gender|Device|  y|
+------+----------+--------+----------+------+------+------+------+---+
|200002|     37671|    1279|       143|    10|     7|     M|     m|  0|
|   296|      9351|     639|        36|     3|     2|     F|     m|  0|
|   124|    456664|    2701|      1826|   147|    38|     F|     m|  0|
|    51|     66817|     452|       266|    24|     7|     M|     w|  0|
|   205|    263650|    1249|      1048|    75|    20|     F|     m|  1|
|   282|    564720|    1864|      2259|   173|    47|     M|     w|  1|
|   232|    332766|    2568|      1354|   109|    29|     F|     m|  1|
|    15|    458152|    1361|      1835|   150|    42|     M|     w|  1|
|    54|    128938|    1806|       517|    41|    13|     F|     w|  0|
|   234|    348658|     870|      1398|   118|    31|     M|     w|  0|
|   155|     35079|    2286|       140|    11|     1|     F|    [...]<jupyter_text>3.2 Study generated features  3.2.1 `Gender`<jupyter_code>sns.barplot(['M','F'], [df_ubc.select('Gender').where((df_ubc.Gender =='M') & (df_ubc.y ==1)).count(),df_ubc.select('Gender').where((df_ubc.Gender =='F') & (df_ubc.y ==1)).count()])
plt.title('Gender distribution in group "Stay user"')
sns.barplot(['M','F'], [df_ubc.select('Gender').where((df_ubc.Gender =='M') & (df_ubc.y ==0)).count(),df_ubc.select('Gender').where((df_ubc.Gender =='F') & (df_ubc.y ==0)).count()])
plt.title('Gender distribution in group "Churn user"')<jupyter_output><empty_output><jupyter_text>3.2.2 `Device`<jupyter_code>sns.barplot(['Windows','Apple','linux'], [df_ubc.select('Device').where((df_ubc.Device =='w') & (df_ubc.y ==1)).count(), \
                                         df_ubc.select('Device').where((df_ubc.Device =='m') & (df_ubc.y ==1)).count(),\
                                         df_ubc.select('Device').where((df_ubc.Device =='l') & (df_ubc.y ==1)).count()])
plt.title('Device distribution in group "Stay user"')
sns.barplot(['Windows','Apple','linux'], [df_ubc.select('Device').where((df_ubc.Device =='w') & (df_ubc.y ==0)).count(), \
                                         df_ubc.select('Device').where((df_ubc.Device =='m') & (df_ubc.y ==0)).count(),\
                                         df_ubc.select('Device').where((df_ubc.Device =='l') & (df_ubc.y ==0)).count()])
plt.title('Device distribution in group "Churn user"')<jupyter_output><empty_output><jupyter_text>3.3.3 `PlayedTime`<jupyter_code>df_ubc = df_ubc.fillna(0)
# there is a None in the list, replace it with 0

sns.distplot( [pt.PlayedTime for pt in df_ubc.select('PlayedTime').where(df_ubc.y == 1).collect()], label="Stay user",kde = False)
sns.distplot( [pt.PlayedTime for pt in df_ubc.select('PlayedTime').where(df_ubc.y == 0).collect()], label="Churn user",kde = False)
plt.legend()<jupyter_output><empty_output><jupyter_text>3.3.4 `PaidTime`<jupyter_code>sns.distplot( [pt.PaidTime for pt in df_ubc.select('PaidTime').where(df_ubc.y == 1).collect()], label="Stay user",kde = False)
sns.distplot( [pt.PaidTime for pt in df_ubc.select('PaidTime').where(df_ubc.y == 0).collect()], label="Churn user",kde = False)
plt.legend()<jupyter_output><empty_output><jupyter_text>3.3.5 `PlayedNumber`<jupyter_code>sns.distplot( [pt.PlayNumber for pt in df_ubc.select('PlayNumber').where(df_ubc.y == 1).collect()], label="Stay user",kde = False)
sns.distplot( [pt.PlayNumber for pt in df_ubc.select('PlayNumber').where(df_ubc.y == 0).collect()], label="Churn user",kde = False)
plt.legend()<jupyter_output><empty_output><jupyter_text>3.3.6 PovEva<jupyter_code>sns.distplot( [pt.PosEva for pt in df_ubc.select('PosEva').where(df_ubc.y == 1).collect()], label="Stay user",kde = False)
sns.distplot( [pt.PosEva for pt in df_ubc.select('PosEva').where(df_ubc.y == 0).collect()], label="Churn user",kde = False)
plt.legend()<jupyter_output><empty_output><jupyter_text>3.3.7 NegEva<jupyter_code>sns.distplot( [pt.NegEva for pt in df_ubc.select('NegEva').where(df_ubc.y == 1).collect()], label="Stay user",kde = False)
sns.distplot( [pt.NegEva for pt in df_ubc.select('NegEva').where(df_ubc.y == 0).collect()], label="Churn user",kde = False)
plt.legend()<jupyter_output><empty_output><jupyter_text>4. Modeling  manuel one hot encoder<jupyter_code># man column
functionf = udf(lambda x: 1 if x == 'M' else 0)
df_ubc = df_ubc.withColumn('Gender_m', functionf(F.col('Gender')))

# woman column
functionf = udf(lambda x: 1 if x == 'F' else 0)
df_ubc = df_ubc.withColumn('Gender_f', functionf(F.col('Gender')))

# remove column gender
df_ubc = df_ubc.drop('Gender')
# w column
functionf = udf(lambda x: 1 if x == 'w' else 0)
df_ubc = df_ubc.withColumn('Device_w', functionf(F.col('Device')))

# a column
functionf = udf(lambda x: 1 if x == 'm' else 0)
df_ubc = df_ubc.withColumn('Device_m', functionf(F.col('Device')))

# l column
functionf = udf(lambda x: 1 if x == 'l' else 0)
df_ubc = df_ubc.withColumn('Device_l', functionf(F.col('Device')))

# remove column gender
df_ubc = df_ubc.drop('Device')
# convert column type

df_ubc = df_ubc.withColumn("userId", df_ubc["userId"].cast(IntegerType()))
df_ubc = df_ubc.withColumn("Gender_m", df_ubc["Gender_m"].cast(IntegerType()))
df_ubc = df_ubc.withColumn("Gender_f", df_ubc["Gender_f"].cast(IntegerType()))
df_ubc = df_ubc.withColumn("Device_w", df_ubc["Device_w"].cast(IntegerType()))
df_ubc = df_ubc.withColumn("Device_m", df_ubc["Device_m"].cast(IntegerType()))
df_ubc = df_ubc.withColumn("Device_l", df_ubc["Device_l"].cast(IntegerType()))
df_ubc.show()
from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, StandardScaler,VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import BinaryClassificationEvaluator

rest, validation = df_ubc.randomSplit([0.7, 0.3], seed=42)


input_col = ['userId','PlayedTime','PaidTime','PlayNumber','PosEva','NegEva','Gender_m','Gender_f','Device_w','Device_m','Device_l']
vecAssembler = VectorAssembler(inputCols=input_col, outputCol= "features")
stringIndexer = StringIndexer(inputCol="y", outputCol="label")

lr =  LogisticRegression(maxIter=10, regParam=0.0, elasticNetParam=0)
pipeline = Pipeline(stages=[vecAssembler,stringIndexer,lr])

paramGrid = ParamGridBuilder().addGrid(lr.regParam,[0.0, 0.1]).build()

crossval = CrossValidator(estimator=pipeline,
                          estimatorParamMaps=paramGrid,
                          evaluator=BinaryClassificationEvaluator(),
                          numFolds=3)

cvModel_q1 = crossval.fit(rest)
predict_df = cvModel_q1.transform(validation)

cvModel_q1.avgMetrics
results = cvModel_q1.transform(validation)
print(results.filter(results.label == results.prediction).count() / results.count())<jupyter_output>0.8584905660377359
{'fraction_non_alphanumeric': 0.1499056927297668, 'fraction_numerical': 0.024048353909465022, 'mean_word_length': 3.1547640249332147, 'pattern_counts': {'":': 0, '<': 83, '<?xml version=': 0, '>': 83, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '21', 'dir': 'jupyter-structured-clean-dedup', 'id': '140796', 'license': 'FTL', 'n_tokens_mistral': 7976, 'n_tokens_neox': 7546, 'n_words': 1629, 'path': 'Sparkify_IBMWatson.ipynb', 'repo_name': 'KangleChen/Sparkify', 'size': '23367'}