<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 libraries
import datetime
import pandas as pd

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, lit, col, desc, current_date, datediff, count
from pyspark.sql.functions import sum as sum_
from pyspark.sql.functions import max as max_
from pyspark.sql.functions import round as round_
from pyspark.sql.types import IntegerType

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression, RandomForestClassifier, GBTClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
# create a Spark session
spark = SparkSession.builder \
    .master("local") \
    .appName("Sparkify Project") \
    .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># read json file
dfRaw = spark.read.json('mini_sparkify_event_data.json')
dfRaw.printSchema()
dfRaw.head()
# look for missing values
dfRaw.describe("userId").show()
dfRaw.describe("sessionId").show()
# number of null userIds and sessionIDs
print(f'# of null userIds: {dfRaw.filter(dfRaw.userId.isNull()).count()}')
print(f'# of null sessionIds: {dfRaw.filter(dfRaw.sessionId.isNull()).count()}')
# number of empty userIds and sessionIDs
print(f'# of empty userIds: {dfRaw.filter(dfRaw.userId == "").count()}')
print(f'# of empty sessionIds: {dfRaw.filter(dfRaw.sessionId == "").count()}')
# filter empty userIds
emptyUserId = dfRaw.filter(dfRaw.userId == "")
# compare the 'auth' of empty userIds with the raw dataframe
dfRaw.groupby(dfRaw.auth).count().show()
emptyUserId.groupby(emptyUserId.auth).count().show()
# compare the 'page' of empty userIds with the raw dataframe
dfRaw.groupby(dfRaw.page).count().show()
emptyUserId.groupby(emptyUserId.page).count().show()<jupyter_output>+--------------------+------+
|                page| count|
+--------------------+------+
|              Cancel|    52|
|    Submit Downgrade|    63|
|         Thumbs Down|  2546|
|                Home| 14457|
|           Downgrade|  2055|
|         Roll Advert|  3933|
|              Logout|  3226|
|       Save Settings|   310|
|Cancellation Conf...|    52|
|               About|   924|
| Submit Registration|     5|
|            Settings|  1514|
|               Login|  3241|
|            Register|    18|
|     Add to Playlist|  6526|
|          Add Friend|  4277|
|            NextSong|228108|
|           Thumbs Up| 12551|
|                Help|  1726|
|             Upgrade|   499|
+--------------------+------+
only showing top 20 rows

+-------------------+-----+
|               page|count|
+-------------------+-----+
|               Home| 4375|
|              About|  429|
|Submit Registration|    5|
|              Login| 3241|
|           Register|   18|
|               Help|  272|
| [...]<jupyter_text>Empty userIds are users who haven't logged in or don't have a sparkify account yet, so we can filter and use only filled userIds<jupyter_code># filter valid userIds
validUsers = dfRaw.filter(dfRaw.userId != "")<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># create udfs to transform timestamp into date and hour
get_hour = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).hour)
get_date = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).strftime('%Y-%m-%d'))
# apply udfs
validUsers = validUsers.withColumn("hour", get_hour(validUsers.ts).cast("float"))
validUsers = validUsers.withColumn("date", get_date(validUsers.ts))
# create a flag to identify the actions of users who churned
churnUsers = validUsers.filter(validUsers.page == 'Cancellation Confirmation').select('userId')
churnUsers = churnUsers.withColumn("churn", lit(1))
churnUsers=churnUsers.select(col("userId").alias("userId_a"), col("churn"))

validUsers = validUsers.join(churnUsers, validUsers.userId == churnUsers.userId_a, 'left')
validUsers = validUsers.drop(validUsers.userId_a)
# create a flag to identify the actions of users who downgraded
downUsers = validUsers.filter(validUsers.page == 'Submit Downgrade').select('userId')
downUsers = downUsers.withColumn("downgrade", lit(1))
downUsers=downUsers.select(col("userId").alias("userId_b"), col("downgrade"))

