<jupyter_start><jupyter_text>Sparkify Project WorkspaceThis workspace contains a tiny subset (128MB) of the full dataset available (12GB). Feel free to use this workspace to build your project, or to explore a smaller subset with Spark before deploying your cluster on the cloud. Instructions for setting up your Spark cluster is included in the last lesson of the Extracurricular Spark Course content.You can follow the steps below to guide your data analysis and model building portion of this project.<jupyter_code>import ibmos2spark
# @hidden_cell
credentials = {
    'endpoint': 'https://s3-api.us-geo.objectstorage.service.networklayer.com',
    'service_id': 'iam-ServiceId-cd73bfae-4dee-4700-9538-9aca46986b8c',
    'iam_service_endpoint': 'https://iam.cloud.ibm.com/oidc/token',
    'api_key': '<KEY>'
}

configuration_name = 'os_0071c2b33ad94ed0b5752143eb58a1c7_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

df_data_1 = spark.read.json(cos.url('mini_sparkify_event_data.json', 'pyspark-donotdelete-pr-vs1bwxhew57w8l'))
df_data_1.take(5)

# import libraries
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import avg, stddev, split, udf, isnull, first, col, format_number, rand
from pyspark.sql.types import IntegerType, FloatType

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression,RandomForestClassifier,DecisionTreeClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import CountVectorizer, IDF, Normalizer, PCA, RegexTokenizer, StandardScaler, StopWordsRemover, StringIndexer, VectorAssembler
from pyspark.ml.regression import LinearRegression
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

import re

from pyspark.sql.types import IntegerType, ArrayType, FloatType, DoubleType, Row, DateType, StringType

import time

import datetime
import pyspark.sql.functions as F
%matplotlib inline
# create a Spark session
# spark = SparkSession.builder \
#     .master("local") \
#     .appName("Creating Features") \
#     .getOrCreate()<jupyter_output><empty_output><jupyter_text>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># mini_sparktify_event_data = 'mini_sparkify_event_data.json'

# df = spark.read.json(mini_sparktify_event_data)
# df.persist()
# df = df.sample(withReplacement=False, fraction=0.6)
df = df_data_1
df.filter("userId IS NULL").count()
df.filter("sessionId IS NULL").count()<jupyter_output><empty_output><jupyter_text>Exploratory Data AnalysisWhen you're working with the full dataset, perform EDA by loading a small subset of the data and doing basic manipulations within Spark. In this workspace, you are already provided a small subset of data you can explore. Define ChurnOnce you've done some preliminary analysis, create a column `Churn` to use as the label for your model. I suggest using the `Cancellation Confirmation` events to define your churn, which happen for both paid and free users. As a bonus task, you can also look into the `Downgrade` events. Explore DataOnce you've defined churn, perform some exploratory data analysis to observe the behavior for users who stayed vs users who churned. You can start by exploring aggregates on these two groups of users, observing how much of a specific action they experienced per a certain time unit or number of songs played.<jupyter_code>df.createOrReplaceTempView("Sparkify_local")
df.select("page").distinct().show(20, False)<jupyter_output>+-------------------------+
|page                     |
+-------------------------+
|Cancel                   |
|Submit Downgrade         |
|Thumbs Down              |
|Home                     |
|Downgrade                |
|Roll Advert              |
|Logout                   |
|Save Settings            |
|Cancellation Confirmation|
|About                    |
|Submit Registration      |
|Settings                 |
|Login                    |
|Register                 |
|Add to Playlist          |
|Add Friend               |
|NextSong                 |
|Thumbs Up                |
|Help                     |
|Upgrade                  |
+-------------------------+
only showing top 20 rows<jupyter_text>Define Churn<jupyter_code>@udf("integer")
def define_churn_event(page):
    """
    Define Churn base on cancellation page
    """
    if (page == 'Cancellation Confirmation'):
        return 1
    else:
        return 0
udf_define_churn = udf(define_churn_event)
df = df.withColumn('Churn', define_churn_event(df.page))<jupyter_output><empty_output><jupyter_text>Define Downgrade<jupyter_code>userw = Window.partitionBy('userId').rangeBetween(Window.unboundedPreceding,Window.unboundedFollowing)
df = df.withColumn('Downgrade',F.max('Churn').over(userw))<jupyter_output><empty_output><jupyter_text>Update Temp Table<jupyter_code>df.createOrReplaceTempView("Sparkify_local")
df.describe('length').show()
df_len = df.select('length')
#Distribution plot for length
plt.figure(figsize=(5,5))
sns.distplot(df_len.toPandas().dropna(), bins = 100, kde = False,  color='darksalmon')
plt.title('length Distribution', 
              fontsize=12, fontweight='bold');
