<jupyter_start><jupyter_text>Build a recommender system with retail data on Vertex AI using PySpark  Table of contents* [Overview](section-1)* [Dataset](section-2)* [Objective](section-3)* [Costs](section-4)* [Create a Dataproc cluster with component gateway enabled and JupyterLab extension](section-5)* [Connect to the cluster from the notebook](section-6)* [Explore the data](section-7)* [Define the ALS Model](section-8)* [Evaluate the model](section-9)* [Save the ALS model to Cloud Storage](section-10)  * [Write the recommendations to BigQuery](section-11)* [Clean up](section-12)  OverviewRecommender systems are powerful tools that model existing customer behavior to generate recommendations. These models generally build complex matrices and map out existing customer preferences in order to find intersecting interests and offer recommendations. These matrices can be very large and will benefit from distributed computing and large memory pools. In a Vertex AI Workbench managed notebooks instance, you can use distributed computing by processing your data in PySpark on a Dataproc cluster.*Note: This notebook file was designed to run in a [Vertex AI Workbench managed notebooks](https://cloud.google.com/vertex-ai/docs/workbench/managed/create-instance) instance using a `Python 3` kernel generated by a Dataproc runtime. Some components of this notebook may not work in other notebook environments.*  DatasetThis notebook uses the `looker-private-demo.retail` dataset in BigQuery. The dataset can be accessed by pinning the `looker-private-demo` project in BigQuery. Instead of going to the BigQuery user interface, this process can be performed from the JupyterLab user interface on a Vertex AI Workbench managed notebooks instance. Vertex AI Workbench managed notebooks instances support browsing through the datasets and tables from BigQuery through its BigQuery integration. In this dataset, the `retail.order_items` table will be used to train the recommendation system using PySpark. This table contains information on various orders related to the users and items (products) in the dataset. ObjectiveThis tutorial builds a recommendation model with a [collaborative filtering](https://en.wikipedia.org/wiki/Collaborative_filtering) approach using the interactive PySpark features offered by the Vertex AI Workbench's managed notebooks instances. You'll set up a remotely-connected Dataproc cluster and use the Alternating Least Squares(ALS) method implemented in PySpark's MLlib library.The steps performed in this notebook are:1. Connect your managed notebooks instance to a Dataproc cluster with PySpark.2. Explore the dataset in BigQuery from within the notebook.3. Preprocess the data.4. Train a PySpark ALS model on the data.5. Evaluate the ALS model.6. Generate recommendations.7. Save the recommendations to a BigQuery table using the PySpark-BigQuery connector.8. Save the ALS model to a Cloud Storage bucket.9. Clean up the resources. CostsThis tutorial uses the following billable components of Google Cloud:* Vertex AI* Dataproc* BigQuery* Cloud StorageLearn about [Vertex AIpricing](https://cloud.google.com/vertex-ai/pricing), [Dataproc pricing](https://cloud.google.com/dataproc/pricing), [BigQuerypricing](https://cloud.google.com/bigquery/pricing) and [Cloud Storagepricing](https://cloud.google.com/storage/pricing) and use the [PricingCalculator](https://cloud.google.com/products/calculator/)to generate a cost estimate based on your projected usage.  Set your project ID**If you don't know your project ID**, you may be able to get your project ID using `gcloud`.<jupyter_code>PROJECT_ID = ""

# Get your Google Cloud project ID from gcloud
if not os.getenv("IS_TESTING"):
    shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID: ", PROJECT_ID)<jupyter_output><empty_output><jupyter_text>Otherwise, set your project ID here.<jupyter_code>if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "[your-project-id]"  # @param {type:"string"}<jupyter_output><empty_output><jupyter_text>TimestampIf you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a timestamp for each instance session, and append it onto the name of resources you create in this tutorial.<jupyter_code>from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")<jupyter_output><empty_output><jupyter_text>Create a Cloud Storage bucket**The following steps are required, regardless of your notebook environment.**When you submit a training job using the Cloud SDK, you upload a Python packagecontaining your training code to a Cloud Storage bucket. Vertex AI runsthe code from this package. In this tutorial, Vertex AI also saves thetrained model that results from your job in the same bucket. Using this model artifact, you can thencreate Vertex AI model and endpoint resources in order to serveonline predictions.Set the name of your Cloud Storage bucket below. It must be unique across allCloud Storage buckets.You may also change the `REGION` variable, which is used for operationsthroughout the rest of this notebook. Make sure to [choose a region where Vertex AI services areavailable](https://cloud.google.com/vertex-ai/docs/general/locationsavailable_regions). You maynot use a Multi-Regional Storage bucket for training with Vertex AI.<jupyter_code>BUCKET_NAME = "gs://[your-bucket-name]"  # @param {type:"string"}
REGION = "[your-region]"  # @param {type:"string"}
if BUCKET_NAME == "" or BUCKET_NAME is None or BUCKET_NAME == "gs://[your-bucket-name]":
    BUCKET_NAME = "gs://" + PROJECT_ID + "aip-" + TIMESTAMP<jupyter_output><empty_output><jupyter_text>**Only if your bucket doesn't already exist**: Run the following cell to create your Cloud Storage bucket.<jupyter_code>! gsutil mb -l $REGION $BUCKET_NAME<jupyter_output><empty_output><jupyter_text>**Finally**, validate access to your Cloud Storage bucket by examining its contents:<jupyter_code>! gsutil ls -al $BUCKET_NAME<jupyter_output><empty_output><jupyter_text>Before you beginThe ALS model approach is compute-intensive and could take a lot of time to train on a regular notebook environment, so this tutorial uses a Dataproc cluster with PySpark environment. Create a Dataproc cluster with component gateway enabled and JupyterLab extensionCreate the cluster using the following `gcloud` command.<jupyter_code>CLUSTER_NAME = "[your-cluster-name]"
CLUSTER_REGION = "[your-cluster-region]"
CLUSTER_ZONE = "[your-cluster-zone]"
MACHINE_TYPE = "[your=machine-type]"
! gcloud dataproc clusters create $CLUSTER_NAME \
--enable-component-gateway \
--region $CLUSTER_REGION \
--zone $CLUSTER_ZONE \
--single-node \
--master-machine-type $MACHINE_TYPE \
--master-boot-disk-size 100 \
--image-version 2.0-debian10 \
--optional-components JUPYTER \
--project $PROJECT_ID<jupyter_output><empty_output><jupyter_text>Alternatively, the cluster can be created through the Dataproc console as well. Additional settings like network configuratons and service-accounts can be configured there if required. While configuring the cluster, make sure you complete the following:- Provide a name for the cluster.- Select a region and zone for the cluster.- Select the cluster type as single-node. For small and proof-of-concept use-cases, a single-node cluster is recommended.- Enable the component gateway.- In the optional components, select Jupyter Notebook.- (Optional) Select the machine-type (preferably a high-mem machine type).- Create the cluster.  Connect to the cluster from the notebookWhen the new Dataproc cluster is running, the corresponding runtime appears as a kernel in the notebook. The created cluster's name will appear in the list of kernels that can be selected for this notebook. In the top right corner of this notebook file, click the current kernel name, **Python (local)**, and then select the Python 3 kernel that is running on your Dataproc cluster.Note the following:- Your Dataproc kernel might take a few minutes to show up in the list of kernels.- PySpark code in this tutorial can be run on either a PySpark or Python 3 kernel on the Dataproc cluster, but to run the optional code that saves recommendations to a BigQuery table, the Python 3 kernel is recommended.  Tutorial   Explore the data Vertex AI Workbench managed notebooks instances let you explore the BigQuery content from within the managed notebooks instance using a BigQuery integration. This feature lets you look at the metadata and preview of table content, query tables, and get a description of the data in the tables. Check the distribution of the `STATUS` field. @bigquerySELECT STATUS, COUNT(*) order_count FROM looker-private-demo.retail.order_items GROUP BY 1 Join the `order_items` table with the `inventory_items` table from the same dataset to retrieve the product IDs for the orders. @bigqueryWITH user_prod_table AS (SELECT USER_ID, PRODUCT_ID, STATUS FROM looker-private-demo.retail.order_items AS ajoin(SELECT ID, PRODUCT_ID FROM looker-private-demo.retail.inventory_items) AS bon a.inventory_item_id = b.ID )SELECT USER_ID, PRODUCT_ID, STATUS from user_prod_table Once the results from BigQuery are displayed in the above cell, click the **Query and load as DataFrame** button and execute the generated code stub to fetch the data into the current notebook as a dataframe.*Note: By default the data is loaded into a `df` variable, though this can be changed before executing the cell if required.*<jupyter_code># The following two lines are only necessary to run once.
# Comment out otherwise for speed-up.
from google.cloud.bigquery import Client

client = Client()

query = """WITH user_prod_table AS (
SELECT USER_ID, PRODUCT_ID, STATUS FROM looker-private-demo.retail.order_items AS a
join
(SELECT ID, PRODUCT_ID FROM looker-private-demo.retail.inventory_items) AS b
on a.inventory_item_id = b.ID )

SELECT USER_ID, PRODUCT_ID, STATUS from user_prod_table"""
job = client.query(query)
df = job.to_dataframe()<jupyter_output><empty_output><jupyter_text>Preprocess the DataTo run PySpark's ALS method on the existing data, there must be some fields to quantify the relationship between a `USER_ID` and a `PRODUCT_ID`, such as *ratings given by the user*. If such fields already exist in the data, they can be treated as an *explicit feedback* for the ALS model. Otherwise, the fields indicative of a relationship can be given as an *implicit feedback*. Learn more about [feedback for PySpark's ALS method](https://spark.apache.org/docs/2.2.0/ml-collaborative-filtering.htmlexplicit-vs-implicit-feedback).In the current dataset, as there are no such numerical fields, the `STATUS` field is further used to quantify the association between a `USER_ID` and  a `PRODUCT_ID`. Based on when they occur during an order lifecycle and how likely the user is going to like the order, the `STATUS` field is assigned one of the following ratings:- Cancelled - 1- Returned - 2- Processing - 3- Shipped - 4- Complete - 5The ratings given are subjective and can be modified according to the use case.<jupyter_code>score_mapping = {
    "Cancelled": 1,
    "Returned": 2,
    "Processing": 3,
    "Shipped": 4,
    "Complete": 5,
}
df["RATING"] = df["STATUS"].map(score_mapping)<jupyter_output><empty_output><jupyter_text>Check the distribution of the newly generated `RATING` field.<jupyter_code>df["RATING"].plot(kind="hist")<jupyter_output><empty_output><jupyter_text>Load the required methods and classes from PySpark MLlib.<jupyter_code>from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.sql import SparkSession<jupyter_output><empty_output><jupyter_text>Generate a Spark session with the BigQuery-Spark connector configured.*Note: If the notebook is connected to a Dataproc cluster, the session object would show `yarn` as the Master.*<jupyter_code>spark = (
    SparkSession.builder.appName("Recommendations")
    .config("spark.jars", "gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar")
    .getOrCreate()
)

