algo1.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#https://grouplens.org/datasets/movielens/\n",
    "import pandas as pd\n",
    "from pyspark.sql.functions import col, explode\n",
    "from pyspark import SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "sc = SparkContext\n",
    "# sc.setCheckpointDir('checkpoint')\n",
    "spark = SparkSession.builder.appName('Recommendations').getOrCreate()"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "outputs": [],
   "source": [
    "movies = spark.read.csv(\"data/movies.csv\",header=True)\n",
    "ratings = spark.read.csv(\"data/ratings.csv\",header=True)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+\n",
      "|userId|movieId|rating|\n",
      "+------+-------+------+\n",
      "|     1|    296|   5.0|\n",
      "|     1|    306|   3.5|\n",
      "|     1|    307|   5.0|\n",
      "|     1|    665|   5.0|\n",
      "|     1|    899|   3.5|\n",
      "|     1|   1088|   4.0|\n",
      "|     1|   1175|   3.5|\n",
      "|     1|   1217|   3.5|\n",
      "|     1|   1237|   5.0|\n",
      "|     1|   1250|   4.0|\n",
      "|     1|   1260|   3.5|\n",
      "|     1|   1653|   4.0|\n",
      "|     1|   2011|   2.5|\n",
      "|     1|   2012|   2.5|\n",
      "|     1|   2068|   2.5|\n",
      "|     1|   2161|   3.5|\n",
      "|     1|   2351|   4.5|\n",
      "|     1|   2573|   4.0|\n",
      "|     1|   2632|   5.0|\n",
      "|     1|   2692|   5.0|\n",
      "+------+-------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ratings = ratings.\\\n",
    "    withColumn('userId', col('userId').cast('integer')).\\\n",
    "    withColumn('movieId', col('movieId').cast('integer')).\\\n",
    "    withColumn('rating', col('rating').cast('float')).\\\n",
    "    drop('timestamp')\n",
    "ratings.show()"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The ratings dataframe is  99.74% empty.\n"
     ]
    }
   ],
   "source": [
    "# Count the total number of ratings in the dataset\n",
    "numerator = ratings.select(\"rating\").count()\n",
    "\n",
    "# Count the number of distinct userIds and distinct movieIds\n",
    "num_users = ratings.select(\"userId\").distinct().count()\n",
    "num_movies = ratings.select(\"movieId\").distinct().count()\n",
    "\n",
    "# Set the denominator equal to the number of users multiplied by the number of movies\n",
    "denominator = num_users * num_movies\n",
    "\n",
    "# Divide the numerator by the denominator\n",
    "sparsity = (1.0 - (numerator *1.0)/denominator)*100\n",
    "print(\"The ratings dataframe is \", \"%.2f\" % sparsity + \"% empty.\")"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "outputs": [
    {
     "data": {
      "text/plain": "pyspark.ml.recommendation.ALS"
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create test and train set\n",
    "(train, test) = ratings.randomSplit([0.8, 0.2], seed = 1234)\n",
    "\n",
    "# Create ALS model\n",
    "als = ALS(userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\", nonnegative = True, implicitPrefs = False, coldStartStrategy=\"drop\")\n",
    "\n",
    "# Confirm that a model called \"als\" was created\n",
    "type(als)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Num models to be tested:  16\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\n",
    "# Add hyperparameters and their respective values to param_grid\n",
    "param_grid = ParamGridBuilder() \\\n",
    "            .addGrid(als.rank, [10, 50, 100, 150]) \\\n",
    "            .addGrid(als.regParam, [.01, .05, .1, .15]) \\\n",
    "            .build()\n",
    "            #             .addGrid(als.maxIter, [5, 50, 100, 200]) \\\n",
    "\n",
    "\n",
    "# Define evaluator as RMSE and print length of evaluator\n",
    "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\")\n",
    "print (\"Num models to be tested: \", len(param_grid))"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CrossValidator_40ae6736ee37\n"
     ]
    }
   ],
   "source": [
    "# Build cross validation using CrossValidator\n",
    "cv = CrossValidator(estimator=als, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)\n",
    "\n",
    "# Confirm cv was built\n",
    "print(cv)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR:root:Exception while sending command.\n",
      "Traceback (most recent call last):\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\", line 480, in send_command\n",
      "    raise Py4JNetworkError(\"Answer from Java side is empty\")\n",
      "py4j.protocol.Py4JNetworkError: Answer from Java side is empty\n",
      "\n",
      "During handling of the above exception, another exception occurred:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/java_gateway.py\", line 1038, in send_command\n",
      "    response = connection.send_command(command)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\", line 504, in send_command\n",
      "    \"Error while sending or receiving\", e, proto.ERROR_ON_RECEIVE)\n",
      "py4j.protocol.Py4JNetworkError: Error while sending or receiving\n",
      "ERROR:root:Exception while sending command.\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/yjx/.local/lib/python3.7/site-packages/IPython/core/interactiveshell.py\", line 3418, in run_code\n",
      "    exec(code_obj, self.user_global_ns, self.user_ns)\n",
      "  File \"<ipython-input-10-e73558e059f0>\", line 2, in <module>\n",
      "    model = cv.fit(train)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\", line 161, in fit\n",
      "    return self._fit(dataset)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/tuning.py\", line 689, in _fit\n",
      "    for j, metric, subModel in pool.imap_unordered(lambda f: f(), tasks):\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/multiprocessing/pool.py\", line 748, in next\n",
      "    raise value\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/multiprocessing/pool.py\", line 121, in worker\n",
      "    result = (True, func(*args, **kwds))\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/tuning.py\", line 689, in <lambda>\n",
      "    for j, metric, subModel in pool.imap_unordered(lambda f: f(), tasks):\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/util.py\", line 326, in wrapped\n",
      "    return f(*args, **kwargs)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/tuning.py\", line 69, in singleTask\n",
      "    index, model = next(modelIter)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\", line 69, in __next__\n",
      "    return index, self.fitSingleModel(index)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\", line 126, in fitSingleModel\n",
      "    return estimator.fit(dataset, paramMaps[index])\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\", line 159, in fit\n",
      "    return self.copy(params)._fit(dataset)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/wrapper.py\", line 335, in _fit\n",
      "    java_model = self._fit_java(dataset)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/wrapper.py\", line 332, in _fit_java\n",
      "    return self._java_obj.fit(dataset._jdf)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/java_gateway.py\", line 1322, in __call__\n",
      "    answer, self.gateway_client, self.target_id, self.name)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/sql/utils.py\", line 111, in deco\n",
      "    return f(*a, **kw)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/protocol.py\", line 328, in get_return_value\n",
      "    format(target_id, \".\", name), value)\n",
      "py4j.protocol.Py4JJavaError: <unprintable Py4JJavaError object>\n",
      "\n",
      "During handling of the above exception, another exception occurred:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/Users/yjx/.local/lib/python3.7/site-packages/IPython/core/interactiveshell.py\", line 2045, in showtraceback\n",
      "    stb = value._render_traceback_()\n",
      "AttributeError: 'Py4JJavaError' object has no attribute '_render_traceback_'\n",
      "\n",
      "During handling of the above exception, another exception occurred:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\", line 480, in send_command\n",
      "    raise Py4JNetworkError(\"Answer from Java side is empty\")\n",
      "py4j.protocol.Py4JNetworkError: Answer from Java side is empty\n",
      "\n",
      "During handling of the above exception, another exception occurred:\n",
      "\n",
      "Traceback (most recent call last):\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/java_gateway.py\", line 1038, in send_command\n",
      "    response = connection.send_command(command)\n",
      "  File \"/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\", line 504, in send_command\n",
      "    \"Error while sending or receiving\", e, proto.ERROR_ON_RECEIVE)\n",
      "py4j.protocol.Py4JNetworkError: Error while sending or receiving\n"
     ]
    },
    {
     "ename": "ConnectionRefusedError",
     "evalue": "[Errno 61] Connection refused",
     "output_type": "error",
     "traceback": [
      "\u001B[0;31m---------------------------------------------------------------------------\u001B[0m",
      "\u001B[0;31mPy4JJavaError\u001B[0m                             Traceback (most recent call last)",
      "    \u001B[0;31m[... skipping hidden 1 frame]\u001B[0m\n",
      "\u001B[0;32m<ipython-input-10-e73558e059f0>\u001B[0m in \u001B[0;36m<module>\u001B[0;34m\u001B[0m\n\u001B[1;32m      1\u001B[0m \u001B[0;31m#Fit cross validator to the 'train' dataset\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m----> 2\u001B[0;31m \u001B[0mmodel\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mcv\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mfit\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mtrain\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      3\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\u001B[0m in \u001B[0;36mfit\u001B[0;34m(self, dataset, params)\u001B[0m\n\u001B[1;32m    160\u001B[0m             \u001B[0;32melse\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 161\u001B[0;31m                 \u001B[0;32mreturn\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_fit\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mdataset\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    162\u001B[0m         \u001B[0;32melse\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/tuning.py\u001B[0m in \u001B[0;36m_fit\u001B[0;34m(self, dataset)\u001B[0m\n\u001B[1;32m    688\u001B[0m                 _parallelFitTasks(est, train, eva, validation, epm, collectSubModelsParam))\n\u001B[0;32m--> 689\u001B[0;31m             \u001B[0;32mfor\u001B[0m \u001B[0mj\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mmetric\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0msubModel\u001B[0m \u001B[0;32min\u001B[0m \u001B[0mpool\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mimap_unordered\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;32mlambda\u001B[0m \u001B[0mf\u001B[0m\u001B[0;34m:\u001B[0m \u001B[0mf\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mtasks\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    690\u001B[0m                 \u001B[0mmetrics\u001B[0m\u001B[0;34m[\u001B[0m\u001B[0mj\u001B[0m\u001B[0;34m]\u001B[0m \u001B[0;34m+=\u001B[0m \u001B[0;34m(\u001B[0m\u001B[0mmetric\u001B[0m \u001B[0;34m/\u001B[0m \u001B[0mnFolds\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/multiprocessing/pool.py\u001B[0m in \u001B[0;36mnext\u001B[0;34m(self, timeout)\u001B[0m\n\u001B[1;32m    747\u001B[0m             \u001B[0;32mreturn\u001B[0m \u001B[0mvalue\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 748\u001B[0;31m         \u001B[0;32mraise\u001B[0m \u001B[0mvalue\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0m\u001B[1;32m    749\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/multiprocessing/pool.py\u001B[0m in \u001B[0;36mworker\u001B[0;34m(inqueue, outqueue, initializer, initargs, maxtasks, wrap_exception)\u001B[0m\n\u001B[1;32m    120\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--> 121\u001B[0;31m             \u001B[0mresult\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0;34m(\u001B[0m\u001B[0;32mTrue\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[0mkwds\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    122\u001B[0m         \u001B[0;32mexcept\u001B[0m \u001B[0mException\u001B[0m \u001B[0;32mas\u001B[0m \u001B[0me\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/tuning.py\u001B[0m in \u001B[0;36m<lambda>\u001B[0;34m(f)\u001B[0m\n\u001B[1;32m    688\u001B[0m                 _parallelFitTasks(est, train, eva, validation, epm, collectSubModelsParam))\n\u001B[0;32m--> 689\u001B[0;31m             \u001B[0;32mfor\u001B[0m \u001B[0mj\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mmetric\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0msubModel\u001B[0m \u001B[0;32min\u001B[0m \u001B[0mpool\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mimap_unordered\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;32mlambda\u001B[0m \u001B[0mf\u001B[0m\u001B[0;34m:\u001B[0m \u001B[0mf\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mtasks\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    690\u001B[0m                 \u001B[0mmetrics\u001B[0m\u001B[0;34m[\u001B[0m\u001B[0mj\u001B[0m\u001B[0;34m]\u001B[0m \u001B[0;34m+=\u001B[0m \u001B[0;34m(\u001B[0m\u001B[0mmetric\u001B[0m \u001B[0;34m/\u001B[0m \u001B[0mnFolds\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/util.py\u001B[0m in \u001B[0;36mwrapped\u001B[0;34m(*args, **kwargs)\u001B[0m\n\u001B[1;32m    325\u001B[0m                 \u001B[0mSparkContext\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_active_spark_context\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_jsc\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0msc\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0msetLocalProperties\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mproperties\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 326\u001B[0;31m                 \u001B[0;32mreturn\u001B[0m \u001B[0mf\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[0mkwargs\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    327\u001B[0m             \u001B[0;32mfinally\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/tuning.py\u001B[0m in \u001B[0;36msingleTask\u001B[0;34m()\u001B[0m\n\u001B[1;32m     68\u001B[0m     \u001B[0;32mdef\u001B[0m \u001B[0msingleTask\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[0m\n\u001B[0;32m---> 69\u001B[0;31m         \u001B[0mindex\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mmodel\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mnext\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mmodelIter\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     70\u001B[0m         \u001B[0;31m# TODO: duplicate evaluator to take extra params from input\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\u001B[0m in \u001B[0;36m__next__\u001B[0;34m(self)\u001B[0m\n\u001B[1;32m     68\u001B[0m             \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mcounter\u001B[0m \u001B[0;34m+=\u001B[0m \u001B[0;36m1\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m---> 69\u001B[0;31m         \u001B[0;32mreturn\u001B[0m \u001B[0mindex\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mfitSingleModel\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mindex\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     70\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\u001B[0m in \u001B[0;36mfitSingleModel\u001B[0;34m(index)\u001B[0m\n\u001B[1;32m    125\u001B[0m         \u001B[0;32mdef\u001B[0m \u001B[0mfitSingleModel\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mindex\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--> 126\u001B[0;31m             \u001B[0;32mreturn\u001B[0m \u001B[0mestimator\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mfit\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mdataset\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mparamMaps\u001B[0m\u001B[0;34m[\u001B[0m\u001B[0mindex\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    127\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/base.py\u001B[0m in \u001B[0;36mfit\u001B[0;34m(self, dataset, params)\u001B[0m\n\u001B[1;32m    158\u001B[0m             \u001B[0;32mif\u001B[0m \u001B[0mparams\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 159\u001B[0;31m                 \u001B[0;32mreturn\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mcopy\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mparams\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_fit\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mdataset\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    160\u001B[0m             \u001B[0;32melse\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/wrapper.py\u001B[0m in \u001B[0;36m_fit\u001B[0;34m(self, dataset)\u001B[0m\n\u001B[1;32m    334\u001B[0m     \u001B[0;32mdef\u001B[0m \u001B[0m_fit\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mself\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mdataset\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--> 335\u001B[0;31m         \u001B[0mjava_model\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_fit_java\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mdataset\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    336\u001B[0m         \u001B[0mmodel\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_create_model\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mjava_model\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/ml/wrapper.py\u001B[0m in \u001B[0;36m_fit_java\u001B[0;34m(self, dataset)\u001B[0m\n\u001B[1;32m    331\u001B[0m         \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_transfer_params_to_java\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--> 332\u001B[0;31m         \u001B[0;32mreturn\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_java_obj\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mfit\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mdataset\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_jdf\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    333\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/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   1321\u001B[0m         return_value = get_return_value(\n\u001B[0;32m-> 1322\u001B[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001B[0m\u001B[1;32m   1323\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/pyspark/sql/utils.py\u001B[0m in \u001B[0;36mdeco\u001B[0;34m(*a, **kw)\u001B[0m\n\u001B[1;32m    110\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--> 111\u001B[0;31m             \u001B[0;32mreturn\u001B[0m \u001B[0mf\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;34m*\u001B[0m\u001B[0ma\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0;34m**\u001B[0m\u001B[0mkw\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    112\u001B[0m         \u001B[0;32mexcept\u001B[0m \u001B[0mpy4j\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mprotocol\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mPy4JJavaError\u001B[0m \u001B[0;32mas\u001B[0m \u001B[0me\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/protocol.py\u001B[0m in \u001B[0;36mget_return_value\u001B[0;34m(answer, gateway_client, target_id, name)\u001B[0m\n\u001B[1;32m    327\u001B[0m                     \u001B[0;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 328\u001B[0;31m                     format(target_id, \".\", name), value)\n\u001B[0m\u001B[1;32m    329\u001B[0m             \u001B[0;32melse\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;31m<class 'str'>\u001B[0m: (<class 'ConnectionRefusedError'>, ConnectionRefusedError(61, 'Connection refused'))",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001B[0;31mConnectionRefusedError\u001B[0m                    Traceback (most recent call last)",
      "    \u001B[0;31m[... skipping hidden 1 frame]\u001B[0m\n",
      "\u001B[0;32m~/.local/lib/python3.7/site-packages/IPython/core/interactiveshell.py\u001B[0m in \u001B[0;36mshowtraceback\u001B[0;34m(self, exc_tuple, filename, tb_offset, exception_only, running_compiled_code)\u001B[0m\n\u001B[1;32m   2048\u001B[0m                                             value, tb, tb_offset=tb_offset)\n\u001B[1;32m   2049\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m-> 2050\u001B[0;31m                     \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_showtraceback\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0metype\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mvalue\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mstb\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   2051\u001B[0m                     \u001B[0;32mif\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mcall_pdb\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m   2052\u001B[0m                         \u001B[0;31m# drop into debugger\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m~/.local/lib/python3.7/site-packages/ipykernel/zmqshell.py\u001B[0m in \u001B[0;36m_showtraceback\u001B[0;34m(self, etype, evalue, stb)\u001B[0m\n\u001B[1;32m    544\u001B[0m             \u001B[0;34mu'traceback'\u001B[0m \u001B[0;34m:\u001B[0m \u001B[0mstb\u001B[0m\u001B[0;34m,\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m    545\u001B[0m             \u001B[0;34mu'ename'\u001B[0m \u001B[0;34m:\u001B[0m \u001B[0municode_type\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0metype\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m__name__\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--> 546\u001B[0;31m             \u001B[0;34mu'evalue'\u001B[0m \u001B[0;34m:\u001B[0m \u001B[0mpy3compat\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0msafe_unicode\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mevalue\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    547\u001B[0m         }\n\u001B[1;32m    548\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m~/.local/lib/python3.7/site-packages/ipython_genutils/py3compat.py\u001B[0m in \u001B[0;36msafe_unicode\u001B[0;34m(e)\u001B[0m\n\u001B[1;32m     63\u001B[0m     \"\"\"\n\u001B[1;32m     64\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---> 65\u001B[0;31m         \u001B[0;32mreturn\u001B[0m \u001B[0municode_type\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0me\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     66\u001B[0m     \u001B[0;32mexcept\u001B[0m \u001B[0mUnicodeError\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m     67\u001B[0m         \u001B[0;32mpass\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/protocol.py\u001B[0m in \u001B[0;36m__str__\u001B[0;34m(self)\u001B[0m\n\u001B[1;32m    469\u001B[0m     \u001B[0;32mdef\u001B[0m \u001B[0m__str__\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mself\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    470\u001B[0m         \u001B[0mgateway_client\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mjava_exception\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_gateway_client\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 471\u001B[0;31m         \u001B[0manswer\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[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mexception_cmd\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    472\u001B[0m         \u001B[0mreturn_value\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mget_return_value\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0manswer\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mgateway_client\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0;32mNone\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0;32mNone\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m    473\u001B[0m         \u001B[0;31m# Note: technically this should return a bytestring 'str' rather than\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/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   1034\u001B[0m          \u001B[0;32mif\u001B[0m\u001B[0;31m \u001B[0m\u001B[0;31m`\u001B[0m\u001B[0mbinary\u001B[0m\u001B[0;31m`\u001B[0m \u001B[0;32mis\u001B[0m\u001B[0;31m \u001B[0m\u001B[0;31m`\u001B[0m\u001B[0;32mTrue\u001B[0m\u001B[0;31m`\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m   1035\u001B[0m         \"\"\"\n\u001B[0;32m-> 1036\u001B[0;31m         \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[0m\u001B[1;32m   1037\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[1;32m   1038\u001B[0m             \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[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\u001B[0m in \u001B[0;36m_get_connection\u001B[0;34m(self)\u001B[0m\n\u001B[1;32m    279\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m    280\u001B[0m         \u001B[0;32mif\u001B[0m \u001B[0mconnection\u001B[0m \u001B[0;32mis\u001B[0m \u001B[0;32mNone\u001B[0m \u001B[0;32mor\u001B[0m \u001B[0mconnection\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0msocket\u001B[0m \u001B[0;32mis\u001B[0m \u001B[0;32mNone\u001B[0m\u001B[0;34m:\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[0;32m--> 281\u001B[0;31m             \u001B[0mconnection\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0m_create_new_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[0m\u001B[1;32m    282\u001B[0m         \u001B[0;32mreturn\u001B[0m \u001B[0mconnection\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m    283\u001B[0m \u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\u001B[0m in \u001B[0;36m_create_new_connection\u001B[0;34m(self)\u001B[0m\n\u001B[1;32m    286\u001B[0m             \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mjava_parameters\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mpython_parameters\u001B[0m\u001B[0;34m,\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m    287\u001B[0m             self.gateway_property, self)\n\u001B[0;32m--> 288\u001B[0;31m         \u001B[0mconnection\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mconnect_to_java_server\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    289\u001B[0m         \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mset_thread_connection\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[1;32m    290\u001B[0m         \u001B[0;32mreturn\u001B[0m \u001B[0mconnection\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n",
      "\u001B[0;32m/opt/anaconda3/envs/cs5304/lib/python3.7/site-packages/py4j/clientserver.py\u001B[0m in \u001B[0;36mconnect_to_java_server\u001B[0;34m(self)\u001B[0m\n\u001B[1;32m    400\u001B[0m                 self.socket = self.ssl_context.wrap_socket(\n\u001B[1;32m    401\u001B[0m                     self.socket, server_hostname=self.java_address)\n\u001B[0;32m--> 402\u001B[0;31m             \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0msocket\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mconnect\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mjava_address\u001B[0m\u001B[0;34m,\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mjava_port\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    403\u001B[0m             \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mstream\u001B[0m \u001B[0;34m=\u001B[0m \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0msocket\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mmakefile\u001B[0m\u001B[0;34m(\u001B[0m\u001B[0;34m\"rb\"\u001B[0m\u001B[0;34m)\u001B[0m\u001B[0;34m\u001B[0m\u001B[0;34m\u001B[0m\u001B[0m\n\u001B[1;32m    404\u001B[0m             \u001B[0mself\u001B[0m\u001B[0;34m.\u001B[0m\u001B[0mis_connected\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;31mConnectionRefusedError\u001B[0m: [Errno 61] Connection refused"
     ]
    }
   ],
   "source": [
    "#Fit cross validator to the 'train' dataset\n",
    "model = cv.fit(train)\n",
    "\n",
    "#Extract best model from the cv model above\n",
    "best_model = model.bestModel"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "# Print best_model\n",
    "print(type(best_model))\n",
    "\n",
    "# Complete the code below to extract the ALS model parameters\n",
    "print(\"**Best Model**\")\n",
    "\n",
    "# # Print \"Rank\"\n",
    "print(\"  Rank:\", best_model._java_obj.parent().getRank())\n",
    "\n",
    "# Print \"MaxIter\"\n",
    "print(\"  MaxIter:\", best_model._java_obj.parent().getMaxIter())\n",
    "\n",
    "# Print \"RegParam\"\n",
    "print(\"  RegParam:\", best_model._java_obj.parent().getRegParam())"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "test_predictions = best_model.transform(test)\n",
    "RMSE = evaluator.evaluate(test_predictions)\n",
    "print(RMSE)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "# Generate n Recommendations for all users\n",
    "nrecommendations = best_model.recommendForAllUsers(10)\n",
    "nrecommendations.limit(10).show()"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "outputs": [],
   "source": [
    "nrecommendations = nrecommendations\\\n",
    "    .withColumn(\"rec_exp\", explode(\"recommendations\"))\\\n",
    "    .select('userId', col(\"rec_exp.movieId\"), col(\"rec_exp.rating\"))\n",
    "\n",
    "nrecommendations.limit(10).show()"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}