spark.sql('''
        SELECT gender, count(DISTINCT userId) AS CNT
        FROM Sparkify_local
        GROUP BY gender
        ORDER BY CNT DESC
''').show(20)
#notice that we have over 8K records that have null in genders
spark.sql('''
        SELECT Churn, gender, count(sessionId) as cnt_session_id
        FROM Sparkify_local
        GROUP BY Churn, gender
        ORDER BY count(sessionId) DESC
''').show(20)
spark.sql('''
        SELECT Downgrade, sum(length), count(sessionId)
        FROM Sparkify_local
        GROUP BY Downgrade, gender
''').show(20)
spark.sql('''
        SELECT Downgrade, count(userId)
        FROM Sparkify_local
        GROUP BY Downgrade
''').show(20)
df.take(1)
spark.sql('''
        SELECT COUNT(DISTINCT userId ), Churn, level
        FROM Sparkify_local
        GROUP BY Churn, level
''').show()
df_level = spark.sql('''
        SELECT COUNT(DISTINCT userId ) as user_cnt , Downgrade, level
        FROM Sparkify_local
        GROUP BY Downgrade, level
''').toPandas()
df_level
df_level.user_cnt.plot(kind='pie',labels = ['Free-Not Churn', 'Paid-Not Churn','Paid-Churn','Free-Churn']);
plt.title('Paid - Churn Relationship', 
              fontsize=12, fontweight='bold');<jupyter_output><empty_output><jupyter_text>Status when user churn<jupyter_code>df.columns
df_at_churn = df.filter(df.Churn == '1').select('level', 'userId').toPandas()
df_at_churn.groupby('level').count().reset_index()
sns.barplot(x = 'level', y = 'userId', data = df_at_churn.groupby('level').count().reset_index())
plt.title('Status when user churn', 
              fontsize=12, fontweight='bold');<jupyter_output><empty_output><jupyter_text>Churn by Location<jupyter_code>spark.sql('''
        SELECT COUNT(DISTINCT userId ) AS CNT, Churn, location
        FROM Sparkify_local
        GROUP BY Churn, location
        ORDER BY CNT DESC
''').show(20, False)
df_loc = spark.sql('''
        SELECT COUNT(DISTINCT userId ) AS CNT, Churn, location
        FROM Sparkify_local
        GROUP BY Churn, location
        ORDER BY CNT DESC
''').toPandas()
df_loc.head()
plt.figure(figsize=(15,10))
sns.barplot(x = 'location', y = 'CNT' , data = df_loc.head(10))
plt.xticks(rotation=70)
plt.title('Churn by Location', 
              fontsize=12, fontweight='bold');
df.take(1)<jupyter_output><empty_output><jupyter_text>Most popular song in the set<jupyter_code>df.select('song', 'sessionId').groupby('song').count().sort('count',ascending = False).show()
df_song = df.select('song', 'sessionId').dropna().groupby('song').count().sort('count',ascending = False).toPandas()
plt.figure(figsize=(15,10))
sns.barplot(x = 'song', y = 'count' , data = df_song.head(20))
plt.xticks(rotation=70)
plt.title('Top 20 Songs', 
              fontsize=12, fontweight='bold');<jupyter_output>/opt/ibm/conda/miniconda/lib/python/site-packages/matplotlib/backends/backend_agg.py:214: RuntimeWarning: Glyph 131 missing from current font.
  font.set_text(s, 0.0, flags=flags)
/opt/ibm/conda/miniconda/lib/python/site-packages/matplotlib/backends/backend_agg.py:183: RuntimeWarning: Glyph 131 missing from current font.
  font.set_text(s, 0, flags=flags)<jupyter_text>Filter out null user_id<jupyter_code>df.select('userId').describe()
