part2_ab_testing_recommender_system.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "afb2a5b3",
   "metadata": {},
   "source": [
    "# Defining, predicting, and preventing disengaged users in FinTech (Part 2)\n",
    "*__Part 2: Developing intervention using contextual recommender system and designing AB testing experiment to test the intervention__*\n",
    "\n",
    "\n",
    "In the previous article, we cover a strategy in addressing disengaged users in a fintech setting, aided with data. We started with defining engagement, built a predictive model, and analyzed the results. The model's task was to predict who will be the disengaged users next month and performed with 0.94 AUC, and ATM use in the previous month is one of the strong predictors. This suggested that ATM is one of the primary services. This exercise covers the techniques to perform experimentation in preventing these users from disengaging."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "95445bed",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as f, SparkSession, DataFrame, Window\n",
    "from pyspark.sql.functions import col, lit, udf\n",
    "from IPython.display import display\n",
    "import matplotlib.pyplot as plt\n",
    "import pandas as pd\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "id": "f6b4a939",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>user_id</th><th>birth_year</th><th>home_country</th><th>home_city</th><th>device</th><th>num_contacts</th><th>plan</th><th>num_successful_referrals</th><th>joined_date</th><th>transaction_id</th><th>created_date</th><th>merchant_country</th><th>merchant_city</th><th>card_presence</th><th>direction</th><th>transactions_type</th><th>transactions_state</th><th>amount_usd</th></tr>\n",
       "<tr><td>015aff8c321742c6b...</td><td>1992</td><td>SR</td><td>New Cassandra</td><td>Apple</td><td>8</td><td>STANDARD</td><td>0</td><td>2018-06-16 02:33:...</td><td>ae3a61b8c0254daca...</td><td>2018-06-15 06:41:...</td><td>null</td><td>null</td><td>null</td><td>INBOUND</td><td>TOPUP</td><td>COMPLETED</td><td>6.29</td></tr>\n",
       "<tr><td>015aff8c321742c6b...</td><td>1992</td><td>SR</td><td>New Cassandra</td><td>Apple</td><td>8</td><td>STANDARD</td><td>0</td><td>2018-06-16 02:33:...</td><td>5f8f57ed3aa74775a...</td><td>2018-06-16 00:16:...</td><td>null</td><td>null</td><td>null</td><td>INBOUND</td><td>TOPUP</td><td>REVERTED</td><td>1.36</td></tr>\n",
       "<tr><td>015aff8c321742c6b...</td><td>1992</td><td>SR</td><td>New Cassandra</td><td>Apple</td><td>8</td><td>STANDARD</td><td>0</td><td>2018-06-16 02:33:...</td><td>7986278db1824634a...</td><td>2018-06-16 01:37:...</td><td>null</td><td>null</td><td>null</td><td>OUTBOUND</td><td>FEE</td><td>COMPLETED</td><td>6.36</td></tr>\n",
       "<tr><td>015aff8c321742c6b...</td><td>1992</td><td>SR</td><td>New Cassandra</td><td>Apple</td><td>8</td><td>STANDARD</td><td>0</td><td>2018-06-16 02:33:...</td><td>596f2f7d24754ef48...</td><td>2018-06-18 03:09:...</td><td>null</td><td>null</td><td>null</td><td>INBOUND</td><td>TOPUP</td><td>COMPLETED</td><td>13.3</td></tr>\n",
       "<tr><td>015aff8c321742c6b...</td><td>1992</td><td>SR</td><td>New Cassandra</td><td>Apple</td><td>8</td><td>STANDARD</td><td>0</td><td>2018-06-16 02:33:...</td><td>6f8ad9c8af9a40fdb...</td><td>2018-06-28 00:50:...</td><td>BR</td><td>da Rosa</td><td>FALSE</td><td>OUTBOUND</td><td>CARD_PAYMENT</td><td>COMPLETED</td><td>2.84</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+----------+------------+-------------+------+------------+--------+------------------------+--------------------+--------------------+--------------------+----------------+-------------+-------------+---------+-----------------+------------------+----------+\n",
       "|             user_id|birth_year|home_country|    home_city|device|num_contacts|    plan|num_successful_referrals|         joined_date|      transaction_id|        created_date|merchant_country|merchant_city|card_presence|direction|transactions_type|transactions_state|amount_usd|\n",
       "+--------------------+----------+------------+-------------+------+------------+--------+------------------------+--------------------+--------------------+--------------------+----------------+-------------+-------------+---------+-----------------+------------------+----------+\n",
       "|015aff8c321742c6b...|      1992|          SR|New Cassandra| Apple|           8|STANDARD|                       0|2018-06-16 02:33:...|ae3a61b8c0254daca...|2018-06-15 06:41:...|            null|         null|         null|  INBOUND|            TOPUP|         COMPLETED|      6.29|\n",
       "|015aff8c321742c6b...|      1992|          SR|New Cassandra| Apple|           8|STANDARD|                       0|2018-06-16 02:33:...|5f8f57ed3aa74775a...|2018-06-16 00:16:...|            null|         null|         null|  INBOUND|            TOPUP|          REVERTED|      1.36|\n",
       "|015aff8c321742c6b...|      1992|          SR|New Cassandra| Apple|           8|STANDARD|                       0|2018-06-16 02:33:...|7986278db1824634a...|2018-06-16 01:37:...|            null|         null|         null| OUTBOUND|              FEE|         COMPLETED|      6.36|\n",
       "|015aff8c321742c6b...|      1992|          SR|New Cassandra| Apple|           8|STANDARD|                       0|2018-06-16 02:33:...|596f2f7d24754ef48...|2018-06-18 03:09:...|            null|         null|         null|  INBOUND|            TOPUP|         COMPLETED|      13.3|\n",
       "|015aff8c321742c6b...|      1992|          SR|New Cassandra| Apple|           8|STANDARD|                       0|2018-06-16 02:33:...|6f8ad9c8af9a40fdb...|2018-06-28 00:50:...|              BR|      da Rosa|        FALSE| OUTBOUND|     CARD_PAYMENT|         COMPLETED|      2.84|\n",
       "+--------------------+----------+------------+-------------+------+------------+--------+------------------------+--------------------+--------------------+--------------------+----------------+-------------+-------------+---------+-----------------+------------------+----------+"
      ]
     },
     "execution_count": 71,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Conf and set up\n",
    "\n",
    "spark = (SparkSession.builder.master(\"local[*]\")\n",
    "                     .config(\"spark.sql.repl.eagerEval.enabled\", True)\n",
    "                     .getOrCreate())\n",
    "\n",
    "\n",
    "transaction = spark.read.option('compression', 'gzip').option('inferSchema', 'true').csv('data/raw_data', header=True)\n",
    "transaction.limit(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "643a6330",
   "metadata": {},
   "source": [
    "# Contextual Recommender System\n",
    "\n",
    "\n",
    "Context is essential in recommender systems. Given the day of the week as a context, what drinks, movies, items we consume on a Monday night may not be the same as on a Friday night. Adding recency, weather, geolocation, preferences, or other information as the context may make more personalized recommendations, alas, more complex to design as well. Finding which context is practical requires substantial experimentations.\n",
    "To start, we will use the month of the year as the context. Since the data is limited (there is no geodata or merchant information), we can reframe the problem and scope it down. For example: for each user, rank the top five cities where ATMs services will likely be used and provide promotional rates for these cities. As a context, we will use the month of the year. For example, users who usually utilize ATMs in New York are more likely to use the ATM in Florida in February but Chicago in August.\n",
    "\n",
    "### Designing Contextual Recommendation using ALS\n",
    "\n",
    "Alternating Least Squares (ALS) is one of the common techniques used for collaborative filtering. Like other collaborative filtering techniques, ALS decomposes the original matrix (user and interaction) into matrix U and V, such as the multiplication of both matrices produces a close approximation to the original matrix. What makes ALS different from the others is how ALS learns and approximates U and V.\n",
    "While some other algorithm uses gradient descent or other methods, ALS uses similar methods to Ordinary Least Squares (OLS) in regression problems. A simplified explanation of how ALS learns:\n",
    "It starts by initializing U and V pseudo-randomly.\n",
    "Learn matrix V by performing a variation of OLS on each row in matrix V, with every row as the feature and its corresponding column vector from the original matrix as the label.\n",
    "Similar to step 2; except that we switch with column vector in matrix V as the features and its corresponding row from the original matrix the label.\n",
    "Iterates until it converges or maximum iteration is reached.\n",
    "\n",
    "Spark ML has the ALS provided, so we don't have to implement the algorithm from scratch. Furthermore, Spark ML can help when the data is large, as collaborative filtering results in big and sparse matrices.\n",
    "\n",
    "Features for contextual recommender using ALSSimilar to other collaborative filtering techniques, ALS takes user and interaction as the input. One of the possible approaches to add context is by simply splitting the interaction into different contexts. This approach is not recommended for extensive context, as it further increases the size and sparsity.\n",
    "\n",
    "ALS is evaluated by comparing the original matrix and the reconstructed matrix. The comparison can be calculated by measuring the mean absolute differences (MAE) or Root Mean Square of Error (RMSE) to penalize large errors."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "cfaf4951",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>user_id</th><th>context</th><th>count</th></tr>\n",
       "<tr><td>00009eb59e5143159...</td><td>April|ATM|New Julia</td><td>1</td></tr>\n",
       "<tr><td>00009eb59e5143159...</td><td>March|ATM|Aragão</td><td>2</td></tr>\n",
       "<tr><td>00009eb59e5143159...</td><td>May|ATM|Bushtown</td><td>3</td></tr>\n",
       "<tr><td>00009eb59e5143159...</td><td>May|ATM|New Travi...</td><td>1</td></tr>\n",
       "<tr><td>00009eb59e5143159...</td><td>May|ATM|Roweshire</td><td>1</td></tr>\n",
       "<tr><td>0001623b9afc43748...</td><td>April|ATM|Amberside</td><td>3</td></tr>\n",
       "<tr><td>0001623b9afc43748...</td><td>April|ATM|Ballia</td><td>1</td></tr>\n",
       "<tr><td>0001623b9afc43748...</td><td>April|ATM|East De...</td><td>1</td></tr>\n",
       "<tr><td>0001623b9afc43748...</td><td>April|ATM|Jodifort</td><td>1</td></tr>\n",
       "<tr><td>0001623b9afc43748...</td><td>April|ATM|New Joseph</td><td>1</td></tr>\n",
       "</table>\n"
      ],
      "text/plain": [
       "+--------------------+--------------------+-----+\n",
       "|             user_id|             context|count|\n",
       "+--------------------+--------------------+-----+\n",
       "|00009eb59e5143159...| April|ATM|New Julia|    1|\n",
       "|00009eb59e5143159...|    March|ATM|Aragão|    2|\n",
       "|00009eb59e5143159...|    May|ATM|Bushtown|    3|\n",
       "|00009eb59e5143159...|May|ATM|New Travi...|    1|\n",
       "|00009eb59e5143159...|   May|ATM|Roweshire|    1|\n",
       "|0001623b9afc43748...| April|ATM|Amberside|    3|\n",
       "|0001623b9afc43748...|    April|ATM|Ballia|    1|\n",
       "|0001623b9afc43748...|April|ATM|East De...|    1|\n",
       "|0001623b9afc43748...|  April|ATM|Jodifort|    1|\n",
       "|0001623b9afc43748...|April|ATM|New Joseph|    1|\n",
       "+--------------------+--------------------+-----+"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "features = (transaction\n",
    "            .where('transactions_type==\"ATM\"')\n",
    "            .groupBy('user_id',  \n",
    "                     f.concat_ws('|',\n",
    "                                 f.date_format(f.col('created_date'), \"MMMM\"),\n",
    "                                 f.when(f.col('transactions_type')=='ATM', 'ATM').otherwise('NonATM'),\n",
    "                                 'merchant_city').alias('context'))\n",
    "            .agg(f.count('merchant_city').alias('count'))\n",
    "            .orderBy('user_id', 'context', 'count')\n",
    "            )\n",
    "\n",
    "features.limit(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "f6e83606",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "RMSE:  2.3485932867050283 MAE:  1.7029504149895982\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.feature import IndexToString, StringIndexer\n",
    "\n",
    "# Model\n",
    "als = ALS(maxIter=5, regParam=0.01, implicitPrefs=True, seed=777, nonnegative=True, coldStartStrategy='drop',\n",
    "          userCol='user_index', itemCol='context_index', ratingCol='count')\n",
    "rmse = RegressionEvaluator(metricName='rmse', labelCol='count',\n",
    "                                predictionCol='prediction')\n",
    "\n",
    "mae = RegressionEvaluator(metricName='mae', labelCol='count',\n",
    "                                predictionCol='prediction')\n",
    "\n",
    "# Convert IDs into Index\n",
    "userid_to_index = StringIndexer(inputCol='user_id', outputCol='user_index', handleInvalid='keep')\n",
    "context_to_index = StringIndexer(inputCol='context', outputCol='context_index', handleInvalid='keep')\n",
    "id_to_index = Pipeline(stages=[userid_to_index, context_to_index]).fit(features)\n",
    "\n",
    "features_indexed = id_to_index.transform(features)\n",
    "user_mapping = features_indexed.select('user_id', 'user_index').distinct()\n",
    "context_mapping = features_indexed.select('context', 'context_index').distinct()\n",
    "\n",
    "\n",
    "# Training and evaluation\n",
    "training, test = features_indexed.randomSplit([0.8, 0.2])\n",
    "model = als.fit(training)\n",
    "predictions = model.transform(test)\n",
    "rmse_value = rmse.evaluate(predictions)\n",
    "mae_value = mae.evaluate(predictions)\n",
    "print('RMSE: ', rmse_value, 'MAE: ', mae_value)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "ccf1c2c7",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>user_id</th><th>context_month</th><th>context_type</th><th>recommended_city</th><th>rating</th></tr>\n",
       "<tr><td>7dd8a2168e474526a...</td><td>August</td><td>ATM</td><td>New York</td><td>0.35182032</td></tr>\n",
       "<tr><td>628d6a4c8f084994b...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.12443738</td></tr>\n",
       "<tr><td>4acbb9c424264241a...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.11963231</td></tr>\n",
       "<tr><td>628d6a4c8f084994b...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.10954083</td></tr>\n",
       "<tr><td>4acbb9c424264241a...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.107432485</td></tr>\n",
       "<tr><td>6c52a1c328f3450ba...</td><td>August</td><td>ATM</td><td>New York</td><td>0.10485288</td></tr>\n",
       "<tr><td>d1cca52cde5245518...</td><td>August</td><td>ATM</td><td>North Darren</td><td>0.09719041</td></tr>\n",
       "<tr><td>a097d7e832cc4adfb...</td><td>August</td><td>ATM</td><td>New York</td><td>0.09043781</td></tr>\n",
       "<tr><td>dfbc87f7daab47cf9...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.084438555</td></tr>\n",
       "<tr><td>dfbc87f7daab47cf9...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.07145732</td></tr>\n",
       "<tr><td>86997befc22f4a268...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.06263213</td></tr>\n",
       "<tr><td>86997befc22f4a268...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.05624505</td></tr>\n",
       "<tr><td>2c834f0096594fcc8...</td><td>August</td><td>ATM</td><td>New York</td><td>0.048627466</td></tr>\n",
       "<tr><td>8ac9faab0c0049bfa...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.045304846</td></tr>\n",
       "<tr><td>3b03a257b5064ce1a...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.044972476</td></tr>\n",
       "<tr><td>8ac9faab0c0049bfa...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.04399141</td></tr>\n",
       "<tr><td>3b03a257b5064ce1a...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.04038629</td></tr>\n",
       "<tr><td>986f9f2a15a548218...</td><td>August</td><td>ATM</td><td>West Monicaland</td><td>0.038827162</td></tr>\n",
       "<tr><td>986f9f2a15a548218...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.035048462</td></tr>\n",
       "<tr><td>c03b3397fc1a4b118...</td><td>August</td><td>ATM</td><td>Jessicabury</td><td>0.021966068</td></tr>\n",
       "</table>\n",
       "only showing top 20 rows\n"
      ],
      "text/plain": [
       "+--------------------+-------------+------------+----------------+-----------+\n",
       "|             user_id|context_month|context_type|recommended_city|     rating|\n",
       "+--------------------+-------------+------------+----------------+-----------+\n",
       "|7dd8a2168e474526a...|       August|         ATM|        New York| 0.35182032|\n",
       "|628d6a4c8f084994b...|       August|         ATM|     Jessicabury| 0.12443738|\n",
       "|4acbb9c424264241a...|       August|         ATM|     Jessicabury| 0.11963231|\n",
       "|628d6a4c8f084994b...|       August|         ATM| West Monicaland| 0.10954083|\n",
       "|4acbb9c424264241a...|       August|         ATM| West Monicaland|0.107432485|\n",
       "|6c52a1c328f3450ba...|       August|         ATM|        New York| 0.10485288|\n",
       "|d1cca52cde5245518...|       August|         ATM|    North Darren| 0.09719041|\n",
       "|a097d7e832cc4adfb...|       August|         ATM|        New York| 0.09043781|\n",
       "|dfbc87f7daab47cf9...|       August|         ATM|     Jessicabury|0.084438555|\n",
       "|dfbc87f7daab47cf9...|       August|         ATM| West Monicaland| 0.07145732|\n",
       "|86997befc22f4a268...|       August|         ATM|     Jessicabury| 0.06263213|\n",
       "|86997befc22f4a268...|       August|         ATM| West Monicaland| 0.05624505|\n",
       "|2c834f0096594fcc8...|       August|         ATM|        New York|0.048627466|\n",
       "|8ac9faab0c0049bfa...|       August|         ATM|     Jessicabury|0.045304846|\n",
       "|3b03a257b5064ce1a...|       August|         ATM|     Jessicabury|0.044972476|\n",
       "|8ac9faab0c0049bfa...|       August|         ATM| West Monicaland| 0.04399141|\n",
       "|3b03a257b5064ce1a...|       August|         ATM| West Monicaland| 0.04038629|\n",
       "|986f9f2a15a548218...|       August|         ATM| West Monicaland|0.038827162|\n",
       "|986f9f2a15a548218...|       August|         ATM|     Jessicabury|0.035048462|\n",
       "|c03b3397fc1a4b118...|       August|         ATM|     Jessicabury|0.021966068|\n",
       "+--------------------+-------------+------------+----------------+-----------+\n",
       "only showing top 20 rows"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Serving\n",
    "def predict_cities(users:DataFrame, model: ALS, indexer:Pipeline, \n",
    "                   user_mapping:DataFrame, context_mapping:DataFrame,\n",
    "                   top_n=10, month=None):\n",
    "    ''' Generate top n cities recommendations for a subset of a user\n",
    "        with a month as the context\n",
    "    '''\n",
    "    users = indexer.transform(users.withColumn('context', f.lit('')))\n",
    "    \n",
    "    recommendations_index = (model.recommendForUserSubset(users, top_n)\n",
    "                             .select('user_index', f.explode('recommendations').alias('r'))\n",
    "                             .select('user_index', f.col('r.context_index'), f.col('r.rating'))\n",
    "                            )\n",
    "\n",
    "    recommendations = (recommendations_index.join(user_mapping, ['user_index'], 'left')\n",
    "                       .join(context_mapping, ['context_index'], 'left')\n",
    "                       .withColumn('c', f.split(f.col('context'), '\\|'))\n",
    "                       .select('user_id', f.col('c')[0].alias('context_month'), f.col('c')[1].alias('context_type'), f.col('c')[2].alias('recommended_city'), 'rating')\n",
    "                       .orderBy(f.desc('rating'), 'user_id', 'context_type', 'context_month')\n",
    "                      )\n",
    "\n",
    "    if month is None:\n",
    "        return recommendations\n",
    "    else:\n",
    "        return recommendations.where(f.col('context_month')==f.lit(month))\n",
    "\n",
    "predict_cities(features.select('user_id').distinct().limit(100), model, \n",
    "               id_to_index, user_mapping, context_mapping, 10, 'August')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e9a195f",
   "metadata": {},
   "source": [
    "# AB Testing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "eaf13da5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "All users:  185987 Group A: 93440 Group B: 92547\n"
     ]
    }
   ],
   "source": [
    "heartbeat = transaction.select(f.max('created_date')).collect()[0][0] - pd.Timedelta('60 days')\n",
    "\n",
    "all_users = (transaction.filter(f.col('created_date')>f.lit(heartbeat))\n",
    "             .select('user_id', f.concat_ws('|', 'birth_year', 'device').alias('stratum'), 'home_city')\n",
    "             .distinct()\n",
    "            )\n",
    "\n",
    "fractions = all_users.select('stratum').distinct().withColumn('frac', lit(0.5)).rdd.collectAsMap()\n",
    "\n",
    "group_A = all_users.sampleBy('stratum', fractions, 555)\n",
    "group_B = all_users.subtract(group_A)\n",
    "\n",
    "\n",
    "print('All users: ', all_users.count(), 'Group A:', group_A.count(), 'Group B:', group_B.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "id": "7b3e7f8c",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table border='1'>\n",
       "<tr><th>user_id</th><th>group</th><th>promotion_month</th><th>home_city</th><th>recommended_city</th><th>promotion_city</th></tr>\n",
       "<tr><td>00009eb59e5143159...</td><td>Group_B</td><td>August</td><td>East Brianmouth</td><td>null</td><td>East Brianmouth</td></tr>\n",
       "<tr><td>0001623b9afc43748...</td><td>Group_B</td><td>August</td><td>Haridwar</td><td>Jessicabury</td><td>Jessicabury</td></tr>\n",
       "<tr><td>00061b403fd84d91a...</td><td>Group_B</td><td>August</td><td>Vieja Tonga</td><td>null</td><td>Vieja Tonga</td></tr>\n",
       "<tr><td>002583a3828e49079...</td><td>Group_B</td><td>August</td><td>Port Richard</td><td>null</td><td>Port Richard</td></tr>\n",
       "<tr><td>00265da8830d489aa...</td><td>Group_B</td><td>August</td><td>New Jennifer</td><td>null</td><td>New Jennifer</td></tr>\n",
       "<tr><td>00266dbf306f478ba...</td><td>Group_B</td><td>August</td><td>Thompsonfurt</td><td>Vancouver</td><td>Vancouver</td></tr>\n",
       "<tr><td>002b638849be490aa...</td><td>Group_B</td><td>August</td><td>Lake Elizabethburgh</td><td>null</td><td>Lake Elizabethburgh</td></tr>\n",
       "<tr><td>002c28b399c148dc9...</td><td>Group_B</td><td>August</td><td>Douglasview</td><td>null</td><td>Douglasview</td></tr>\n",
       "<tr><td>002ceec7ee22437aa...</td><td>Group_B</td><td>August</td><td>New York</td><td>Jessicabury</td><td>Jessicabury</td></tr>\n",
       "<tr><td>002ceec7ee22437aa...</td><td>Group_B</td><td>August</td><td>New York</td><td>West Monicaland</td><td>West Monicaland</td></tr>\n",
       "<tr><td>002e5b9d28df45ebb...</td><td>Group_B</td><td>August</td><td>New York</td><td>North Darren</td><td>North Darren</td></tr>\n",
       "<tr><td>00308fefd9374516b...</td><td>Group_B</td><td>August</td><td>South William</td><td>null</td><td>South William</td></tr>\n",
       "<tr><td>0037966565704268a...</td><td>Group_B</td><td>August</td><td>North Karenchester</td><td>North Darren</td><td>North Darren</td></tr>\n",
       "<tr><td>0039f91f6b5148f49...</td><td>Group_B</td><td>August</td><td>South William</td><td>null</td><td>South William</td></tr>\n",
       "<tr><td>00456b2aba5645b7a...</td><td>Group_B</td><td>August</td><td>Rodrigueztown</td><td>null</td><td>Rodrigueztown</td></tr>\n",
       "<tr><td>004a75aaa5174de4a...</td><td>Group_B</td><td>August</td><td>Williamsonmouth</td><td>null</td><td>Williamsonmouth</td></tr>\n",
       "<tr><td>004deb72cf3d445b9...</td><td>Group_B</td><td>August</td><td>New York</td><td>null</td><td>New York</td></tr>\n",
       "<tr><td>0050724cd03a4f729...</td><td>Group_B</td><td>August</td><td>Austinfort</td><td>North Darren</td><td>North Darren</td></tr>\n",
       "<tr><td>0051fb318521432f8...</td><td>Group_B</td><td>August</td><td>Bikaner</td><td>null</td><td>Bikaner</td></tr>\n",
       "<tr><td>005b01180a174af28...</td><td>Group_B</td><td>August</td><td>West Joshuashire</td><td>null</td><td>West Joshuashire</td></tr>\n",
       "</table>\n",
       "only showing top 20 rows\n"
      ],
      "text/plain": [
       "+--------------------+-------+---------------+-------------------+----------------+-------------------+\n",
       "|             user_id|  group|promotion_month|          home_city|recommended_city|     promotion_city|\n",
       "+--------------------+-------+---------------+-------------------+----------------+-------------------+\n",
       "|00009eb59e5143159...|Group_B|         August|    East Brianmouth|            null|    East Brianmouth|\n",
       "|0001623b9afc43748...|Group_B|         August|           Haridwar|     Jessicabury|        Jessicabury|\n",
       "|00061b403fd84d91a...|Group_B|         August|        Vieja Tonga|            null|        Vieja Tonga|\n",
       "|002583a3828e49079...|Group_B|         August|       Port Richard|            null|       Port Richard|\n",
       "|00265da8830d489aa...|Group_B|         August|       New Jennifer|            null|       New Jennifer|\n",
       "|00266dbf306f478ba...|Group_B|         August|       Thompsonfurt|       Vancouver|          Vancouver|\n",
       "|002b638849be490aa...|Group_B|         August|Lake Elizabethburgh|            null|Lake Elizabethburgh|\n",
       "|002c28b399c148dc9...|Group_B|         August|        Douglasview|            null|        Douglasview|\n",
       "|002ceec7ee22437aa...|Group_B|         August|           New York|     Jessicabury|        Jessicabury|\n",
       "|002ceec7ee22437aa...|Group_B|         August|           New York| West Monicaland|    West Monicaland|\n",
       "|002e5b9d28df45ebb...|Group_B|         August|           New York|    North Darren|       North Darren|\n",
       "|00308fefd9374516b...|Group_B|         August|      South William|            null|      South William|\n",
       "|0037966565704268a...|Group_B|         August| North Karenchester|    North Darren|       North Darren|\n",
       "|0039f91f6b5148f49...|Group_B|         August|      South William|            null|      South William|\n",
       "|00456b2aba5645b7a...|Group_B|         August|      Rodrigueztown|            null|      Rodrigueztown|\n",
       "|004a75aaa5174de4a...|Group_B|         August|    Williamsonmouth|            null|    Williamsonmouth|\n",
       "|004deb72cf3d445b9...|Group_B|         August|           New York|            null|           New York|\n",
       "|0050724cd03a4f729...|Group_B|         August|         Austinfort|    North Darren|       North Darren|\n",
       "|0051fb318521432f8...|Group_B|         August|            Bikaner|            null|            Bikaner|\n",
       "|005b01180a174af28...|Group_B|         August|   West Joshuashire|            null|   West Joshuashire|\n",
       "+--------------------+-------+---------------+-------------------+----------------+-------------------+\n",
       "only showing top 20 rows"
      ]
     },
     "execution_count": 69,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "promotion_month = 'August'\n",
    "recommended_cities = predict_cities(group_B, model, id_to_index, user_mapping, \n",
    "                                    context_mapping, 10, promotion_month)\n",
    "\n",
    "promotions = (group_B.join(recommended_cities, ['user_id'], 'left')\n",
    "             .select('user_id', \n",
    "                     f.lit('Group_B').alias('group'),\n",
    "                     f.lit(promotion_month).alias('promotion_month'),\n",
    "                     'home_city',\n",
    "                     'recommended_city',\n",
    "                     f.when(f.col('recommended_city').isNull(), f.col('home_city')).otherwise(f.col('recommended_city')).alias('promotion_city'),\n",
    "                    )\n",
    "             .distinct()\n",
    "             .orderBy('user_id', 'promotion_city')\n",
    "             )\n",
    "\n",
    "promotions              "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ee77657d",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}