# ---
# 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
# ---

# # Spark on Tour
# ## Segmentación automática de usuarios
#
# A partir de la información de perfil de usuarios por género de película generada en ejemplos anterires, vamos a utilizar un algoritmo de clusterización automática mendiante ML no supervisada para analizar patrones de comportamientos similares en nuestros usuarios.
#
# La segmentación mediante ML nos permitira buscar este tipo relaciones/patrones de forma automática y basada en los propios datos, minizando sesgos y subjetividades, y posibilitando el descubrimiento de nueva información a partir de los datos en forma de relaciones no conocidas en el comportamiento de los usuarios.
#

# ### Importamos librerías, definimos esquemas e inicializamos la sesión Spark.

# +
import findspark
findspark.init()

import pyspark
from pyspark.sql.types import *
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

import plotly.express as px


ratingSchema = StructType([
    StructField("user", IntegerType()),
    StructField("movie", IntegerType()),
    StructField("rating", FloatType())
])

movieSchema = StructType([
    StructField("movie", IntegerType()),
    StructField("title", StringType()),
    StructField("genres", StringType())
])


#setup spark session
sparkSession = (SparkSession.builder
                .appName("Introducción API estructurada")
                .master("local[*]")
                .config("spark.scheduler.mode", "FAIR")
                .getOrCreate())
sparkSession.sparkContext.setLogLevel("ERROR")
# -

# ### Leemos el dataset de ratings usuario / película

ratings = sparkSession.read.csv("/tmp/movielens/ratings.csv", schema=ratingSchema, header=True)
ratings.show(10)

# +
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS

(training, test) = ratings.randomSplit([0.8, 0.2])
als = ALS(maxIter=20, rank=10, regParam=0.01, userCol="user", itemCol="movie", ratingCol="rating",
          coldStartStrategy="drop", implicitPrefs=True)
model = als.fit(training)

# -

predictions = model.transform(test)
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))

# ### Leemos el dataset de películas

movies = sparkSession.read.csv("/tmp/movielens/movies.csv", schema=movieSchema, header=True)
movies.show(10, truncate=False)

# ### Transformamos el dataset de películas para asociar cada película con cada uno de sus genéros 
#
# El resultado es un dataset con N filas por película, tantas como género.

movies = movies.select("movie", "title", split("genres", "\|").alias("genres"))
movies = movies.select("movie", "title", explode("genres").alias("genre"))
movies.show(10)

# ### Mezclamos movies y ratings
# Enriquecemos la información de ratings con los géneros de cada película, y nos quedaos con un dataframe donde cada película aparece en varias filas, una por cada género

movieRatings = ratings.join(movies, "movie", "left_outer")
movieRatings.show(10)

# ### Agregamos por género y usuario
# Calculamos indicadores de interés de un usuario en cada género, como el nº total de películas votadas, la media de rating, el máximo rating y el mínimo rating.
#
# Nuestro objetivo es calcular un perfil de usuario por género, por lo que no nos interesan las películas individuales, sino la agregación de los ratings por género para cada usuario

userRatingsGenres = movieRatings.groupBy("user", "genre") \
            .agg( \
                count("rating").alias("num_ratings"),  \
                avg("rating").alias("avg_rating"), \
                min("rating").alias("min_rating"), \
                max("rating").alias("max_rating")) \
            .sort(asc("user"))
userRatingsGenres.toPandas()

# ### Generamos el dataset final de perfil de usuario
# Ya tenemos la información que queríamos, pero no en la forma que necesitamos para, por ejemplo entrenar un modelo de ML y hacer segmentación de usuarios o predicción de cuanto le va a gusar una película en función de los géneros a los que pertenece.
#
# Necesitamos generar una única fila por cada usuario que represente el perfil del usuario, y por tanto sus 'gustos' con respecto a los diverso géneros de película.

userRatingProfile = userRatingsGenres.groupBy("user") \
                .pivot("genre") \
                .agg(sum("avg_rating").alias("rating"), sum("num_ratings").alias("num"))
userRatingProfile.toPandas()

from pyspark.ml.feature import StringIndexer, IndexToString
from pyspark.ml.linalg import Vectors
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
from pyspark.ml.feature import StandardScaler

filteredProfiles = userRatingProfile \
                .select("user", "Action_num", "Adventure_num", "Animation_num", "Children_num", "Romance_num") \
                .na.drop()
filteredProfiles.toPandas()

# +
featureAssembler = VectorAssembler(
        inputCols=["Action_num", "Adventure_num", "Animation_num", "Children_num", "Romance_num"],
        outputCol="features")

clusteringProfiles = featureAssembler.setHandleInvalid("keep").transform(filteredProfiles).select("user", "features")
clusteringProfiles.show(10, truncate=False)
# -

#train K-Means and use elbow method (graphically) to decide the best K
evaluator = ClusteringEvaluator()
kIndex = []
sseError = []
for k in range(2, 12):
    kmeans = KMeans().setK(k)
    kmeansModel = kmeans.fit(clusteringProfiles)
    predictions = kmeansModel.transform(clusteringProfiles)
    silhouette = evaluator.evaluate(predictions)
    sse = kmeansModel.computeCost(clusteringProfiles)
    print("K = ", k, " | Silhouette = ", str(silhouette), " | SSE = ", sse)
    kIndex.append(k)
    sseError.append(sse)

#show results for elbow-method
import plotly.express as px
fig = px.line(x=kIndex, y=sseError)
fig.show()

# +
#K=7 seems the best option, train the segmentation model
kmeans = KMeans().setK(6)
kmeansModel = kmeans.fit(clusteringProfiles)

#predict the cluster of each user
predictions = kmeansModel.transform(clusteringProfiles)
# -

predictedProfiles = filteredProfiles.join(predictions, "user", "left_outer").drop("features")
predictedProfiles.limit(10).toPandas()

#visualize clustering results
fig = px.scatter(predictedProfiles.toPandas(), x="Action_num", y="Adventure_num", color="prediction")
fig.show()

#visualize clustering results
fig = px.scatter_matrix(predictedProfiles.toPandas(), dimensions=["Action_num", "Adventure_num", "Animation_num", "Children_num", "Romance_num"], color="prediction")
fig.show()

#visualize clustering results
fig = px.scatter(predictedProfiles.toPandas(), x="Children_num", y="Romance_num", color="prediction")
fig.show()

#visualize clustering results
fig = px.scatter(predictedProfiles.toPandas(), x="Children_num", y="Animation_num", color="prediction")
fig.show()

# +
#calculate statistics of each cluster+gener
result1 = predictedProfiles.select("user", "prediction", "Action_num", "Adventure_num", "Animation_num", "Children_num", "Romance_num") \
    .groupBy("prediction") \
    .agg(expr("count(*) as count"), \
         expr("avg(Action_num) as avgAct"), expr("min(Action_num) as minAct"), expr("max(Action_num) as maxAct"), \
         expr("avg(Adventure_num) as avgAdv"), expr("min(Adventure_num) as minAdv"), expr("max(Adventure_num) as maxAdv"), \
         expr("avg(Animation_num) as avgAni"), expr("min(Animation_num) as minAni"), expr("max(Animation_num) as maxAni"), \
         expr("avg(Children_num) as avgChild"), expr("min(Children_num) as minChild"), expr("max(Children_num) as maxChild"), \
         expr("avg(Romance_num) as avgRom"), expr("min(Romance_num) as minRom"), expr("max(Romance_num) as maxRom")) \
    .sort(asc("prediction"))

result1.toPandas()
# -



StarcoderdataJupyterScriptsDedupFiltered