df = df.filter(df['userId']!='')
df = df.filter(col('auth')!='LoggedOut')<jupyter_output><empty_output><jupyter_text>Feature EngineeringOnce you've familiarized yourself with the data, build out the features you find promising to train your model on. To work with the full dataset, you can follow the following steps.- Write a script to extract the necessary features from the smaller subset of data- Ensure that your script is scalable, using the best practices discussed in Lesson 3- Try your script on the full data set, debugging your script if necessaryIf you are working in the classroom workspace, you can just extract features based on the small subset of data contained here. Be sure to transfer over this work to the larger dataset when you work on your Spark cluster.  Get State<jupyter_code>df.select('location','userId').withColumn('location', split(col('location'),',').getItem(1)).take(1)
def clean_data(df):
    """
    clean data to remove pages with null user_is and logged out users
    """
    df = df.filter(df['userId']!='')
    df = df.filter(col('auth')!='LoggedOut')
    get_hour = udf(lambda x:datetime.datetime.fromtimestamp(x/1000.0). hour)
    df = df.withColumn('hour', get_hour(df.ts))
    df = df.withColumn("hour", df.hour.cast(IntegerType()))
    #udf to convert the ts to timestamp
    get_timestamp = udf(lambda x : datetime.datetime.fromtimestamp(x/ 1000.0).strftime("%Y-%m-%d %H:%M:%S"))
    #apply this udf in the dataframe with your timestamp
    df = df.withColumn("datetime", get_timestamp(df.ts))
    #turn registered hixtime to datetime
    df = df.withColumn("registration_date", get_timestamp(df.registration))
    #create date
    df = df.withColumn('date', F.from_unixtime(col('ts')/1000).cast(DateType()))
    df = df.withColumn('location', split(col('location'),',').getItem(1))
    return df
df = clean_data(df)
df.take(1)<jupyter_output><empty_output><jupyter_text>Account Age<jupyter_code>def account_age(df = df):
    """
    INPUT df
    OUTPUT number of days from the start of registeration
    """
    df_account_age = df.select('ts', 'userId' , 'registration') \
    .withColumn('account_age', df.ts - df.registration) \
    .groupby('userId').agg({'account_age' :'max' }) \
    .withColumnRenamed('max(account_age)', 'account_age') \
    .withColumn('account_age', col('account_age')/1000/3600/24)
    return df_account_age
df_account_age = account_age()<jupyter_output><empty_output><jupyter_text>Hour of churn<jupyter_code>def hr_of_churn(df = df):
    """
    INPUT df
    OUTPUT hour of churn event happended
    """
    # Hour of churn
    hr_of_churn = df.select('hour','userId').filter('Churn == "1"') \
        .groupby(['userId']) \
        .max() \
        .withColumnRenamed('max(hour)', 'hour_of_churn')
    return hr_of_churn
hr_of_churn = hr_of_churn()<jupyter_output><empty_output><jupyter_text>Get Date udf to convert the ts to timestampget_timestamp = udf(lambda x : datetime.datetime.fromtimestamp(x/ 1000.0).strftime("%Y-%m-%d %H:%M:%S"))apply this udf in the dataframe with your timestampdf.withColumn("datetime", get_timestamp(df.ts))  get total num of songs<jupyter_code>def tot_song(df = df):
    """
    INPUT df
    OUTPUT total num of song each user played till date
    """
    # Number of songs listened by each users
    df_tot_song = df.where('page == "NextSong"') \
        .groupby(['userId']) \
        .count() \
        .withColumnRenamed('count', 'tot_songs_played')
    return df_tot_song
df_tot_song = tot_song()<jupyter_output><empty_output><jupyter_text>total thumbs up and down<jupyter_code>spark.sql('''
        SELECT page, count(DISTINCT userId) AS CNT
        FROM Sparkify_local
        GROUP BY page
        ORDER BY CNT DESC
''').show(20)<jupyter_output>+--------------------+---+
|                page|CNT|
+--------------------+---+
|            NextSong|225|
|                Home|224|
|           Thumbs Up|220|
|     Add to Playlist|215|
|              Logout|213|
|         Roll Advert|207|
|          Add Friend|206|
|         Thumbs Down|203|
|            Settings|195|
|                Help|193|
|             Upgrade|168|
|               About|156|
|           Downgrade|154|
|       Save Settings|132|
|      Submit Upgrade|131|
|               Error|118|
|Cancellation Conf...| 52|
|              Cancel| 52|
|    Submit Downgrade| 49|
|            Register|  1|
+--------------------+---+
only showing top 20 rows<jupyter_text>df_tot_thumbup = df.select('page', 'userId').where('page = "Thumbs Up"').groupby('userId').count().withColumnRenamed('count', 'tot_thumbsup')df_tot_thumbdown = df.select('page', 'userId').where('page = "Thumbs Down"').groupby('userId').count().withColumnRenamed('count', 'tot_thumbsdown')  Create Page Cnt Function<jupyter_code>def create_page_cnts(df, page_name):
    """
    Create Function to count different page activities by userid
    """
    df_cnt = df.select('page', 'userId') \
    .where(df.page == page_name) \
    .groupby('userId') \
    .count().withColumnRenamed('count', page_name)
    print(df_cnt.show(5))
    return df_cnt
