Load_and_Train_new.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import print_function\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "from pyspark.ml import Pipeline, PipelineModel\n",
    "from pyspark.sql.functions import udf, round, col\n",
    "from pyspark.sql.types import ArrayType, StringType\n",
    "from pyspark.ml.feature import Tokenizer, StopWordsRemover, VectorAssembler, CountVectorizer, IDF, IDFModel, CountVectorizerModel, StringIndexer\n",
    "from nltk.stem import PorterStemmer\n",
    "from torchtext.vocab import GloVe\n",
    "import numpy as np\n",
    "from functools import reduce\n",
    "from datetime import datetime\n",
    "from google.cloud import storage\n",
    "\n",
    "# from pyspark.mllib.tree import GradientBoostedTrees\n",
    "from pyspark.ml.classification import LogisticRegression\n",
    "\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "from pyspark.mllib.evaluation import MulticlassMetrics\n",
    "from io import BytesIO\n",
    "from tensorflow.python.lib.io import file_io\n",
    "from pyspark.mllib.linalg import Vector as MLLibVector, Vectors as MLLibVectors\n",
    "from pyspark.ml.linalg import Vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate()\n",
    "spark = SparkSession(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load Data\n",
    "full_data = spark.read.json('gs://bdl_project/yelp.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "raw_data = full_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "bucket = 'bdl_project'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Loading and applying all pre-trained models\n",
    "indexer = PipelineModel.load('gs://' + bucket + '/indexer')\n",
    "raw_data = indexer.transform(raw_data)\n",
    "raw_data = raw_data.drop('business_id', 'user_id')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "pre_process = PipelineModel.load('gs://' + bucket + '/pre_process')\n",
    "data = pre_process.transform(raw_data)\n",
    "data = data.drop('text', 'tokenized_text')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Perform stemming\n",
    "ps = PorterStemmer()\n",
    "stemmer = udf(lambda text: [ps.stem(token) for token in text], ArrayType(StringType()))\n",
    "data = data.withColumn('stemmed_text', stemmer('filtered_text'))\n",
    "data = data.drop('filtered_text')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Train test split\n",
    "(train, test) = data.randomSplit([0.95, 0.05], seed=5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "cv_model = CountVectorizerModel().load('gs://' + bucket + '/cv_model')  \n",
    "cvData = cv_model.transform(train)\n",
    "idf_model = IDFModel().load('gs://' + bucket + '/idf_model')\n",
    "tfidfData = idf_model.transform(cvData)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['cool',\n",
       " 'date',\n",
       " 'funny',\n",
       " 'review_id',\n",
       " 'stars',\n",
       " 'useful',\n",
       " 'business_ind',\n",
       " 'user_ind',\n",
       " 'stemmed_text',\n",
       " 'tf',\n",
       " 'tf_idf']"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tfidfData.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "f = BytesIO(file_io.read_file_to_string('gs://nithya1998/glove.npy', binary_mode=True))\n",
    "glove = np.load(f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#embedding_glove = GloVe(name='6B', dim=100)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get vocabulary and word index\n",
    "vocab = cv_model.vocabulary\n",
    "ind2word = dict(zip(range(len(vocab)), vocab))\n",
    "vocab_len = len(vocab)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "262144"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "vocab_len"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Obtain glove stacking for the words in the vocabulary\n",
    "\n",
    "# for i in range(vocab_len):\n",
    "#     if i ==0:\n",
    "#         glove = embedding_glove[vocab[i]].numpy()\n",
    "#     else:\n",
    "#         glove = np.vstack((glove, embedding_glove[vocab[i]].numpy()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Feature engineering to obtain sentence representation using weighted glove representations\n",
    "glove_cols = tfidfData.rdd.map(lambda x:(x[0], x[1], x[2], x[3], x[4], x[5], x[6], x[7], x[8], x[9], Vectors.dense(x[10].dot(glove)/x[10].dot(np.ones(vocab_len))))).toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Renaming glove column names with original column names\n",
    "oldColumns = glove_cols.schema.names\n",
    "newColumns = tfidfData.schema.names[:-1] + [\"glove_emb\"]\n",
    "\n",
    "train = reduce(lambda glove_cols, idx: glove_cols.withColumnRenamed(oldColumns[idx], newColumns[idx]), range(len(oldColumns)), glove_cols)\n",
    "train = train.drop(\"lemmatized_text\", \"tf\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Done\n"
     ]
    }
   ],
   "source": [
    "print('Done')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "bucket = 'nithya1998'\n",
    "min_date = datetime.now()\n",
    "client = storage.Client()\n",
    "bucket_obj = client.get_bucket(bucket)\n",
    "blob = bucket_obj.blob('min_date.txt')\n",
    "blob.upload_from_string(str(min_date))\n",
    "\n",
    "delta_days = udf(lambda x: int((min_date-datetime.strptime(x, '%Y-%m-%d %H:%M:%S')).days))\n",
    "train = train.withColumn('delta_days', delta_days('date'))\n",
    "train = train.drop('date')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import IntegerType\n",
    "train = train.withColumn(\"delta_days\",train[\"delta_days\"].cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- cool: long (nullable = true)\n",
      " |-- funny: long (nullable = true)\n",
      " |-- review_id: string (nullable = true)\n",
      " |-- stars: double (nullable = true)\n",
      " |-- useful: long (nullable = true)\n",
      " |-- business_ind: double (nullable = true)\n",
      " |-- user_ind: double (nullable = true)\n",
      " |-- stemmed_text: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- glove_emb: vector (nullable = true)\n",
      " |-- delta_days: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "train.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## LR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "bucket_store = <your-bucket>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "assembler = VectorAssembler(inputCols = ['cool','funny','useful','glove_emb','delta_days','business_ind','user_ind'], outputCol=\"features\")\n",
    "train = assembler.transform(train).select(\"features\", \"stars\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr_train = train.rdd.map(lambda x: LabeledPoint(x.stars, MLLibVectors.fromML(x.features))).toDF()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr_train.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)\n",
    "model = lr.fit(lr_train)\n",
    "model.save('gs://' + bucket_store + '/lr_model')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Trained and saved')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "trainingSummary = model.summary\n",
    "accuracy = trainingSummary.accuracy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print('Accuracy =', accuracy)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# XGBOOST"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "xgbassembler = VectorAssembler(inputCols = ['cool','funny','useful','glove_emb','delta_days','business_ind','user_ind'], outputCol=\"features\")\n",
    "train = xgbassembler.transform(train).select(\"features\", \"stars\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The input to xgboost needs to be an RDD of LabeledPoints and the feature vector needs to be of type mllib.linalg.vector\n",
    "xgb_train = train.rdd.map(lambda x: LabeledPoint(x.stars, MLLibVectors.fromML(x.features)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Done\n"
     ]
    }
   ],
   "source": [
    "print('Done')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#xgb_train.toDF().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#df.agg({\"your-column\": \"max\"}).collect()[0][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-26-136b8334d198>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0mnum_bus\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;36m209393\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0mnum_users\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;36m1968703\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 4\u001b[0;31m \u001b[0mmodel\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mGradientBoostedTrees\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtrainRegressor\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mxgb_train\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcategoricalFeaturesInfo\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m{\u001b[0m\u001b[0;36m104\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0mnum_bus\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m105\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0mnum_users\u001b[0m\u001b[0;34m}\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mnumIterations\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m3\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      5\u001b[0m \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'Model trained'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/spark/python/pyspark/mllib/tree.py\u001b[0m in \u001b[0;36mtrainRegressor\u001b[0;34m(cls, data, categoricalFeaturesInfo, loss, numIterations, learningRate, maxDepth, maxBins)\u001b[0m\n\u001b[1;32m    641\u001b[0m         \"\"\"\n\u001b[1;32m    642\u001b[0m         return cls._train(data, \"regression\", categoricalFeaturesInfo,\n\u001b[0;32m--> 643\u001b[0;31m                           loss, numIterations, learningRate, maxDepth, maxBins)\n\u001b[0m\u001b[1;32m    644\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    645\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/spark/python/pyspark/mllib/tree.py\u001b[0m in \u001b[0;36m_train\u001b[0;34m(cls, data, algo, categoricalFeaturesInfo, loss, numIterations, learningRate, maxDepth, maxBins)\u001b[0m\n\u001b[1;32m    504\u001b[0m         \u001b[0;32massert\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfirst\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mLabeledPoint\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"the data should be RDD of LabeledPoint\"\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    505\u001b[0m         model = callMLlibFunc(\"trainGradientBoostedTreesModel\", data, algo, categoricalFeaturesInfo,\n\u001b[0;32m--> 506\u001b[0;31m                               loss, numIterations, learningRate, maxDepth, maxBins)\n\u001b[0m\u001b[1;32m    507\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mGradientBoostedTreesModel\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmodel\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    508\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/spark/python/pyspark/mllib/common.py\u001b[0m in \u001b[0;36mcallMLlibFunc\u001b[0;34m(name, *args)\u001b[0m\n\u001b[1;32m    128\u001b[0m     \u001b[0msc\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mSparkContext\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgetOrCreate\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    129\u001b[0m     \u001b[0mapi\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mgetattr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPythonMLLibAPI\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mname\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 130\u001b[0;31m     \u001b[0;32mreturn\u001b[0m \u001b[0mcallJavaFunc\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mapi\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m*\u001b[0m\u001b[0margs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    131\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    132\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/usr/lib/spark/python/pyspark/mllib/common.py\u001b[0m in \u001b[0;36mcallJavaFunc\u001b[0;34m(sc, func, *args)\u001b[0m\n\u001b[1;32m    121\u001b[0m     \u001b[0;34m\"\"\" Call Java Function \"\"\"\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    122\u001b[0m     \u001b[0margs\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;34m[\u001b[0m\u001b[0m_py2java\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0ma\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mfor\u001b[0m \u001b[0ma\u001b[0m \u001b[0;32min\u001b[0m \u001b[0margs\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 123\u001b[0;31m     \u001b[0;32mreturn\u001b[0m \u001b[0m_java2py\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msc\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfunc\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m*\u001b[0m\u001b[0margs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    124\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    125\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/anaconda/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1253\u001b[0m             \u001b[0mproto\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mEND_COMMAND_PART\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1254\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1255\u001b[0;31m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1256\u001b[0m         return_value = get_return_value(\n\u001b[1;32m   1257\u001b[0m             answer, self.gateway_client, self.target_id, self.name)\n",
      "\u001b[0;32m/opt/conda/anaconda/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36msend_command\u001b[0;34m(self, command, retry, binary)\u001b[0m\n\u001b[1;32m    983\u001b[0m         \u001b[0mconnection\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_get_connection\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    984\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 985\u001b[0;31m             \u001b[0mresponse\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mconnection\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    986\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mbinary\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    987\u001b[0m                 \u001b[0;32mreturn\u001b[0m \u001b[0mresponse\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_create_connection_guard\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mconnection\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/anaconda/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36msend_command\u001b[0;34m(self, command)\u001b[0m\n\u001b[1;32m   1150\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1151\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1152\u001b[0;31m             \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0msmart_decode\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstream\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mreadline\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m-\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1153\u001b[0m             \u001b[0mlogger\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdebug\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Answer received: {0}\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0manswer\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1154\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0manswer\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mproto\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mRETURN_MESSAGE\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/anaconda/lib/python3.7/socket.py\u001b[0m in \u001b[0;36mreadinto\u001b[0;34m(self, b)\u001b[0m\n\u001b[1;32m    587\u001b[0m         \u001b[0;32mwhile\u001b[0m \u001b[0;32mTrue\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    588\u001b[0m             \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 589\u001b[0;31m                 \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_sock\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrecv_into\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mb\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    590\u001b[0m             \u001b[0;32mexcept\u001b[0m \u001b[0mtimeout\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    591\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_timeout_occurred\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mTrue\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "# categoricalFeaturesInfo is a dict with key as column index of categorical variable and value as no. of classes\n",
    "num_bus = 209393\n",
    "num_users = 1968703\n",
    "model = GradientBoostedTrees.trainRegressor(xgb_train, categoricalFeaturesInfo={104:num_bus, 105:num_users}, numIterations=3)\n",
    "print('Model trained')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model.save('gs://nithya1998/xgb_model')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_predictions = model.predict(xgb_train.map(lambda x: x.features))\n",
    "stars_and_pred = train.rdd.map(lambda lp: lp.stars).zip(train_predictions).toDF()\n",
    "stars_and_pred = stars_and_pred.withColumn('pred', round(col('_2'))).withColumnRenamed('_1', 'stars')\n",
    "stars_and_pred = stars_and_pred.select('stars','pred')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "metrics_multi = MulticlassMetrics(stars_and_pred.rdd.map(tuple))\n",
    "acc = metrics_multi.accuracy\n",
    "print(\"Training Accuracy =\", acc)"
   ]
  }
 ],
 "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}