Learn to Build AI-Enhanced Retail Search Solutions with MongoDB and Databricks

1
%pip install pymongo tqdm
2
     
3
4
import pandas as pd
5
import json
6
from collections import Counter
7
from tqdm import tqdm
8
from pymongo import MongoClient
9
from pyspark.sql import functions as F
10
from pyspark.sql import types as T
11
from pyspark.sql import Window
12
import pyspark
13
from pyspark import SparkContext
14
from pyspark.sql import SparkSession
15
conf = pyspark.SparkConf()
16
17
18
import copy
19
import numpy as np
20
21
22
tqdm.pandas()
23
24
MONGO_CONN = 'mongodb+srv://:@retail-demo.2wqno.mongodb.net/?retryWrites=true&w=majority'     

1
atp = spark.readStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "atp_status_myn").\ option('spark.mongodb.change.stream.publish.full.document.only','true').\ option('spark.mongodb.aggregation.pipeline',[]).\ option("forceDeleteTempCheckpointLocation", "true").load()

1
atp = atp.drop("_id")

1
atp.writeStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "catalog_myn").\ option('spark.mongodb.operationType', "update").\ option('spark.mongodb.upsertDocument', True).\ option('spark.mongodb.idFieldList', "id").\

1
option("forceDeleteTempCheckpointLocation", "true").\ option("checkpointLocation", "/tmp/retail-atp-myn4/_checkpoint/").\ outputMode("append").\ start()

1
@F.udf() def get_vec(text): 
2
    return model.encode(text)

1
catalog_status = spark.readStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "catalog_myn").\ option('spark.mongodb.change.stream.publish.full.document.only','true').\ option('spark.mongodb.aggregation.pipeline',[]).\ option("forceDeleteTempCheckpointLocation", "true").load()

1
catalog_status = catalog_status.withColumn("discountedPrice", F.col("price") * F.col("pred_price")) catalog_status = catalog_status.withColumn("atp", (F.col("atp").cast("boolean") & F.lit(1).cast("boolean")).cast("integer")) 

1
catalog_status.withColumn("vec", get_vec("title")) catalog_status = catalog_status.drop("_id")

1
catalog_status.writeStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "catalog_final_myn").\ option('spark.mongodb.operationType', "update").\ option('spark.mongodb.idFieldList', "id").\ option("forceDeleteTempCheckpointLocation", "true").\ option("checkpointLocation", "/tmp/retail-atp-myn5/_checkpoint/").\ outputMode("append").\ start()

1
conf = pyspark.SparkConf() conf.set("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.12:10.1.0")

1
spark = SparkSession.builder \
2
 .master("local") \
3
 .appName("test1") \
4
 .config(conf = conf) \
5
 .getOrCreate()

1
@F.udf(T.FloatType()) 
2
3
def beta_fn(pct,a,b):
4
    return float(100*beta.cdf(pct, a,b)) w = Window().partitionBy() df =
5
df.withColumn("purchase_alpha", F.avg('purchase').over(w)) df = df.withColumn("cart_alpha", F.avg('cart').over(w)) df = df.withColumn("total_views_mean", F.avg('total_views').over(w)) df = df.withColumn("purchase_beta", F.expr('total_views_mean - purchase_alpha')) 
6
7
df = df.withColumn("cart_beta", F.expr('total_views_mean - cart_alpha')) df = df.withColumn("purchase_pct", F.expr('(purchase+purchase_alpha)/(total_views+purchase_alpha+purchase_beta)')) 
8
df = df.withColumn("cart_pct", F.expr('(purchase+cart_alpha)/(total_views+cart_alpha+cart_beta)'))

1
model_name = "retail_competitive_pricing_model_1"
2
with mlflow.start_run(run_name=model_name):
3
    # Create and fit a linear regression model
4
    model = RandomForestRegressor(n_estimators=50, max_depth=3)
5
    model.fit(X_train, y_train)
6
    wrappedModel = CompPriceModelWrapper(model)
7
8
    # Log model parameters and metrics
9
    mlflow.log_params(model.get_params())
10
    mlflow.log_metric("mse", np.mean((model.predict(X_test) - y_test) ** 2))
11
    
12
    # Log the model with a signature that defines the schema of the model's inputs and outputs. 
13
    # When the model is deployed, this signature will be used to validate inputs.
14
    signature = infer_signature(X_train, wrappedModel.predict(None,X_train))
15
16
    # MLflow contains utilities to create a conda environment used to serve models.
17
    # The necessary dependencies are added to a conda.yaml file which is logged along with the model.
18
    conda_env =  _mlflow_conda_env(
19
        additional_conda_deps=None,
20
        additional_pip_deps=["scikit-learn=={}".format(sklearn.__version__)],
21
        additional_conda_channels=None,
22
    )
23
    mlflow.pyfunc.log_model(model_name, python_model=wrappedModel, conda_env=conda_env, signature=signature)

1
model_name = "retail_competitive_pricing_model_1"
2
apply_model_udf = mlflow.pyfunc.spark_udf(spark, f"models:/{model_name}/staging")
3
    
4
# Apply the model to the new data
5
columns = ['old_sales','total_sales','min_price','max_price','avg_price','old_avg_price']
6
udf_inputs = struct(*columns)
7
udf_inputs

1
sales = sales.fillna(0.0)
2
sales = sales.withColumn("pred_price",apply_model_udf(udf_inputs))
3
4
sales = sales.withColumn("price_elasticity", F.expr("((old_sales - total_sales)/(old_sales + total_sales))/(((old_avg_price - avg_price)+1)/(old_avg_price + avg_price))"))
5
6
sales = sales.withColumn("discount", F.ceil((F.lit(1) - F.col("pred_price"))*F.lit(100)))

1
sales.select("id", "pred_price", "price_elasticity").write.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "price_myn").\ option('spark.mongodb.idFieldList', 'id').\ mode('overwrite').\ save()

1
{
2
  "mappings": {
3
    "dynamic": false,
4
    "fields": {
5
      "query": [
6
        {
7
          "foldDiacritics": false,
8
          "maxGrams": 7,
9
          "minGrams": 3,
10
          "tokenization": "edgeGram",
11
          "type": "autocomplete"
12
        }
13
      ]
14
    }
15
  }
16
}