movielens-memsql.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.sql import Row\n",
    "from pyspark.sql.functions import udf, log\n",
    "from pyspark.sql.types import StringType\n",
    "from pyspark import SparkConf, SparkContext, sql\n",
    "\n",
    "import pandas as pd\n",
    "from sqlalchemy import create_engine\n",
    "\n",
    "import struct\n",
    "import binascii\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName('ops').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100 5778k  100 5778k    0     0   842k      0  0:00:06  0:00:06 --:--:-- 1133k\n"
     ]
    }
   ],
   "source": [
    "!curl http://files.grouplens.org/datasets/movielens/ml-1m.zip --output ml-1m.zip"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Archive:  ml-1m.zip\n",
      "  inflating: ml-1m/movies.dat        \n",
      "  inflating: ml-1m/ratings.dat       \n",
      "  inflating: ml-1m/README            \n",
      "  inflating: ml-1m/users.dat         \n"
     ]
    }
   ],
   "source": [
    "!unzip -o ml-1m.zip "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 162,
   "metadata": {},
   "outputs": [],
   "source": [
    "movies = pd.read_csv('ml-1m/movies.dat', sep=\"::\", header=None, names=['MovieID','Title','Genres'], engine='python')\n",
    "users = pd.read_csv('ml-1m/users.dat', sep=\"::\", header=None, names=['UserID','Gender','Age','Occupation','Zip-code'], engine='python')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 163,
   "metadata": {},
   "outputs": [],
   "source": [
    "movies['year'] = movies['Title'].str.extract('\\(([0-9]{4})\\)', expand=False).str.strip()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 142,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = spark.read.text('ml-1m/ratings.dat').rdd\n",
    "parts = lines.map(lambda row: row.value.split('::'))\n",
    "\n",
    "ratingsRDD = parts.map(lambda p: Row(userID=int(p[0]),movieID=int(p[1]),rating=int(p[2])))\n",
    "ratings = spark.createDataFrame(ratingsRDD)\n",
    "\n",
    "(training, test) = ratings.randomSplit([0.8, 0.2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+------+\n",
      "|movieID|rating|userID|\n",
      "+-------+------+------+\n",
      "|   1193|     5|     1|\n",
      "|    661|     3|     1|\n",
      "|    914|     3|     1|\n",
      "|   3408|     4|     1|\n",
      "|   2355|     5|     1|\n",
      "|   1197|     3|     1|\n",
      "|   1287|     5|     1|\n",
      "|   2804|     5|     1|\n",
      "|    594|     4|     1|\n",
      "|    919|     4|     1|\n",
      "|    595|     5|     1|\n",
      "|    938|     4|     1|\n",
      "|   2398|     4|     1|\n",
      "|   2918|     4|     1|\n",
      "|   1035|     5|     1|\n",
      "|   2791|     4|     1|\n",
      "|   2687|     3|     1|\n",
      "|   2018|     4|     1|\n",
      "|   3105|     5|     1|\n",
      "|   2797|     4|     1|\n",
      "+-------+------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ratings.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 144,
   "metadata": {},
   "outputs": [],
   "source": [
    "als = ALS(maxIter=5, regParam=0.01, userCol=\"userID\", itemCol=\"movieID\", ratingCol=\"rating\",\n",
    "          coldStartStrategy=\"drop\")\n",
    "\n",
    "model = als.fit(training)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 145,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root-mean-square error = 0.8941641483795894\n"
     ]
    }
   ],
   "source": [
    "predictions = model.transform(test)\n",
    "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\",\n",
    "                                predictionCol=\"prediction\")\n",
    "rmse = evaluator.evaluate(predictions)\n",
    "print(\"Root-mean-square error = \" + str(rmse))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [],
   "source": [
    "itemfactors = spark.createDataFrame(model.itemFactors.rdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [],
   "source": [
    "userfactors = spark.createDataFrame(model.userFactors.rdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {},
   "outputs": [],
   "source": [
    "def vector2hex(vector):\n",
    "    vectorStr = b\"\".join([struct.pack('f', elem) for elem in vector])\n",
    "    return str(binascii.hexlify(vectorStr))[2:-1] # not to elegant but this works\n",
    "\n",
    "udf_vector2hex = udf(vector2hex, StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [],
   "source": [
    "itemfactors_with_hex = itemfactors.withColumn(\"factors\", udf_vector2hex(\"features\"))\n",
    "userfactors_with_hex = userfactors.withColumn(\"factors\", udf_vector2hex(\"features\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {},
   "outputs": [],
   "source": [
    "items_frame = itemfactors_with_hex.select('id','factors').toPandas().rename(columns={\"id\": \"movie_id\", \"factors\": \"factors\"})\n",
    "users_frame = userfactors_with_hex.select('id','factors').toPandas().rename(columns={\"id\": \"user_id\", \"factors\": \"factors\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {},
   "outputs": [],
   "source": [
    "db_users = users.merge(users_frame, left_on='UserID', right_on='user_id').drop(columns=['user_id'])\n",
    "db_users.rename(\n",
    "    columns={'UserID':'id','Gender':'gender','Age':'age','Occupation':'occupation','Zip-code':'zip_code','factors':'factors'}, inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 152,
   "metadata": {},
   "outputs": [],
   "source": [
    "db_movies = movies.merge(items_frame, left_on='MovieID', right_on='movie_id').drop(columns=['movie_id'])\n",
    "db_movies.rename(\n",
    "    columns={'MovieID':'id','Title':'title','year':'year','Genres':'genres','factors':'factors'}, inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 153,
   "metadata": {},
   "outputs": [],
   "source": [
    "from memsql.common import database\n",
    "\n",
    "HOST = \"127.0.0.1\"\n",
    "PORT = 3306"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_connection(host=None, port=None, db=None):\n",
    "    \"\"\" Returns a new connection to the database. \"\"\"\n",
    "    if host is None:\n",
    "        host = HOST\n",
    "    if port is None:\n",
    "        port = PORT\n",
    "\n",
    "    return database.connect(\n",
    "        host=host,\n",
    "        port=port,\n",
    "        user='root',\n",
    "        password='',\n",
    "        database=db)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 156,
   "metadata": {},
   "outputs": [],
   "source": [
    "with get_connection(db=\"information_schema\") as conn:\n",
    "    conn.query('USE data')\n",
    "    conn.query('CREATE TABLE IF NOT EXISTS movies (id INT AUTO_INCREMENT PRIMARY KEY, title VARCHAR(255), year INT, genres VARCHAR(255), factors BINARY(80))')\n",
    "    conn.query('CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY, gender VARCHAR(5), age INT, occupation INT, zip_code VARCHAR(255), factors BINARY(80))')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sqlalchemy.ext.declarative import declarative_base\n",
    "from sqlalchemy import create_engine\n",
    "\n",
    "Base = declarative_base()\n",
    "\n",
    "engine = create_engine('mysql://root@127.0.0.1')\n",
    "engine.execute('USE data')\n",
    "Base.metadata.create_all(engine)\n",
    "\n",
    "db_movies.to_sql('movies', con=engine, if_exists='append', index=False)\n",
    "db_users.to_sql('users', con=engine, if_exists='append', index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}