spark<jupyter_output><empty_output><jupyter_text>Convert the pandas dataframe to a spark dataframe for further processing.<jupyter_code>spark_df = spark.createDataFrame(df[["USER_ID", "PRODUCT_ID", "RATING"]])
spark_df.printSchema()
spark_df.show()<jupyter_output><empty_output><jupyter_text>Split the data into train and test<jupyter_code>(train, test) = spark_df.randomSplit([0.8, 0.2], seed=36)
train.count(), test.count()<jupyter_output><empty_output><jupyter_text>Define the ALS Model The PySpark ALS recommender, Alternating Least Squares, is a matrix factorization algorithm. The idea is to build a matrix that maps users to actions. The actions can be reviews, purchases, various options taken, and more. Due to the complexity and size of the matrix, PySpark can run the algorithm in parallel.ALS will attempt to estimate the rating matrix R as the product of two lower-rank matrices, X and Y. Typically these approximations are called "factor" matrices. During each iteration, one of the factor matrices is held constant, while the other is solved for using least squares. The newly-solved factor matrix is then held constant while solving for the other factor matrix.PySpark uses a blocked implementation of the ALS factorization algorithm that groups the two sets of factors (referred to as “users” and “products”) into blocks and reduces communication by only sending one copy of each user vector to each product block on each iteration, and only for the product blocks that need that user’s feature vector.Essentially instead of finding the low-rank approximations to the rating matrix R, this finds the approximations for a preference matrix P where the elements of P are 1 if r > 0 and 0 if r <= 0. The ratings then act as confidence values related to the strength of indicated user preferences rather than explicit ratings given to items. Learn more [about PySpark's ALS algorithm](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.recommendation.ALS.html).<jupyter_code>als = ALS(
    userCol="USER_ID",
    itemCol="PRODUCT_ID",
    ratingCol="RATING",
    nonnegative=True,
    implicitPrefs=False,
    coldStartStrategy="drop",
)<jupyter_output><empty_output><jupyter_text>The ALS model tries to predict the ratings between users and items and so RMSE can be used for evaluating the model.<jupyter_code>evaluator = RegressionEvaluator(
    metricName="rmse", labelCol="RATING", predictionCol="prediction"
)<jupyter_output><empty_output><jupyter_text>Define a hyperparameter grid for cross-validation.<jupyter_code>param_grid = (
    ParamGridBuilder()
    .addGrid(als.rank, [10, 50])
    .addGrid(als.regParam, [0.01, 0.1, 0.2])
    .build()
)
print("No. of settings to be tested: ", len(param_grid))<jupyter_output><empty_output><jupyter_text>Perform cross-validation and save the best model.<jupyter_code>cv = CrossValidator(
    estimator=als, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=3
)
model = cv.fit(train)
best_model = model.bestModel
print("##Parameters for the Best Model##")
print("Rank:", best_model._java_obj.parent().getRank())
print("MaxIter:", best_model._java_obj.parent().getMaxIter())
print("RegParam:", best_model._java_obj.parent().getRegParam())<jupyter_output><empty_output><jupyter_text>Evaluate the modelEvaluate the model by computing the RMSE on the train and test data.<jupyter_code># View the rating predictions by the model on train and test sets
train_predictions = best_model.transform(train)
train_RMSE = evaluator.evaluate(train_predictions)