df_tot_thumbup = create_page_cnts(df, "Thumbs Up")

df_tot_thumbdown = create_page_cnts(df, "Thumbs Down")<jupyter_output>+------+---------+
|userId|Thumbs Up|
+------+---------+
|100010|       17|
|200002|       21|
|   124|      171|
|     7|        7|
|    51|      100|
+------+---------+
only showing top 5 rows

None
+------+-----------+
|userId|Thumbs Down|
+------+-----------+
|100010|          5|
|200002|          6|
|   124|         41|
|    51|         21|
|     7|          1|
+------+-----------+
only showing top 5 rows

None<jupyter_text>add to playlist<jupyter_code>df_tot_playlist_song = create_page_cnts(df, "Add to Playlist")<jupyter_output>+------+---------------+
|userId|Add to Playlist|
+------+---------------+
|100010|              7|
|200002|              8|
|    51|             52|
|   124|            118|
|     7|              5|
+------+---------------+
only showing top 5 rows

None<jupyter_text>Add Friend<jupyter_code>df_friends = create_page_cnts(df, "Add Friend")<jupyter_output>+------+----------+
|userId|Add Friend|
+------+----------+
|100010|         4|
|200002|         4|
|    51|        28|
|   124|        74|
|     7|         1|
+------+----------+
only showing top 5 rows

None<jupyter_text>Average time<jupyter_code>df.take(1)
def avg_time_per_page(df=df):
    """
    input: df
    output: average music played time/length per user
    """
    df_avg_time = df.where('page == "NextSong"') \
    .select('length', 'userId') \
    .groupby('userId') \
    .avg().withColumnRenamed('avg(length)', 'length')
    return df_avg_time
df_avg_time_pp = avg_time_per_page()
df_avg_time_pp.show(5)<jupyter_output>+------+------------------+
|userId|            length|
+------+------------------+
|100010|  243.421444909091|
|200002|242.91699209302305|
|   125|261.13913750000006|
|   124|248.17653659965657|
|     7| 253.5605806666668|
+------+------------------+
only showing top 5 rows<jupyter_text>Average Daily Time<jupyter_code>df.take(1)
def avg_daily_active_hour(df = df):
    """
    input: df
    output: average daily active hours per user_id
    """
    df_avg_daily_len = df.select('length', 'userId','date') \
    .groupby('date', 'userId') \
    .sum('length') \
    .withColumnRenamed('sum(length)', 'tot_daily_len') \
    .groupby('userId') \
    .avg('tot_daily_len') \
    .withColumnRenamed('avg(tot_daily_len)', 'avg_daily_len')
    return df_avg_daily_len
df_avg_daily_len = avg_daily_active_hour()<jupyter_output><empty_output><jupyter_text>Join All Features<jupyter_code>def agg_data():
    """
    Joining all data to get the modeling dataframe
    
    """
    joined_df = df_account_age.join(df_avg_time_pp, on='userId', how ='left') \
    .join(df_friends, on='userId', how ='left') \
    .join(df_tot_playlist_song, on='userId', how ='left') \
    .join(df_tot_thumbdown, on='userId', how ='left') \
    .join(df_tot_thumbup, on='userId', how ='left') \
    .join(df_tot_song, on='userId', how ='left') \
    .join(hr_of_churn, on='userId', how ='left') \
    .join(df_avg_daily_len, on='userId', how ='left')
    
    joined_df = joined_df.fillna(0)

    df_cat_feature = df.select('gender', 'location', 'Downgrade', 'userId', 'Churn').drop_duplicates()

    all_df= joined_df.join(df_cat_feature,on='userId')
    all_df = all_df.drop('userId')
    return all_df 
all_df = agg_data()
ls_col = [
 'account_age',
 'length',
 'Add Friend',
 'Add to Playlist',
 'Thumbs Down',
 'Thumbs Up',
 'tot_songs_played',
 'hour_of_churn',
 'avg_daily_len',
 'gender_index',
#  'location_index'
]
# ls_col.append('gender_index')
# ls_col.append('location_index')
def processing_data_pipeline():
    """
    Create data transformation pipline for categorical varaibles
    """
    indexer_gender = StringIndexer(inputCol='gender', outputCol='gender_index')
    indexer_location = StringIndexer(inputCol='location', outputCol='location_index')
    assembler = VectorAssembler(inputCols=ls_col, outputCol='features')
    data_pipeline = Pipeline(stages=[indexer_gender, assembler])
    return data_pipeline
