notebooks/Step 07 - Cloudant Datastore Recommender.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Important:** Before running this notebook, ensure you have installed spark-cloudant 1.6.4 by running the notebook: **Step 05 - Install Spark Cloudant**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, add your cloudant credentials below, delete the hash before the 'echo' command and run the cell to save your credentials"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "! # echo '{ \"username\": \"changeme\", \"password\": \"changeme\", \"host\": \"changeme\", \"port\": 443, \"url\": \"changeme\" }' > cloudant_credentials.json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "! python -c 'import cloudant' || pip install cloudant --user\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# utility method for timestamps\n",
    "import time\n",
    "def ts():\n",
    "    return time.strftime(\"%Y-%m-%d %H:%M:%S %Z\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# utility method for logging\n",
    "log4jLogger = sc._jvm.org.apache.log4j\n",
    "LOGGER = log4jLogger.LogManager.getLogger(\"CloudantRecommender\")\n",
    "\n",
    "def info(*args):\n",
    "    \n",
    "    # sends output to notebook\n",
    "    print(args)\n",
    "    \n",
    "    # sends output to kernel log file\n",
    "    LOGGER.info(args)\n",
    "    \n",
    "def error(*args):\n",
    "    \n",
    "    # sends output to notebook\n",
    "    print(args)\n",
    "    \n",
    "    # sends output to kernel log file\n",
    "    LOGGER.error(args)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# utility class for holding cloudant connection details\n",
    "import json\n",
    "\n",
    "def set_attr_if_exists(obj, data, k):\n",
    "    try:\n",
    "        setattr(obj, k, data[k])\n",
    "    except AttributeError:\n",
    "        pass\n",
    "\n",
    "class CloudantConfig:\n",
    "    def __init__(self, database, json_file=None, host=None, username=None, password=None):\n",
    "       \n",
    "        self.database = database\n",
    "        self.host = None\n",
    "        self.username = None\n",
    "        self.password = None\n",
    "\n",
    "        with open(json_file) as data_file:    \n",
    "            data = json.load(data_file)\n",
    "            \n",
    "            set_attr_if_exists(self, data, 'host')\n",
    "            set_attr_if_exists(self, data, 'username')\n",
    "            set_attr_if_exists(self, data, 'password')\n",
    "        \n",
    "        # override json attributes if provided\n",
    "        if host:     self.host = host\n",
    "        if username: self.username = username\n",
    "        if password: self.password = password"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sourceDB = CloudantConfig(\n",
    "                    json_file='cloudant_credentials.json', \n",
    "                    database=\"ratingdb\"\n",
    "                    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " - We generate recommendations, create a new Cloudant database for the recommendations and save them into the new Cloudant database.\n",
    " - When we have finished writing the recommendations to Cloudant, we save a metadata record into the recommendation_meta database with the name of the new database.\n",
    " - Client applications use the metadata record to determine which database to retrieve the recommendations from.\n",
    " - We delete older databases after writing the metadata, but keep the five latest ones.\n",
    " - We need to keep at least one database because if a client reads the meta pointing to the previous database it will try to read from that database.\n",
    " - We don't have just one database and continually update the recommendation records in Cloudant because lots of changes can be considered an anti-pattern.\n",
    " - The recommendation_meta database is created for us by the web application setup scripts.\n",
    " - The spark-cloudant package is used to read the data from Cloudant but not to write the data to Cloudant because of this issue: https://github.com/cloudant-labs/spark-cloudant/issues/82\n",
    " - The python-cloudant package is used to write the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating\n",
    "from pyspark.sql.functions import monotonically_increasing_id\n",
    "\n",
    "import json\n",
    "import numpy as np\n",
    "\n",
    "# we use the cloudant python library to save the recommendations\n",
    "from cloudant.client import Cloudant\n",
    "from cloudant.adapters import Replay429Adapter\n",
    "\n",
    "class CloudantMovieRecommender:\n",
    "    \n",
    "    def __init__(self, sc):\n",
    "        self.sc = sc\n",
    "    \n",
    "    def train(self, sourceDB):\n",
    "                      \n",
    "        info(\"Starting load from Cloudant: \", ts())\n",
    "\n",
    "        dfReader = sqlContext.read.format(\"com.cloudant.spark\")\n",
    "        dfReader.option(\"cloudant.host\", sourceDB.host)\n",
    "        dfReader.option(\"schemaSampleSize\", -1)\n",
    "        \n",
    "        if sourceDB.username:\n",
    "            dfReader.option(\"cloudant.username\", sourceDB.username)\n",
    "            \n",
    "        if sourceDB.password:\n",
    "            dfReader.option(\"cloudant.password\", sourceDB.password)\n",
    "            \n",
    "        df = dfReader.load(sourceDB.database).cache()\n",
    "        \n",
    "        from pyspark.sql.functions import udf\n",
    "        from pyspark.sql.types import IntegerType, StringType, StructType, StructField\n",
    "        \n",
    "        # the Cloudant userId may be a string type\n",
    "        def get_cloudant_user_id(data):\n",
    "            (user_id, _) = data.split('/')\n",
    "            return user_id.replace('user_', '')\n",
    "        \n",
    "        # extract the user id from the cloudant record\n",
    "        cloudant_user_id = udf(lambda _id: get_cloudant_user_id(_id), StringType())\n",
    "        df = df.withColumn(\"cloudant_user_id\", cloudant_user_id(df['_id']))\n",
    "\n",
    "        def get_movie_id(data):\n",
    "            (_, movie_id) = data.split('/')\n",
    "            return long(movie_id.replace('movie_', ''))\n",
    "        \n",
    "        # extract the movie id from the cloudant record\n",
    "        movie_id = udf(lambda _id: get_movie_id(_id), IntegerType())\n",
    "        df = df.withColumn(\"movie_id\", movie_id(df['_id']))     \n",
    "        \n",
    "        # The Cloudant user Id is a string type (uuid), but ALS requires an int type\n",
    "        # so we create a new integer user ID field here.  The generated recommendations\n",
    "        # will have the new user ID, so we will need to map it back to the original \n",
    "        # Cloudant user Id when we have finished.\n",
    "        \n",
    "        df_user_ids = df.select(\"cloudant_user_id\").distinct().cache()\n",
    "        \n",
    "        # create a new schema with user_id field appended\n",
    "        newSchema = StructType([StructField(\"als_user_id\", IntegerType(), False)]\n",
    "                       + df_user_ids.schema.fields)\n",
    "        \n",
    "        # zip with the index, map it to a dictionary which includes new field\n",
    "        df_user_ids = df_user_ids.rdd.zipWithIndex()\\\n",
    "                              .map(lambda (row, id): {k:v\n",
    "                                                      for k, v\n",
    "                                                      in row.asDict().items() + [(\"als_user_id\", id)]})\\\n",
    "                              .toDF(newSchema)\n",
    "        \n",
    "        # add the new user id (als_user_id) to the dataframe\n",
    "        df = df.join(df_user_ids, df.cloudant_user_id == df_user_ids.cloudant_user_id)\n",
    "        \n",
    "        # save the dataframe as we will need to map back from the\n",
    "        # new user id to the orgional Cloudant user id.\n",
    "        self.df_user_ids = df_user_ids.cache()\n",
    "        \n",
    "        info(\"Finished load from Cloudant: \", ts())\n",
    "        info(\"Found\", df.count(), \"records in Cloudant\")\n",
    "        \n",
    "        # convert cloudant docs into Rating objects\n",
    "        ratings = df.map( lambda row: Rating(row.als_user_id, row.movie_id, float(row.rating)) )\n",
    "        \n",
    "        rank = 50\n",
    "        numIterations = 20\n",
    "        lambdaParam = 0.1\n",
    "\n",
    "        info(\"Starting train model: \", ts())\n",
    "        self.model = ALS.train(ratings, rank, numIterations, lambdaParam)\n",
    "        info(\"Finished train model: \", ts())\n",
    "        \n",
    "    def get_top_recommendations(self):\n",
    "        info(\"Starting __get_top_recommendations: \", ts())\n",
    "        df = self.model.recommendProductsForUsers(10).toDF()\n",
    "        df.cache()\n",
    "        info(\"Finished __get_top_recommendations: \", ts())\n",
    "        return df\n",
    "        \n",
    "    def del_old_recommendationdbs(self, cloudant_client, db_name_prefix):\n",
    "        dbs_to_del = cloudant_client.all_dbs()\n",
    "\n",
    "        # only delete dbs we are using for recommendations\n",
    "        dbs_to_del = [db for db in dbs_to_del if db.startswith(db_name_prefix + '_') ]\n",
    "\n",
    "        # ensure the list is in timestamp order\n",
    "        dbs_to_del.sort()\n",
    "\n",
    "        # keeping the last 5 dbs and delete the rest\n",
    "        for db in dbs_to_del[:-5]:\n",
    "            cloudant_client.delete_database(db)\n",
    "            info(\"Deleted old recommendations db\", db)\n",
    "            \n",
    "    def update_meta_document(self, cloudant_client, meta_db_name, latest_db_name):\n",
    "        \n",
    "        meta_db = cloudant_client[meta_db_name]\n",
    "        \n",
    "        from datetime import datetime\n",
    "        ts = datetime.utcnow().isoformat()\n",
    "\n",
    "        try:\n",
    "            # update doc if exists\n",
    "            meta_doc = meta_db['recommendation_metadata']\n",
    "            meta_doc['latest_db'] = latest_db_name\n",
    "            meta_doc['timestamp_utc'] = ts\n",
    "            meta_doc.save()\n",
    "            info(\"Updated recommendationdb metadata record with latest_db\", latest_db_name, meta_doc)\n",
    "        except KeyError:\n",
    "            # create a new doc\n",
    "            data = {\n",
    "                '_id': 'recommendation_metadata',\n",
    "                'latest_db': latest_db_name,\n",
    "                'timestamp_utc': ts,\n",
    "                }\n",
    "            meta_doc = meta_db.create_document(data)\n",
    "            meta_doc.save()\n",
    "            \n",
    "            if meta_doc.exists():\n",
    "                info(\"Saved recommendationdb metadata record\", str(data))\n",
    "                \n",
    "        # save product features to enable later generationg of Vt\n",
    "        # see: http://stackoverflow.com/questions/41537470/als-model-how-to-generate-full-u-vt-v\n",
    "        pf = self.model.productFeatures().sortByKey()\n",
    "\n",
    "        pf_keys = json.dumps(pf.sortByKey().keys().collect())\n",
    "        pf_vals = json.dumps(pf.sortByKey().map(lambda x: list(x[1])).collect())               \n",
    "        \n",
    "        # the pf_keys/pf_vals are too big and exceed the >1mb document size limit\n",
    "        # so we save them as attachments\n",
    "        \n",
    "        meta_doc.put_attachment(\n",
    "            attachment='product_feature_keys', \n",
    "            content_type='application/json', \n",
    "            data=pf_keys\n",
    "        )\n",
    "\n",
    "        meta_doc.put_attachment(\n",
    "            attachment='product_feature_vals', \n",
    "            content_type='application/json', \n",
    "            data=pf_vals\n",
    "        )\n",
    "    \n",
    "    def create_recommendationdb(self, cloudant_client):\n",
    "        # create a database for recommendations\n",
    "        import time\n",
    "        db_name = destDB.database + '_' + str(int(time.time()))\n",
    "        \n",
    "        db = cloudant_client.create_database(db_name)\n",
    "        info(\"Created new recommendations db\", db_name)\n",
    "        return db\n",
    "        \n",
    "    def save_recommendations(self, destDB):\n",
    "        df = movieRecommender.get_top_recommendations()\n",
    "        \n",
    "        cloudant_client = Cloudant(\n",
    "                                destDB.username,\n",
    "                                destDB.password,\n",
    "                                account=destDB.username, \n",
    "                                adapter=Replay429Adapter(retries=10, initialBackoff=1)\n",
    "                                )\n",
    "        cloudant_client.connect()\n",
    "        self.del_old_recommendationdbs(cloudant_client, destDB.database)\n",
    "        recommendations_db = self.create_recommendationdb(cloudant_client)\n",
    "\n",
    "        # get the user ids stored in cloudant so that we can map them back\n",
    "        user_id_map = {row[0]: row[1] for row in self.df_user_ids.collect() }\n",
    "\n",
    "        # reformat data for saving\n",
    "        docs = df.map(lambda x: {'_id':str(user_id_map[x[0]]), 'recommendations':x[1]}).collect()\n",
    "        \n",
    "        # we could hit cloudant resource limits if trying to save entire doc\n",
    "        # so we save it in smaller sized chunks\n",
    "        \n",
    "        for i in range(0, len(docs), 100):\n",
    "            chunk = docs[i:i + 100]\n",
    "            recommendations_db.bulk_docs(chunk) # TODO check for errors saving the chunk\n",
    "            info(\"Saved recommendations chunk\", i, ts())\n",
    "        \n",
    "        self.update_meta_document(cloudant_client, destDB.database, recommendations_db.database_name)\n",
    "        \n",
    "        info(\"Saved recommendations to: \", recommendations_db.database_name, ts())\n",
    "\n",
    "        cloudant_client.disconnect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now the code to start the recommender ..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sourceDB = CloudantConfig(\n",
    "                    json_file='cloudant_credentials.json', \n",
    "                    database=\"ratingdb\"\n",
    "                    )\n",
    "\n",
    "destDB = CloudantConfig(\n",
    "                    json_file='cloudant_credentials.json', \n",
    "                    database=\"recommendationdb\", \n",
    "                    )\n",
    "\n",
    "import traceback\n",
    "try:\n",
    "    movieRecommender = CloudantMovieRecommender(sc)\n",
    "    movieRecommender.train(sourceDB)\n",
    "    movieRecommender.save_recommendations(destDB)\n",
    "except Exception as e:\n",
    "    error(str(e), traceback.format_exc(), ts())\n",
    "    raise e"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scheduling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After you have successfully run the notebook interactively you can schedule your notebook to run on a timer, for example hourly.\n",
    "\n",
    "Timer jobs work again a specific saved version of a notebook.  If you haven't saved a version when you first create the schedule a version will be saved for you.  \n",
    "\n",
    "Note that if you make changes to your notebook, you need to save a new version and re-schedule the notebook with the new version of the notebook selected in the schedule configuration form."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## For debugging issues"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Get the local time on the cluster\n",
    "! date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# dump the latest kernel log\n",
    "! cat $(ls -1 $HOME/logs/notebook/*pyspark* | sort -r | head -1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# look for our log output in the latest kernel log file\n",
    "! grep 'CloudantRecommender' $(ls -1 $HOME/logs/notebook/*pyspark* | sort -r | head -1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# look for our log output in all kernel log files\n",
    "! grep 'CloudantRecommender' $HOME/logs/notebook/*pyspark* "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "! ls -l $HOME/logs/notebook/*pyspark* "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2 with Spark 1.6",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}