# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Data Wrangling with DataFrames Coding Quiz
#
# Use this Jupyter notebook to find the answers to the quiz in the previous section. There is an answer key in the next part of the lesson.

# +
from pyspark.sql import SparkSession

# TODOS: 
# 1) import any other libraries you might need
# 2) instantiate a Spark session 
# 3) read in the data set located at the path "data/sparkify_log_small.json"
# 4) write code to answer the quiz questions 
from pyspark.sql.functions import udf,desc,asc, isnan, count, when, col, sort_array,avg
from pyspark.sql.types import StringType, IntegerType
from pyspark.sql.functions import sum as Fsum
from pyspark.sql.window import Window
import pandas as pd
import numpy as np
# %matplotlib inline
import matplotlib.pyplot as plt

spark = SparkSession \
        .builder \
        .appName("Wrangling Data") \
        .getOrCreate()

path = "data/sparkify_log_small.json"
df = spark.read.json(path)
df.show(n=2)
# -

# # Question 1
#
# Which page did user id "" (empty string) NOT visit?

# TODO: write your code to answer question 1
blank_page = df.filter(df.userId == "").select(col('page')).distinct().collect()
blank_page

all_page = df.select(col('page')).distinct().collect()
all_page

answer1 = set(all_page) - set(blank_page)
answer1

for row in answer1:
    print(row.page)

# # Question 2 - Reflect
#
# What type of user does the empty string user id most likely refer to?
#

# TODO: use this space to explore the behavior of the user with an empty string
q2 = df.filter(df.userId == "")

q2.select(col('auth')).dropDuplicates().show()

q2.select(col('page')).dropDuplicates().show()

# # Question 3
#
# How many female users do we have in the data set?

# TODO: write your code to answer question 3
df.filter(df.gender == 'F').select('userId').dropDuplicates().count()

# # Question 4
#
# How many songs were played from the most played artist?

# TODO: write your code to answer question 4
df.filter(df.artist.isNotNull()).groupby(df.artist).count().sort(col('count')).show()

df.filter(df.artist.isNotNull()).groupby(df.artist).count().orderBy(desc('count')).show()

df.filter(df.artist.isNotNull()) \
    .groupBy('Artist')\
    .agg({'Artist':'count'}) \
    .withColumnRenamed('count(Artist)','count')\
    .sort(desc('count')).show()

# # Question 5 (challenge)
#
# How many songs do users listen to on average between visiting our home page? Please round your answer to the closest integer.
#
#

# TODO: write your code to answer question 5
filter_func = udf(lambda x : int(x == 'Home'), IntegerType())
window = Window.partitionBy('userID')\
                    .orderBy(desc('ts'))\
                    .rangeBetween(Window.unboundedPreceding, 0)

cumsum = df.filter((df.page == 'NextSong') | (df.page == 'Home')) \
            .select('userID','page','ts').show(n=5)

cumsum = df.filter((df.page == 'NextSong') | (df.page == 'Home')) \
            .select('userID','page','ts') \
            .withColumn('homevisit', filter_func(df.page)).show(n=5)

cumsum = df.filter((df.page == 'NextSong') | (df.page == 'Home')) \
            .select('userID','page','ts') \
            .withColumn('homevisit', filter_func(df.page))\
            .withColumn('period', Fsum(col('homevisit')).over(window)).show(n=20)

cumsum = df.filter((df.page == 'NextSong') | (df.page == 'Home')) \
            .select('userID','page','ts') \
            .withColumn('homevisit', filter_func(df.page))\
            .withColumn('period', Fsum(col('homevisit')).over(window))

cumsum.filter((cumsum.page == 'NextSong')) \
    .groupBy('userID', 'period') \
    .agg({'period':'count'}).show(n=5)


cumsum.filter((cumsum.page == 'NextSong')) \
    .groupBy('userID', 'period') \
    .agg({'period':'count'}) \
    .agg({'count(period)':'avg'}).show()



StarcoderdataJupyterScriptsDedupFiltered