test_predictions = best_model.transform(test)
test_RMSE = evaluator.evaluate(test_predictions)

print("Train RMSE ", train_RMSE)
print("Test RMSE ", test_RMSE)<jupyter_output><empty_output><jupyter_text>Generate recommendations for all usersThe required number of recommendations for the users can be generated using the ALS model's `recommendForAllUsers()` method.<jupyter_code># Generate 10 product recommendations for all users
nrecommendations = best_model.recommendForAllUsers(10)
nrecommendations.limit(10).show()<jupyter_output><empty_output><jupyter_text>Generate recommendations for a specific userThe earlier step already generated and stored the specified number of product recommendations for all users in the `nrecommendations` dataframe object. To obtain recommendations for a single user, this dataframe object can be queried.<jupyter_code># get product recommendations for the selected user (USER_ID = 1)
nrecommendations.where(nrecommendations.USER_ID == 1).select(
    "recommendations.PRODUCT_ID", "recommendations.rating"
).collect()<jupyter_output><empty_output><jupyter_text>Save the ALS model to Cloud Storage (optional)PySpark's `ALS.save()` method creates a folder at the specified path where it saves the trained model. A Cloud Storage file browser is available in the managed notebooks instance's environment, which you can use to save the model to a Cloud Storage bucket. Use the ALS object's `.save()` function to write the model to the Cloud Storage bucket.<jupyter_code># Save the trained model
GCS_MODEL_PATH = "gs://" + BUCKET_NAME + "/recommender_systems/"
best_model.save(GCS_MODEL_PATH + "rcmd_model")<jupyter_output><empty_output><jupyter_text>Write the recommendations to BigQuery (optional)In order to serve the recommendations to the end-users or any applications, the output from the `recommendForAllUsers()` method can be saved to a BigQuery table using Spark's BigQuery connector. Create a Dataset in BigQueryThe following cell creates a new dataset in BigQuery. @bigquery-- create a dataset in BigQueryCREATE SCHEMA recommender_sysOPTIONS(  location="us"  )  Write the Recommendations to BigQueryPySpark's BigQuery connector requires two necessary fields: a *BigQuery Table name* and a *Cloud Storage path to write the temporary files* while saving the model. These two fields can be provided while writing the recommendations to BigQuery.<jupyter_code>DATASET = "[your-dataset-name]"
TABLE = "[your-bigquery-table-name]"
GCS_TEMP_PATH = "[your-cloud-storage-path]"

