<jupyter_start><jupyter_text>Comics Rx [A comic book recommendation system](https://github.com/MangrobanGit/comics_rx) ---  3 - ALS Model - All transactions data  Libraries<jupyter_code>%matplotlib inline
%load_ext autoreload
%autoreload 2  # 1 would be where you need to specify the files
#%aimport data_fcns

import pandas as pd # dataframes
import os
import time
import numpy as np

# Data storage
from sqlalchemy import create_engine # SQL helper
import psycopg2 as psql #PostgreSQL DBs

# import necessary libraries
import pyspark
from pyspark.sql import SparkSession
from pyspark.ml.evaluation import RegressionEvaluator
# from pyspark.sql.types import (StructType, StructField, IntegerType
#                                ,FloatType, LongType, StringType)
from pyspark.sql.types import *

import pyspark.sql.functions as F
from pyspark.sql.functions import col, explode, lit, isnan, when, count
from pyspark.ml.recommendation import ALS, ALSModel
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder, TrainValidationSplit
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Custom
import lib.data_fcns as dfc
import lib.keys  # Custom keys lib
import lib.comic_recs as cr

# instantiate SparkSession object
spark = pyspark.sql.SparkSession.builder.master("local[*]").getOrCreate()
# spark = SparkSession.builder.master("local").getOrCreate()<jupyter_output><empty_output><jupyter_text>Import the dataThere is way to directly hit PostgreSQL through JDBC, but I don't know how to do that yet. So have worked around by saving the candidate dataset to JSON, and then will use that as input to Spark.<jupyter_code>trans = spark.read.json('raw_data/trans.json')
# Persist the data
trans.persist()
print(trans.count(), len(trans.columns))
# check schema
trans.printSchema()<jupyter_output>root
 |-- account_num: string (nullable = true)
 |-- comic_title: string (nullable = true)
 |-- date_sold: long (nullable = true)
 |-- item_id: string (nullable = true)
 |-- publisher: string (nullable = true)
 |-- qty_sold: long (nullable = true)
 |-- title_and_num: string (nullable = true)<jupyter_text>More exploration/testingWe won't be using pandas dataframes in the matrix factorization through Spark, but let's cast to one anyway as it will be easier to work with for EDA.<jupyter_code># cast to Pandas dataframe to turn timestamp data to datetime and check nulls. 
trans_df = trans.select('*').toPandas()
trans_df.info()
# Let's double check the data is how we expect it
trans_df.head()
trans_df['dt'] = pd.to_datetime(trans_df['date_sold'], unit='ms')<jupyter_output><empty_output><jupyter_text>Yes. Reverse-confirmed versus the original transactions dataframe in the other notebook that this datetime is correct.   Data Prep for ALSLet's aggregate the data to the two columns we need:- `account_num` - This is the identifier for individual customers.- `comic_title` - The comic. Represents individual volumes/runs of a comic.- `score` - We need to figure out what we want to use to act as a `score`. If these were Amazon items then review scores would be natural fit; but we don't have that. We can maybe use a binary flag of `bought`/`not bought`. Or we can use the `qty_sold`. This might be interesting in that it might capture some interesting behavior from comic 'collectors/speculators'. Since this is first pass, I'm curious as to what `qty_sold` might do! We only care about `account_num`, `comic_title` and `qty_sold`.<jupyter_code>comics_sold = trans[['account_num', 'comic_title', 'qty_sold']]
comics_sold.persist()
comics_sold = comics_sold.withColumn('bought', lit(1))
comics_sold.show(10)
comics_sold = trans[['account_num', 'comic_title', 'qty_sold']]
comics_sold.persist()
total_comics_sold = comics_sold.groupBy(['account_num', 'comic_title']).agg({'qty_sold':'sum'})
total_comics_sold.persist()<jupyter_output><empty_output><jupyter_text>Ok, let's take a look at the results.<jupyter_code>total_comics_sold.show(10)
print(total_comics_sold.count(), len(total_comics_sold.columns))
total_comics_sold = total_comics_sold.withColumn('bought', lit(1))<jupyter_output><empty_output><jupyter_text>I don't like that default column name. Let's fix that to be `qty_sold` again.<jupyter_code>total_comics_sold.show(10)
total_comics_sold = total_comics_sold[['account_num', 'comic_title', 'bought']]
print(total_comics_sold.count(), len(total_comics_sold.columns))<jupyter_output>92098 3<jupyter_text>FormattingSooooooo, I forgot that the values need to be numeric. So need to fix that. Convert `account_id` to integer<jupyter_code>to_int_udf = F.udf(dfc.make_int, IntegerType())
account_num_col = total_comics_sold['account_num']
total_comics_sold = total_comics_sold.withColumn('account_id'
                                        ,to_int_udf(account_num_col))
total_comics_sold.persist()
total_comics_sold.show(10)
print(total_comics_sold.count(), len(total_comics_sold.columns))<jupyter_output>92098 4<jupyter_text>Now I need to find a way to give ids to the `comic_title`. Kind of clunky, but I do have the version in PostgreSQL of the big table. I can just build an ID table up there as source of truth. I could do something on PySpark side, but then think would want to save it somewhere (e.g. the DB) anyway. So might as well do it from the top. Get `comic_id`<jupyter_code>comics = spark.read.json('raw_data/comics.json')
comics.persist()
comics.count()
comics.show(10)
print(comics.count(), len(comics.columns))<jupyter_output>7202 2<jupyter_text>Now we need to join this back into `total_comics_sold`.<jupyter_code># Set aliases
tot = total_comics_sold.alias('tot')
com = comics.alias('com')
tot_sold_ids_only = tot.join(com.select('comic_id','comic_title')
                      ,tot.comic_title==com.comic_title).select('account_id'
                                                                , 'comic_id'
                                                                , 'bought')
tot_sold_ids_only.persist()
tot_sold_ids_only.show(10)
tot_sold_ids_only.printSchema()
print(tot_sold_ids_only.count(), len(tot_sold_ids_only.columns))<jupyter_output>92098 3<jupyter_text>Save this intermediate table.To save work, if needed.<jupyter_code>als_input_df = tot_sold_ids_only.toPandas()
als_input_df.shape
als_input_df.to_json('raw_data/als_input_all.json', orient='records'
                     ,lines=True)
!head raw_data/als_input_all.json<jupyter_output><empty_output><jupyter_text>ALS ModelLet's start with  train/test split.<jupyter_code># Split data into training and test set
(train, test) = tot_sold_ids_only.randomSplit([.8, .2], seed=41916)<jupyter_output><empty_output><jupyter_text>Make sure shapes make sense.<jupyter_code>print(train.count(), len(train.columns))
print(test.count(), len(test.columns))
# Create ALS instance and fit model
als = ALS(maxIter=20,
          rank=10,
          userCol='account_id',
          itemCol='comic_id',
          ratingCol='bought',
          implicitPrefs=True,
          alpha=40,
          seed=41916)
model = als.fit(train)
now = time.ctime(int(time.time()))
print("Completed on {}.".format(now))<jupyter_output>Completed on Fri Jun 28 04:46:16 2019.<jupyter_text>Evaluation on Test<jupyter_code># Generate predictions on TEST
predictions = model.transform(test)
predictions.persist()
predictions.show(10)<jupyter_output>+----------+--------+------+-----------+
|account_id|comic_id|bought| prediction|
+----------+--------+------+-----------+
|       217|     463|     1|0.121892005|
|      1489|     496|     1|-0.36663285|
|       690|     496|     1| 0.06409974|
|      1695|     833|     1|  0.7557971|
|      2573|     833|     1|  1.1742297|
|        62|    1088|     1|-0.01460916|
|      1781|    1238|     1|0.008104252|
|       226|    1342|     1|   0.533931|
|       191|    1591|     1| 0.97786826|
|        24|    1645|     1| 0.13036774|
+----------+--------+------+-----------+
only showing top 10 rows<jupyter_text>`BinaryClassificationEvaluator` only likes doubles for `rawPredictionCol`, so cast it.<jupyter_code>predictions = predictions.withColumn("prediction", predictions["prediction"].cast(DoubleType()))
predictions.show(10)<jupyter_output>+----------+--------+------+--------------------+
|account_id|comic_id|bought|          prediction|
+----------+--------+------+--------------------+
|       217|     463|     1| 0.12189200520515442|
|      1489|     496|     1| -0.3666328489780426|
|       690|     496|     1|  0.0640997365117073|
|      1695|     833|     1|  0.7557970881462097|
|      2573|     833|     1|  1.1742297410964966|
|        62|    1088|     1|-0.01460915990173...|
|      1781|    1238|     1|0.008104251697659492|
|       226|    1342|     1|  0.5339310169219971|
|       191|    1591|     1|  0.9778682589530945|
|        24|    1645|     1| 0.13036774098873138|
+----------+--------+------+--------------------+
only showing top 10 rows<jupyter_text>Initial EvaluationBased on our first swing of the bat:- `maxIter` = 20- `rank` = 10<jupyter_code># Evaluate the model by computing the RMSE on the test data
eval_reg = RegressionEvaluator(metricName="rmse", labelCol="bought",
                                predictionCol="prediction")
rmse = eval_reg.evaluate(predictions)
print("RMSE= " + str(rmse))<jupyter_output>RMSE= nan<jupyter_text>Oh boy, better remove some nans (for now). evaluator = BinaryClassificationEvaluator(metricName='areaUnderROC'                                          ,labelCol='bought'                                          ,rawPredictionCol='prediction'                                          ) auc = evaluator.evaluate(predictions) print("Area Under the Curve = " + str(auc))  Check nans<jupyter_code># Convert to pandas dataframe
pred_df = predictions.select('*').toPandas()

# Check nulls
pred_num_nulls = pred_df['prediction'].isna().sum()

# Num rows
preds_attempted = pred_df.shape[0]
print("There are {} nulls out of {}.".format(pred_num_nulls, preds_attempted))
print("So {:.2f}% are nulls.".format(pred_num_nulls/preds_attempted))<jupyter_output>So 0.02% are nulls.<jupyter_text>Since, they're such a tiny portion of the test set, for now let's remove the nulls for now.<jupyter_code>predictions.count()
# Convert back to spark dataframe
predictions = spark.createDataFrame(pred_df)
predictions.select([count(when(isnan(c), c)).alias(c) for c in predictions.columns]).show()
pred_no_na = predictions.na.drop()
pred_no_na.persist()
pred_no_na.count()
# Evaluate the model by computing the rmse on the test data
rmse2 = eval_reg.evaluate(pred_no_na)

print("RMSE = " + str(rmse2))<jupyter_output>RMSE = 0.4390995770366977<jupyter_text>Ok, so that gives us a baseline. Let's see if we can do a little grid search action.  Get Top N recommendations for Single User Let's make a reference list of `account_id`'s, for testing purposes.<jupyter_code>n_to_test = 5

users = (tot_sold_ids_only.select(als.getUserCol())
                          .sample(False
                                  ,n_to_test/tot_sold_ids_only.count()
                                  ,41916)
        )
users.persist()
users.show()<jupyter_output>+----------+
|account_id|
+----------+
|       593|
|       315|
|      1165|
|      1621|
|       121|
|       770|
+----------+<jupyter_text>We developed and wrote the functionality out to a function in `comic_recs.py`   Testing function!- Pass the function to a pandas dataframe. - Function will ask for an account_id.- Will return top n, n defined in parameters.<jupyter_code>top_n_df = cr.get_top_n_recs_for_user(spark=spark, model=model, topn=5)
top_n_df
top_n_df = cr.get_top_n_recs_for_user(spark=spark, model=model, topn=5)
top_n_df
top_n_df = cr.get_top_n_recs_for_user(spark=spark, model=model, topn=5)
top_n_df<jupyter_output>161<jupyter_text>Conclusions after single run- Seems realistic? Only three tests, but the results seem 'individualized' in the sense that there is no overlap between the sets (albeit small samples). ---  Grid Search<jupyter_code># Create a parameter grid
params = (ParamGridBuilder()
          .addGrid(als.regParam, [1, 0.1])
          .addGrid(als.maxIter, [20, 30])
          .addGrid(als.rank, [10, 20])
          .addGrid(als.alpha, [30, 40])).build()<jupyter_output><empty_output><jupyter_text>cv = TrainValidationSplit(estimator=als                          , evaluator=eval_reg                          , estimatorParamMaps=params                          , trainRatio=0.8)cv = CrossValidator(estimator=als_implicit, estimatorParamMaps=paramGrid, evaluator=RegressionEvaluator())model_implicit = cv.fit(train)<jupyter_code>### Cross validate for best hyperparameters
cv = CrossValidator(estimator=als
                    ,estimatorParamMaps=params
                    ,evaluator=eval_reg
                    ,parallelism=4
                   )
now = time.ctime(int(time.time()))
print("Completed on {}.".format(now))
# Fit and store model
best_model = cv.fit(test)
als_model = best_model.bestModel
now = time.ctime(int(time.time()))
print("Completed on {}.".format(now))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1278315917931508, 'fraction_numerical': 0.04355121653573335, 'mean_word_length': 3.560695652173913, 'pattern_counts': {'":': 0, '<': 72, '<?xml version=': 0, '>': 72, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '21', 'dir': 'jupyter-structured-clean-dedup', 'id': '413074', 'license': 'MIT', 'n_tokens_mistral': 4728, 'n_tokens_neox': 4279, 'n_words': 1069, 'path': 'comrx/dev/deprecated/comics_rx-03a_model_als_all_data.ipynb', 'repo_name': 'MangrobanGit/comic_rx', 'size': '13111'}