validUsers = validUsers.join(downUsers, validUsers.userId == downUsers.userId_b, 'left')
validUsers = validUsers.drop(validUsers.userId_b)
# split into 'churned' and 'nchurned' (not churned) dataframes to compare
churnedDf = validUsers.filter(validUsers.churn == 1).select(['userId','gender','page','hour','date', 'downgrade','level','ts','registration'])
nChurnedDf = validUsers.filter(validUsers.churn.isNull()).select(['userId','gender','page','hour','date', 'downgrade','level','ts','registration'])<jupyter_output><empty_output><jupyter_text>Music and engagement analysis<jupyter_code># filter only 'NextSong' actions
churnedSongs = churnedDf.filter(churnedDf.page == 'NextSong')
nChurnedSongs = nChurnedDf.filter(nChurnedDf.page == 'NextSong')
# get number of users
numUsersChurn = churnUsers.count()
numUsersNChurn = nChurnedSongs.select(nChurnedSongs.userId).dropDuplicates().count()
print(f'avg number of songs listened by churned users: {churnedSongs.count() / numUsersChurn}') 
print(f'avg number of songs listened by active users: {nChurnedSongs.count() / numUsersNChurn}')<jupyter_output>avg number of songs listened by churned users: 699.8846153846154
avg number of songs listened by active users: 1365.2890173410406<jupyter_text>Users who churned hear less music and indicate less engagement on the platform  Usage analysis by day - churned users<jupyter_code># get number of users and songs by day
songByDay = churnedSongs.groupby(churnedSongs.date).count().orderBy(churnedSongs.date).select(col("date").alias("date_a"), col("count").alias("qtdSongs"))
userByDay = churnedSongs.select(['date','userId']).dropDuplicates().groupby(churnedSongs.date).count().orderBy(churnedSongs.date).select(col("date").alias("date_b"), col("count").alias("qtdUsers"))
# get relative number of songs by user
songsByUserDay = songByDay.join(userByDay, songByDay.date_a == userByDay.date_b, 'left').orderBy(songByDay.date_a)
songsByUserDay = songsByUserDay.withColumn("songs_user", songsByUserDay.qtdSongs/songsByUserDay.qtdUsers).drop(songsByUserDay.date_b)
# transform in pandas dataframe
pdSongsDayChurn = songsByUserDay.toPandas()
pdSongsDayChurn.head()
pdSongsDayChurn[['date_a','qtdSongs']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by day - churned users<jupyter_code>pdSongsDayChurn[['date_a','qtdUsers']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of users by day - churned users<jupyter_code>pdSongsDayChurn[['date_a','songs_user']].plot.bar(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by user per day - churned users<jupyter_code>pdSongsDayChurn.describe()<jupyter_output><empty_output><jupyter_text>The number of users and music listened drops over time. Which is to be expected as users leave the platform over time. The avg is ~75 songs listen by user per day  Usage analysis by day - active users<jupyter_code># get number of users and songs by day
aSongByDay = nChurnedSongs.groupby(nChurnedSongs.date).count().orderBy(nChurnedSongs.date).select(col("date").alias("date_a"), col("count").alias("qtdSongs"))
aUserByDay = nChurnedSongs.select(['date','userId']).dropDuplicates().groupby(nChurnedSongs.date).count().orderBy(nChurnedSongs.date).select(col("date").alias("date_b"), col("count").alias("qtdUsers"))
# get relative number of songs by user
aSongsByUserDay = aSongByDay.join(aUserByDay, aSongByDay.date_a == aUserByDay.date_b, 'left').orderBy(aSongByDay.date_a)
aSongsByUserDay = aSongsByUserDay.withColumn("songs_user", aSongsByUserDay.qtdSongs/aSongsByUserDay.qtdUsers).drop(aSongsByUserDay.date_b)
# transform in pandas dataframe
pdSongsDayNChurn = aSongsByUserDay.toPandas()
pdSongsDayNChurn.head()
pdSongsDayNChurn[['date_a','qtdSongs']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by day - active users<jupyter_code>pdSongsDayNChurn[['date_a','qtdUsers']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of users by day - active users<jupyter_code>pdSongsDayNChurn[['date_a','songs_user']].plot.bar(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by user per day - active users<jupyter_code>pdSongsDayNChurn.describe()<jupyter_output><empty_output><jupyter_text>Considering the active users, we have an average of 40 users and 3750 songs per day. The avg of songs listen by user per day is ~89, which demonstrates that users who churn listened to considerably less music per day, having a lower engagement.  Usage analysis by hour - churned users<jupyter_code># get number of users and songs by hour
songByHour = churnedSongs.groupby(churnedSongs.hour).count().orderBy(churnedSongs.hour).select(col("hour").alias("hour_a"), col("count").alias("qtdSongs"))
userByHour = churnedSongs.select(['hour','userId']).dropDuplicates().groupby(churnedSongs.hour).count().orderBy(churnedSongs.hour).select(col("hour").alias("hour_b"), col("count").alias("qtdUsers"))
# get relative number of songs by hour
songsByUserHour = songByHour.join(userByHour, songByHour.hour_a == userByHour.hour_b, 'left').orderBy(songByHour.hour_a)
songsByUserHour = songsByUserHour.withColumn("songs_user", songsByUserHour.qtdSongs/songsByUserHour.qtdUsers).drop(songsByUserHour.hour_b)
# transform in pandas dataframe
pdSongsHourChurn = songsByUserHour.toPandas()
pdSongsHourChurn.head()
pdSongsHourChurn[['qtdSongs']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by hour - churned users<jupyter_code>pdSongsHourChurn[['qtdUsers']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of users by hour - churned users<jupyter_code>pdSongsHourChurn[['songs_user']].plot.bar(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by user per hour - churned users<jupyter_code>pdSongsHourChurn.describe()<jupyter_output><empty_output><jupyter_text>Usage analysis by hour - active users<jupyter_code># get number of users and songs by hour
aSongByHour = nChurnedSongs.groupby(nChurnedSongs.hour).count().orderBy(nChurnedSongs.hour).select(col("hour").alias("hour_a"), col("count").alias("qtdSongs"))
aUserByHour = nChurnedSongs.select(['hour','userId']).dropDuplicates().groupby(nChurnedSongs.hour).count().orderBy(nChurnedSongs.hour).select(col("hour").alias("hour_b"), col("count").alias("qtdUsers"))
# get relative number of songs by hour
aSongsByUserHour = aSongByHour.join(aUserByHour, aSongByHour.hour_a == aUserByHour.hour_b, 'left').orderBy(aSongByHour.hour_a)
aSongsByUserHour = aSongsByUserHour.withColumn("songs_user", aSongsByUserHour.qtdSongs/aSongsByUserHour.qtdUsers).drop(aSongsByUserHour.hour_b)
# transform in pandas dataframe
pdSongsHourNChurn = aSongsByUserHour.toPandas()
pdSongsHourNChurn.head()
pdSongsHourNChurn[['qtdSongs']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by hour - active users<jupyter_code>pdSongsHourNChurn[['qtdUsers']].plot.line(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of users by hour - active users<jupyter_code>pdSongsHourNChurn[['songs_user']].plot.bar(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>number of songs by user per hour - active users<jupyter_code>pdSongsHourNChurn.describe()<jupyter_output><empty_output><jupyter_text>Comparing users who churn with active users, we do not see a significant difference in behavior and the usage curve. But we can again observe that the average of songs listened per user is below, showing less engagement.  Thumbs Up/Down analysis<jupyter_code># filter Thumbs Up and Down actions
churnedTU = churnedDf.filter(churnedDf.page == 'Thumbs Up')
nChurnedTU = nChurnedDf.filter(nChurnedDf.page == 'Thumbs Up')

churnedTD = churnedDf.filter(churnedDf.page == 'Thumbs Down')
nChurnedTD = nChurnedDf.filter(nChurnedDf.page == 'Thumbs Down')
print(f'number of Thumbs Up by churned users: {churnedTU.count() / numUsersChurn}')
print(f'number of Thumbs Up by active users: {nChurnedTU.count() / numUsersNChurn}')
print(f'number of Thumbs Down by churned users: {churnedTD.count() / numUsersChurn}')
print(f'number of Thumbs Down by active users: {nChurnedTD.count() / numUsersNChurn}')
print(f'number of Thumbs Up by the number of songs listened of churned users: {churnedTU.count() / churnedSongs.count()}')
print(f'number of Thumbs Up by the number of songs listened of active users: {nChurnedTU.count() / nChurnedSongs.count()}')
print(f'number of Thumbs Down by the number of songs listened of churned users: {churnedTD.count() / churnedSongs.count()}')
print(f'number of Thumbs Down by the number of songs listened of active users: {nChurnedTD.count() / nChurnedSongs.count()}')<jupyter_output>number of Thumbs Down by the number of songs listened of churned users: 0.013628620102214651
number of Thumbs Down by the number of songs listened of active users: 0.010897775143419632<jupyter_text>When we divide the number of TU and TD by the number of users, we see that active users are more active than churn users, which should be explained by the greater engagement on the platform. But when we relativize by the amount of music listened, we realize that users who churn tend to give more Thumbs Down than active users.  Level analysis<jupyter_code># look the level of the users who churned
numChurnedLevel = churnedDf.filter(churnedDf.page == 'Cancellation Confirmation').groupby(churnedDf.level).count()
numChurnedLevel.show()
# look if users who had level free when they canceled had downgraded before
numChurnedLevelDown = churnedDf.filter(churnedDf.page == 'Cancellation Confirmation').filter(churnedDf.level=='free').agg(sum_(churnedDf.downgrade))
numChurnedLevelDown.show()<jupyter_output>+--------------+
|sum(downgrade)|
+--------------+
|             5|
+--------------+<jupyter_text>60% of users who canceled were paid and if you consider those who were paid but downgraded, it goes up to 70%  Registration analysis<jupyter_code># get the date diff between the last action of user and the registration
churnedReg = churnedDf.groupBy(churnedDf.userId, churnedDf.registration).agg(max_(churnedDf.date))
churnedReg = churnedReg.withColumn("regDate", get_date(churnedReg.registration))
churnedReg = churnedReg.withColumn("dateDiff", datediff(col('max(date)'),col('regDate')))
churnedReg = churnedReg.select(churnedReg.userId,churnedReg.dateDiff)
# transform in pandas dataframe
pdChurnedReg = churnedReg.toPandas()
pdChurnedReg[['dateDiff']].plot.hist(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>histogram of date diff between last action and registration of churned users<jupyter_code># get the date diff between the last action of user and the registration
nChurnedReg = nChurnedDf.groupBy(churnedDf.userId, churnedDf.registration).agg(max_(churnedDf.date))
nChurnedReg = nChurnedReg.withColumn("regDate", get_date(nChurnedReg.registration))
nChurnedReg = nChurnedReg.withColumn("dateDiff", datediff(col('max(date)'),col('regDate')))
nChurnedReg = nChurnedReg.select(nChurnedReg.userId,nChurnedReg.dateDiff)
# transform in pandas dataframe
pdNChurnedReg = nChurnedReg.toPandas()
pdNChurnedReg[['dateDiff']].plot.hist(figsize=(15,5))<jupyter_output><empty_output><jupyter_text>histogram of date diff between last action and registration of active users  we can see that there is a concentration of cancellations in the first 100 days  Friends analysis<jupyter_code>print(f"number of friends by user who churn: {churnedDf.filter(churnedDf.page=='Add Friend').count()/numUsersChurn}")
print(f"number of friends by active user: {nChurnedDf.filter(nChurnedDf.page=='Add Friend').count()/numUsersNChurn}")<jupyter_output>number of friends by user who churn: 12.23076923076923
number of friends by active user: 25.959537572254334<jupyter_text>Active users have twice the number of friends  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.<jupyter_code>def prepare_dataset(df):
    """
    Prepare the data to be able to run in the model.
    After cleaning, the following attributes are created for each user:
        songs: number of songs listened to by the user
        avgThumbsDown: Average user thumbs down per song
        level: most current user level (1=paid and 0=free)
        downgrade: flag if the user has downgraded
        dateDiffReg: difference in days between last user interaction and registration date
        friends: number of friends on the platform
        churn: class that identifies whether user churn or not
        
    input: df - raw dataframe os sparkify use
    
    output: dataframe of features to be used in model
    
    """
    # udfs
    get_date = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).strftime('%Y-%m-%d'))
    get_level = udf(lambda x: 1 if x == 'paid' else 0)
    
    # get validusers
    validUsers = df.filter(dfRaw.userId != "")
    validUsers = validUsers.withColumn("date", get_date(validUsers.ts))
    
    # get number of songs
    numSongs = validUsers.filter(validUsers.page == 'NextSong').groupBy(validUsers.userId).agg(count(validUsers.page))
    numSongs = numSongs.select(col('userId'),col('count(page)').alias('songs'))

    # get number of Thumbs Down
    numThumbsDown = validUsers.filter(churnedDf.page == 'Thumbs Down').groupBy(validUsers.userId).agg(count(validUsers.page))
    numThumbsDown = numThumbsDown.select(col('userId'),col('count(page)').alias('thumbsDown'))

    # get date of last interaction
    lastIteraction = validUsers.groupBy(validUsers.userId).agg(max_(validUsers.ts))
    lastIteraction = lastIteraction.select(col('userId'), col('max(ts)').alias('ts'))
    lastIteraction = lastIteraction.join(validUsers, ['userId','ts'], 'left')
    
    # get level of user
    lastLevel = lastIteraction.select(lastIteraction.userId, lastIteraction.level).dropDuplicates()
    lastLevel = lastLevel.withColumn("level", get_level(lastLevel.level).cast(IntegerType()))

    # see if user have downgraded
    downgrade = validUsers.filter(validUsers.page == 'Submit Downgrade').select('userId').dropDuplicates()
    downgrade = downgrade.withColumn("downgrade", lit(1))

    # get days since registration
    daysReg = validUsers.groupBy(validUsers.userId, validUsers.registration).agg(max_(validUsers.date))
    daysReg = daysReg.withColumn('regDate', get_date(daysReg.registration))
    daysReg = daysReg.withColumn('dateDiffReg', datediff(col('max(date)'),col('regDate')))
    daysReg = daysReg.select(daysReg.userId, daysReg.dateDiffReg)

    # get number of friends
    numFriends = validUsers.filter(validUsers.page=='Add Friend').groupBy(validUsers.userId).agg(count(validUsers.page))
    numFriends = numFriends.select(col('userId'),col('count(page)').alias('friends'))

    # join data
    final = numSongs.join(numThumbsDown, ['userId'], 'left')
    final = final.withColumn('avgThumbsDown', round_(final.thumbsDown/final.songs,5))
    final = final.join(lastLevel, ['userId'], 'left')
    final = final.join(downgrade, ['userId'], 'left')
    final = final.join(daysReg, ['userId'], 'left')
    final = final.join(numFriends, ['userId'], 'left')
    final = final.join(churnUsers, final.userId == churnUsers.userId_a, 'left')

    final = final.drop(final.thumbsDown)
    final = final.drop(final.userId_a)
    final = final.fillna(0)

    return final
dfSparkify = prepare_dataset(dfRaw)
dfSparkify.show(5)<jupyter_output>+------+-----+-------------+-----+---------+-----------+-------+-----+
|userId|songs|avgThumbsDown|level|downgrade|dateDiffReg|friends|churn|
+------+-----+-------------+-----+---------+-----------+-------+-----+
|100010|  275|      0.01818|    0|        0|         55|      4|    0|
|200002|  387|       0.0155|    1|        0|         70|      4|    0|
|   125|    8|          0.0|    0|        0|         72|      0|    1|
|   124| 4079|      0.01005|    1|        0|        131|     74|    0|
|    51| 2111|      0.00995|    1|        0|         20|     28|    1|
+------+-----+-------------+-----+---------+-----------+-------+-----+
only showing top 5 rows<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 scale_features(df):
    """
    Transform coluns into features using VectorAssembler and StandartScale.
    
    input: df - prepeared dataframe of sparkify users.
    
    output: dataframe with scaled features and a label 
    
    """
    
    ls_features = df.columns[1:-1]
    
    assembler = VectorAssembler(inputCols=ls_features, outputCol='features', handleInvalid = 'skip')
    dfVector = assembler.transform(dfSparkify)
    
    scaler = StandardScaler(inputCol='features', outputCol='sc_features', withMean=True, withStd=True)
    scaler_fit = scaler.fit(dfVector) 
    
    df_result = scaler_fit.transform(dfVector)
    
    return df_result.select(df_result.sc_features.alias('features'), df_result.churn.alias('label'))
    
train, test = scale_features(dfSparkify).randomSplit([0.8, 0.2], seed=42)
def evaluate_model(result):
    """
    Returns Accuracy and F1 score of the predicted dataset
    
    input: result - predicted dataset
    
    outputs: accuracy - accuracy of the predicted dataset
             f1 - F1 score of the predicted dataset
    
    """
    
    eval_accuracy = MulticlassClassificationEvaluator(metricName='accuracy')
    eval_f1 = MulticlassClassificationEvaluator(metricName='f1')

    accuracy = eval_accuracy.evaluate(result.select(result.label, result.prediction))
    f1 = eval_f1.evaluate(result.select(result.label, result.prediction))
    
    return accuracy, f1
def test_evaluate_models(train, test):
    """
    Fits and evaluate models and prints the evaluation results.
    
    inputs: 
        train - train dataframe
        test - test dataframe    
    
    """
    
    lsModels = [LogisticRegression(), RandomForestClassifier(), GBTClassifier()]
    
    for ml in lsModels:
          
        model = ml.fit(train)
        predict = model.transform(test)

        accuracy, f1 = evaluate_model(predict)

        print(f'==== {ml.__class__.__name__} ====')
        print(f'   Accuracy: {accuracy}')
        print(f'   F1 score: {f1}')
        print('')
test_evaluate_models(train, test)
def tune_model(ml, train, test):
    """
    Uses the ParamGridBuilder and CrossValidator to tune the LogisticRegression model with the best parameters.
    
    inputs: 
        train - train dataframe
        test - test dataframe
    
    """
    if ml == 'LogisticRegression':
        
        model = LogisticRegression()

        paramGrid = ParamGridBuilder() \
                    .addGrid(model.elasticNetParam,[0.0, 0.1, 0.5, 1.0]) \
                    .addGrid(model.regParam,[0.0, 0.05, 0.1]) \
                    .build()
    elif ml == 'RandomForestClassifier':
        
        model = RandomForestClassifier()
    
        paramGrid = ParamGridBuilder() \
                 .addGrid(model.maxDepth, [2, 5, 10, 25]) \
                 .addGrid(model.maxBins, [10, 20, 50, 100]) \
                 .addGrid(model.numTrees, [5, 20, 50, 100]) \
                 .build()
    else:
        print('Model not found. Try LogisticRegression or RandomForestClassifier')

    
    crossval = CrossValidator(estimator = model,
                              estimatorParamMaps = paramGrid,
                              evaluator = MulticlassClassificationEvaluator(metricName='f1'),
                              numFolds = 3)

    cvModel = crossval.fit(train)
    bestModel = cvModel.bestModel
    
    print(f'==== {model.__class__.__name__} avgMetrics ====')
    print(cvModel.avgMetrics)
    
    prediction = cvModel.transform(test)
    
    lr_accuracy, lr_f1 = evaluate_model(prediction)
    print(f'---- Tuned {model.__class__.__name__} ----')
    print(f'   Accuracy: {lr_accuracy}')
    print(f'   F1 score: {lr_f1}')
    
    return bestModel
bestLR = tune_model('LogisticRegression', train, test)
bestRF = tune_model('RandomForestClassifier', train, test)
bestLR.extractParamMap()
bestRF.extractParamMap()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12468335568469677, 'fraction_numerical': 0.016465504395768143, 'mean_word_length': 4.454083705810646, 'pattern_counts': {'":': 0, '<': 107, '<?xml version=': 0, '>': 107, 'https://': 0, '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': '28', 'dir': 'jupyter-structured-clean-dedup', 'id': '100158', 'license': 'FTL', 'n_tokens_mistral': 8861, 'n_tokens_neox': 8416, 'n_words': 2150, 'path': 'Sparkify.ipynb', 'repo_name': 'gustavo-leandro/sparkify-project', 'size': '26844'}