nrecommendations.write.format("bigquery").option(
    "table", "{}.{}".format(DATASET, TABLE)
).option("temporaryGcsBucket", GCS_TEMP_PATH).mode("overwrite").save()<jupyter_output><empty_output><jupyter_text>Clean upTo clean up all Google Cloud resources used in this project, you can [delete the Google Cloudproject](https://cloud.google.com/resource-manager/docs/creating-managing-projectsshutting_down_projects) you used for the tutorial.Otherwise, you can delete the individual resources you created in this tutorial:<jupyter_code># remove the BigQuery dataset created for storing the recommendations and all of its tables
! bq rm -r -f -d $PROJECT:$DATASET

# remove the Cloud Storage bucket created and all of its tables
! gsutil rm -r gs://$BUCKET_NAME
# delete the created Dataproc cluster
! gcloud dataproc clusters delete $CLUSTER_NAME --region=$CLUSTER_REGION<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07790176243612473, 'fraction_numerical': 0.004380018771509021, 'mean_word_length': 5.684907633321715, 'pattern_counts': {'":': 6, '<': 98, '<?xml version=': 0, '>': 99, 'https://': 11, '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': '24', 'dir': 'jupyter-structured-clean-dedup', 'id': '634575', 'license': 'Apache-2.0', 'n_tokens_mistral': 5390, 'n_tokens_neox': 5187, 'n_words': 2181, 'path': 'notebooks/community/managed_notebooks/recommender_systems/recommender-system-on-retail-data.ipynb', 'repo_name': 'nayaknishant/vertex-ai-samples', 'size': '19178'}