data_pipeline = processing_data_pipeline()<jupyter_output><empty_output><jupyter_text>Fit the data and split into train & test<jupyter_code>model_df = data_pipeline.fit(all_df).transform(all_df)

final_df = model_df.select(model_df.Churn.alias('label'), 'features')

train, test = final_df.randomSplit([0.7, 0.3])
model_df.take(1)<jupyter_output><empty_output><jupyter_text>Split into Validation and Train Set<jupyter_code>train,validate = train.randomSplit([0.7,0.2])<jupyter_output><empty_output><jupyter_text>ModelingSplit the full dataset into train, test, and validation sets. Test out several of the machine learning methods you learned. Evaluate the accuracy of the various models, tuning parameters as necessary. Determine your winning model based on test accuracy and report results on the validation set. Since the churned users are a fairly small subset, I suggest using F1 score as the metric to optimize.<jupyter_code>def fit_classifier(model_name , train = train, test = test):
    """
    Input: model name, training set, testing set
    Output: Classifer, prediction on the testing set
    """
    print('The {} model is training....'.format(model_name))
    if model_name == 'LogisticRegression':
        md = LogisticRegression()
    if model_name == 'RandomForestClassifier':
        md = RandomForestClassifier()
    if model_name == 'DecisionTreeClassifier':
        md = DecisionTreeClassifier()
    clf = md.fit(train)
    test_results = clf.transform(test)
    f1_score_evaluator = MulticlassClassificationEvaluator(metricName='f1')
    f1_score = f1_score_evaluator.evaluate(test_results.select(col('label'), col('prediction')))
    print('The F1 score on the test set is {:.2%}'.format(f1_score))
    return clf,test_results
for model_name in ['LogisticRegression', 'RandomForestClassifier', 'DecisionTreeClassifier']:
    fit_classifier(model_name)<jupyter_output>The LogisticRegression model is training....
The F1 score on the test set is 70.57%
The RandomForestClassifier model is training....
The F1 score on the test set is 65.17%
The DecisionTreeClassifier model is training....
The F1 score on the test set is 68.23%<jupyter_text>Tunning  In this section, I have used the grid search to tune the logistic regression model because it has the best baseline performance out of the box.<jupyter_code>lr = LogisticRegression()
paramGrid = ParamGridBuilder().\
            addGrid(lr.regParam,[0.01, 0.05]).\
            addGrid(lr.elasticNetParam,[0.1, 0.5]).\
            build()

crossval = CrossValidator(estimator=lr,
                          estimatorParamMaps=paramGrid,
                          evaluator=MulticlassClassificationEvaluator(metricName="f1"),
                          numFolds=3)
print('The LR model is training...')
cvModel_lr = crossval.fit(train)
test_results = cvModel_lr.transform(test)
f1_score_evaluator = MulticlassClassificationEvaluator(metricName='f1')
f1_score = f1_score_evaluator.evaluate(test_results.select(col('label'), col('prediction')))
print('The F1 score on the test set is {:.2%}'.format(f1_score))<jupyter_output>The LR model is training...
The F1 score on the test set is 71.71%<jupyter_text>Validation<jupyter_code>validate_result = cvModel_lr.transform(validate)
f1_score_evaluator = MulticlassClassificationEvaluator(metricName='f1')
f1_score = f1_score_evaluator.evaluate(validate_result.select(col('label'), col('prediction')))
print('The F1 score on the validation set is {:.2%}'.format(f1_score))
cvModel_lr.save('lr')
best_model = cvModel_lr.bestModel<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13886474866212323, 'fraction_numerical': 0.022915427891872113, 'mean_word_length': 3.6697992310978216, 'pattern_counts': {'":': 0, '<': 97, '<?xml version=': 0, '>': 97, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '26', 'dir': 'jupyter-structured-clean-dedup', 'id': '476062', 'license': 'MIT', 'n_tokens_mistral': 7465, 'n_tokens_neox': 7011, 'n_words': 1709, 'path': 'Sparkify.ipynb', 'repo_name': 'jerryanziyuan/Churn-Analysis-with-PySpark', 'size': '21902'}