SparkML - 01 Regression.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Data Source: https://archive.ics.uci.edu/ml/datasets/Combined+Cycle+Power+Plant\n",
    "\n",
    "Features consist of hourly average ambient variables\n",
    "\n",
    "Temperature (T) in the range 1.81°C and 37.11°C,\n",
    "Ambient Pressure (AP) in the range 992.89-1033.30 milibar,\n",
    "Relative Humidity (RH) in the range 25.56% to 100.16%\n",
    "Exhaust Vacuum (V) in teh range 25.36-81.56 cm Hg\n",
    "Net hourly electrical energy output (EP) 420.26-495.76 MW\n",
    "The averages are taken from various sensors located around the plant that record the ambient variables every second. The variables are given without normalization.\n",
    "\n",
    "Dataset Information:\n",
    "\n",
    "The dataset contains 9568 data points collected from a Combined Cycle Power Plant over 6 years (2006-2011), when the power plant was set to work with full load. Features consist of hourly average ambient variables Temperature (T), Ambient Pressure (AP), Relative Humidity (RH) and Exhaust Vacuum (V) to predict the net hourly electrical energy output (EP) of the plant. \n",
    "A combined cycle power plant (CCPP) is composed of gas turbines (GT), steam turbines (ST) and heat recovery steam generators. In a CCPP, the electricity is generated by gas and steam turbines, which are combined in one cycle, and is transferred from one turbine to another. While the Vacuum is colected from and has effect on the Steam Turbine, he other three of the ambient variables effect the GT performance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 1584948\r\n",
      "drwxr-xr-x 2 training training      4096 Jun  5  2009 cifar-10-batches-py\r\n",
      "-rw-r--r-- 1 training training 150828752 Nov  5  2016 creditcard-fraud.csv\r\n",
      "-rw-r--r-- 1 training training    133638 Jun  6  2017 credit-default.csv\r\n",
      "-rw-r--r-- 1 training training   1215506 Jun 18  2017 CleanCreditScoring.csv\r\n",
      "-rw-r--r-- 1 training training     57459 Sep 12  2017 istanbul-stock.csv\r\n",
      "-rw-r--r-- 1 training training      2249 Nov  8 07:15 story.txt\r\n",
      "-rw-r--r-- 1 training training 127744095 Nov  8 07:16 stocks.csv\r\n",
      "-rw-r--r-- 1 training training       160 Nov  8 07:16 poem.txt\r\n",
      "-rw-r--r-- 1 training training  69055807 Nov  8 07:16 imdb-comments.json\r\n",
      "-rw-r--r-- 1 training training 526915613 Nov  8 07:18 tweets.json\r\n",
      "drwxr-xr-x 2 training training      4096 Nov  8 07:18 MNIST\r\n",
      "drwxr-xr-x 2 training training      4096 Nov  8 07:18 weblogs\r\n",
      "drwxr-xr-x 2 training training      4096 Dec  1 20:05 kaggle-house-price\r\n",
      "-rw-r--r-- 1 training training   2772143 Dec  2 13:46 diamonds.csv\r\n",
      "-rw-rw-r-- 1 training training      5107 Feb 19 00:12 iris.csv\r\n",
      "-rw-r--r-- 1 training training   1543414 Mar 14 20:44 Combined_Cycle_Power_Plant.csv\r\n",
      "-rw-r--r-- 1 training training 742579829 Mar 16 10:34 kddcup.data\r\n",
      "-rw-rw-r-- 1 training training     19130 Mar 16 12:38 gdp.csv\r\n",
      "drwxr-xr-x 2 training training      4096 Apr  2 13:12 movielens\r\n",
      "-rw-rw-r-- 1 training training     54292 Apr  2 13:12 insurance.csv\r\n"
     ]
    }
   ],
   "source": [
    "!ls -ltr /data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://10.0.2.15:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>PySparkShell</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fd4119623c8>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.format(\"csv\").option(\"header\",\"true\")\\\n",
    ".option(\"inferSchema\",\"true\").load(\"/data/Combined_Cycle_Power_Plant.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+\n",
      "|   AT|    V|     AP|   RH|    EP|\n",
      "+-----+-----+-------+-----+------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|\n",
      "|25.18|62.96|1020.04|59.08|444.37|\n",
      "| 5.11| 39.4|1012.16|92.14|488.56|\n",
      "|20.86|57.32|1010.24|76.64|446.48|\n",
      "|10.82| 37.5|1009.23|96.62| 473.9|\n",
      "|26.27|59.44|1012.23|58.77|443.67|\n",
      "|15.89|43.96|1014.02|75.24|467.35|\n",
      "| 9.48|44.71|1019.12|66.43|478.42|\n",
      "|14.64| 45.0|1021.78|41.25|475.98|\n",
      "|11.74|43.56|1015.14|70.72| 477.5|\n",
      "|17.99|43.72|1008.64|75.04|453.02|\n",
      "|20.14|46.93|1014.66|64.22|453.99|\n",
      "|24.34| 73.5|1011.31|84.15|440.29|\n",
      "|25.71|58.59|1012.77|61.83|451.28|\n",
      "|26.19|69.34|1009.48|87.59|433.99|\n",
      "|21.42|43.79|1015.76|43.08|462.19|\n",
      "|18.21| 45.0|1022.86|48.84|467.54|\n",
      "|11.04|41.74| 1022.6|77.51| 477.2|\n",
      "|14.45|52.75|1023.97|63.59|459.85|\n",
      "|13.97|38.47|1015.15|55.28| 464.3|\n",
      "+-----+-----+-------+-----+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[AT: double, V: double, AP: double, RH: double, EP: double]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Convert Spark Dataframe to Pandas Dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>AT</th>\n",
       "      <th>V</th>\n",
       "      <th>AP</th>\n",
       "      <th>RH</th>\n",
       "      <th>EP</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>14.96</td>\n",
       "      <td>41.76</td>\n",
       "      <td>1024.07</td>\n",
       "      <td>73.17</td>\n",
       "      <td>463.26</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>25.18</td>\n",
       "      <td>62.96</td>\n",
       "      <td>1020.04</td>\n",
       "      <td>59.08</td>\n",
       "      <td>444.37</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>5.11</td>\n",
       "      <td>39.40</td>\n",
       "      <td>1012.16</td>\n",
       "      <td>92.14</td>\n",
       "      <td>488.56</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>20.86</td>\n",
       "      <td>57.32</td>\n",
       "      <td>1010.24</td>\n",
       "      <td>76.64</td>\n",
       "      <td>446.48</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>10.82</td>\n",
       "      <td>37.50</td>\n",
       "      <td>1009.23</td>\n",
       "      <td>96.62</td>\n",
       "      <td>473.90</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "      AT      V       AP     RH      EP\n",
       "0  14.96  41.76  1024.07  73.17  463.26\n",
       "1  25.18  62.96  1020.04  59.08  444.37\n",
       "2   5.11  39.40  1012.16  92.14  488.56\n",
       "3  20.86  57.32  1010.24  76.64  446.48\n",
       "4  10.82  37.50  1009.23  96.62  473.90"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.limit(10).toPandas().head() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Verctorize the features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+---------------------------+\n",
      "|AT   |V    |AP     |RH   |EP    |features                   |\n",
      "+-----+-----+-------+-----+------+---------------------------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|[14.96,41.76,1024.07,73.17]|\n",
      "|25.18|62.96|1020.04|59.08|444.37|[25.18,62.96,1020.04,59.08]|\n",
      "|5.11 |39.4 |1012.16|92.14|488.56|[5.11,39.4,1012.16,92.14]  |\n",
      "|20.86|57.32|1010.24|76.64|446.48|[20.86,57.32,1010.24,76.64]|\n",
      "|10.82|37.5 |1009.23|96.62|473.9 |[10.82,37.5,1009.23,96.62] |\n",
      "|26.27|59.44|1012.23|58.77|443.67|[26.27,59.44,1012.23,58.77]|\n",
      "|15.89|43.96|1014.02|75.24|467.35|[15.89,43.96,1014.02,75.24]|\n",
      "|9.48 |44.71|1019.12|66.43|478.42|[9.48,44.71,1019.12,66.43] |\n",
      "|14.64|45.0 |1021.78|41.25|475.98|[14.64,45.0,1021.78,41.25] |\n",
      "|11.74|43.56|1015.14|70.72|477.5 |[11.74,43.56,1015.14,70.72]|\n",
      "+-----+-----+-------+-----+------+---------------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "vectorizer = VectorAssembler()\n",
    "vectorizer.setInputCols([\"AT\", \"V\", \"AP\", \"RH\"])\n",
    "vectorizer.setOutputCol(\"features\")\n",
    "\n",
    "df_vect = vectorizer.transform(df)\n",
    "df_vect.show(10, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "inputCols: input column names. (current: ['AT', 'V', 'AP', 'RH'])\n",
      "outputCol: output column name. (default: VectorAssembler_4bd6aa0aa3917dd77da5__output, current: features)\n"
     ]
    }
   ],
   "source": [
    "print(vectorizer.explainParams())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fit Linear Regression Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.regression import LinearRegression"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "aggregationDepth: suggested depth for treeAggregate (>= 2). (default: 2)\n",
      "elasticNetParam: the ElasticNet mixing parameter, in range [0, 1]. For alpha = 0, the penalty is an L2 penalty. For alpha = 1, it is an L1 penalty. (default: 0.0)\n",
      "epsilon: The shape parameter to control the amount of robustness. Must be > 1.0. Only valid when loss is huber (default: 1.35)\n",
      "featuresCol: features column name. (default: features)\n",
      "fitIntercept: whether to fit an intercept term. (default: True)\n",
      "labelCol: label column name. (default: label)\n",
      "loss: The loss function to be optimized. Supported options: squaredError, huber. (default: squaredError)\n",
      "maxIter: max number of iterations (>= 0). (default: 100)\n",
      "predictionCol: prediction column name. (default: prediction)\n",
      "regParam: regularization parameter (>= 0). (default: 0.0)\n",
      "solver: The solver algorithm for optimization. Supported options: auto, normal, l-bfgs. (default: auto)\n",
      "standardization: whether to standardize the training features before fitting the model. (default: True)\n",
      "tol: the convergence tolerance for iterative algorithms (>= 0). (default: 1e-06)\n",
      "weightCol: weight column name. If this is not set or empty, we treat all instance weights as 1.0. (undefined)\n"
     ]
    }
   ],
   "source": [
    "lr = LinearRegression()\n",
    "print(lr.explainParams())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "lr.setLabelCol(\"EP\")\n",
    "lr.setFeaturesCol(\"features\")\n",
    "model = lr.fit(df_vect)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.ml.regression.LinearRegressionModel"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### View model summary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "R2: 0.9286835997167648\n",
      "Intercept:  454.5637357984046 Coefficients [-1.9773160434618613,-0.23402845649906473,0.06212776009866186,-0.15801655439825457]\n"
     ]
    }
   ],
   "source": [
    "print(\"R2:\", model.summary.r2)\n",
    "print(\"Intercept: \", model.intercept, \"Coefficients\", model.coefficients)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Predict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "|   AT|    V|     AP|   RH|    EP|            features|        prediction|\n",
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|[14.96,41.76,1024...| 467.2711634437306|\n",
      "|25.18|62.96|1020.04|59.08|444.37|[25.18,62.96,1020...|444.07766858004396|\n",
      "| 5.11| 39.4|1012.16|92.14|488.56|[5.11,39.4,1012.1...|483.56251796945776|\n",
      "|20.86|57.32|1010.24|76.64|446.48|[20.86,57.32,1010...| 450.5559716382537|\n",
      "|10.82| 37.5|1009.23|96.62| 473.9|[10.82,37.5,1009....| 471.8267489278455|\n",
      "|26.27|59.44|1012.23|58.77|443.67|[26.27,59.44,1012...| 442.3099415850402|\n",
      "|15.89|43.96|1014.02|75.24|467.35|[15.89,43.96,1014...|463.96591866241715|\n",
      "| 9.48|44.71|1019.12|66.43|478.42|[9.48,44.71,1019....| 478.1739705793852|\n",
      "|14.64| 45.0|1021.78|41.25|475.98|[14.64,45.0,1021....|472.04726822434776|\n",
      "|11.74|43.56|1015.14|70.72| 477.5|[11.74,43.56,1015...| 473.0492095425741|\n",
      "|17.99|43.72|1008.64|75.04|453.02|[17.99,43.72,1008...| 459.5670777622559|\n",
      "|20.14|46.93|1014.66|64.22|453.99|[20.14,46.93,1014...|456.64836515783395|\n",
      "|24.34| 73.5|1011.31|84.15|440.29|[24.34,73.5,1011....| 438.7681037606262|\n",
      "|25.71|58.59|1012.77|61.83|451.28|[25.71,58.59,1012...| 443.1661810913976|\n",
      "|26.19|69.34|1009.48|87.59|433.99|[26.19,69.34,1009...| 435.4263567111474|\n",
      "|21.42|43.79|1015.76|43.08|462.19|[21.42,43.79,1015...| 458.2610604716974|\n",
      "|18.21| 45.0|1022.86|48.84|467.54|[18.21,45.0,1022....|463.85600228221267|\n",
      "|11.04|41.74| 1022.6|77.51| 477.2|[11.04,41.74,1022...| 474.2498032497976|\n",
      "|14.45|52.75|1023.97|63.59|459.85|[14.45,52.75,1023...| 467.2152077040968|\n",
      "|13.97|38.47|1015.15|55.28| 464.3|[13.97,38.47,1015...|472.27139648674444|\n",
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_pred = model.transform(df_vect)\n",
    "df_pred.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "labelCol: label column name. (default: label)\n",
      "metricName: metric name in evaluation - one of:\n",
      "                       rmse - root mean squared error (default)\n",
      "                       mse - mean squared error\n",
      "                       r2 - r^2 metric\n",
      "                       mae - mean absolute error. (default: rmse)\n",
      "predictionCol: prediction column name. (default: prediction)\n"
     ]
    }
   ],
   "source": [
    "evaluator = RegressionEvaluator()\n",
    "print(evaluator.explainParams())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.557525128298466"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluator = RegressionEvaluator(labelCol = \"EP\", \n",
    "                                predictionCol = \"prediction\", \n",
    "                                metricName = \"rmse\")\n",
    "evaluator.evaluate(df_pred)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build a pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.pipeline import Pipeline, PipelineModel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "stages: a list of pipeline stages (undefined)\n"
     ]
    }
   ],
   "source": [
    "pipeline = Pipeline()\n",
    "print(pipeline.explainParams())\n",
    "pipeline.setStages([vectorizer, lr])\n",
    "pipelineModel = pipeline.fit(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[VectorAssembler_4bd6aa0aa3917dd77da5, LinearRegression_41e591e8b658af913914]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pipeline.getStages()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([-1.9773, -0.234, 0.0621, -0.158])"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lr_model = pipelineModel.stages[1]\n",
    "lr_model .coefficients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "|   AT|    V|     AP|   RH|    EP|            features|        prediction|\n",
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|[14.96,41.76,1024...| 467.2711634437306|\n",
      "|25.18|62.96|1020.04|59.08|444.37|[25.18,62.96,1020...|444.07766858004396|\n",
      "| 5.11| 39.4|1012.16|92.14|488.56|[5.11,39.4,1012.1...|483.56251796945776|\n",
      "|20.86|57.32|1010.24|76.64|446.48|[20.86,57.32,1010...| 450.5559716382537|\n",
      "|10.82| 37.5|1009.23|96.62| 473.9|[10.82,37.5,1009....| 471.8267489278455|\n",
      "|26.27|59.44|1012.23|58.77|443.67|[26.27,59.44,1012...| 442.3099415850402|\n",
      "|15.89|43.96|1014.02|75.24|467.35|[15.89,43.96,1014...|463.96591866241715|\n",
      "| 9.48|44.71|1019.12|66.43|478.42|[9.48,44.71,1019....| 478.1739705793852|\n",
      "|14.64| 45.0|1021.78|41.25|475.98|[14.64,45.0,1021....|472.04726822434776|\n",
      "|11.74|43.56|1015.14|70.72| 477.5|[11.74,43.56,1015...| 473.0492095425741|\n",
      "|17.99|43.72|1008.64|75.04|453.02|[17.99,43.72,1008...| 459.5670777622559|\n",
      "|20.14|46.93|1014.66|64.22|453.99|[20.14,46.93,1014...|456.64836515783395|\n",
      "|24.34| 73.5|1011.31|84.15|440.29|[24.34,73.5,1011....| 438.7681037606262|\n",
      "|25.71|58.59|1012.77|61.83|451.28|[25.71,58.59,1012...| 443.1661810913976|\n",
      "|26.19|69.34|1009.48|87.59|433.99|[26.19,69.34,1009...| 435.4263567111474|\n",
      "|21.42|43.79|1015.76|43.08|462.19|[21.42,43.79,1015...| 458.2610604716974|\n",
      "|18.21| 45.0|1022.86|48.84|467.54|[18.21,45.0,1022....|463.85600228221267|\n",
      "|11.04|41.74| 1022.6|77.51| 477.2|[11.04,41.74,1022...| 474.2498032497976|\n",
      "|14.45|52.75|1023.97|63.59|459.85|[14.45,52.75,1023...| 467.2152077040968|\n",
      "|13.97|38.47|1015.15|55.28| 464.3|[13.97,38.47,1015...|472.27139648674444|\n",
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pipelineModel.transform(df).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.557525128298466"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluator.evaluate(pipelineModel.transform(df))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save the pipeline to disk to persist the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipelineModel.save(\"/tmp/lr-pipeline\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[01;34m/tmp/lr-pipeline\u001b[00m\r\n",
      "├── \u001b[01;34mmetadata\u001b[00m\r\n",
      "│   ├── part-00000\r\n",
      "│   └── _SUCCESS\r\n",
      "└── \u001b[01;34mstages\u001b[00m\r\n",
      "    ├── \u001b[01;34m0_VectorAssembler_4bd6aa0aa3917dd77da5\u001b[00m\r\n",
      "    │   └── \u001b[01;34mmetadata\u001b[00m\r\n",
      "    │       ├── part-00000\r\n",
      "    │       └── _SUCCESS\r\n",
      "    └── \u001b[01;34m1_LinearRegression_41e591e8b658af913914\u001b[00m\r\n",
      "        ├── \u001b[01;34mdata\u001b[00m\r\n",
      "        │   ├── part-00000-49984b04-af0d-4925-b788-e989c20d1c66-c000.snappy.parquet\r\n",
      "        │   └── _SUCCESS\r\n",
      "        └── \u001b[01;34mmetadata\u001b[00m\r\n",
      "            ├── part-00000\r\n",
      "            └── _SUCCESS\r\n",
      "\r\n",
      "7 directories, 8 files\r\n"
     ]
    }
   ],
   "source": [
    "!tree /tmp/lr-pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the persisted model from the disk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DenseVector([-1.9773, -0.234, 0.0621, -0.158])"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "saved_model = PipelineModel.load(\"/tmp/lr-pipeline\")\n",
    "saved_model.stages[1].coefficients"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "|   AT|    V|     AP|   RH|    EP|            features|        prediction|\n",
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "|14.96|41.76|1024.07|73.17|463.26|[14.96,41.76,1024...| 467.2711634437306|\n",
      "|25.18|62.96|1020.04|59.08|444.37|[25.18,62.96,1020...|444.07766858004396|\n",
      "| 5.11| 39.4|1012.16|92.14|488.56|[5.11,39.4,1012.1...|483.56251796945776|\n",
      "|20.86|57.32|1010.24|76.64|446.48|[20.86,57.32,1010...| 450.5559716382537|\n",
      "|10.82| 37.5|1009.23|96.62| 473.9|[10.82,37.5,1009....| 471.8267489278455|\n",
      "|26.27|59.44|1012.23|58.77|443.67|[26.27,59.44,1012...| 442.3099415850402|\n",
      "|15.89|43.96|1014.02|75.24|467.35|[15.89,43.96,1014...|463.96591866241715|\n",
      "| 9.48|44.71|1019.12|66.43|478.42|[9.48,44.71,1019....| 478.1739705793852|\n",
      "|14.64| 45.0|1021.78|41.25|475.98|[14.64,45.0,1021....|472.04726822434776|\n",
      "|11.74|43.56|1015.14|70.72| 477.5|[11.74,43.56,1015...| 473.0492095425741|\n",
      "|17.99|43.72|1008.64|75.04|453.02|[17.99,43.72,1008...| 459.5670777622559|\n",
      "|20.14|46.93|1014.66|64.22|453.99|[20.14,46.93,1014...|456.64836515783395|\n",
      "|24.34| 73.5|1011.31|84.15|440.29|[24.34,73.5,1011....| 438.7681037606262|\n",
      "|25.71|58.59|1012.77|61.83|451.28|[25.71,58.59,1012...| 443.1661810913976|\n",
      "|26.19|69.34|1009.48|87.59|433.99|[26.19,69.34,1009...| 435.4263567111474|\n",
      "|21.42|43.79|1015.76|43.08|462.19|[21.42,43.79,1015...| 458.2610604716974|\n",
      "|18.21| 45.0|1022.86|48.84|467.54|[18.21,45.0,1022....|463.85600228221267|\n",
      "|11.04|41.74| 1022.6|77.51| 477.2|[11.04,41.74,1022...| 474.2498032497976|\n",
      "|14.45|52.75|1023.97|63.59|459.85|[14.45,52.75,1023...| 467.2152077040968|\n",
      "|13.97|38.47|1015.15|55.28| 464.3|[13.97,38.47,1015...|472.27139648674444|\n",
      "+-----+-----+-------+-----+------+--------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "saved_model.transform(df).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_train, df_test = df.randomSplit(weights=[0.7, 0.3], seed = 200)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.563138184940591"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pipelineModel = pipeline.fit(df_train)\n",
    "evaluator.evaluate(pipelineModel.transform(df_test))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tune the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "paramGrid = ParamGridBuilder()\\\n",
    "    .addGrid(lr.regParam, [0.1, 0.01]) \\\n",
    "    .addGrid(lr.fitIntercept, [False, True])\\\n",
    "    .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])\\\n",
    "    .build()\n",
    "\n",
    "# In this case the estimator is simply the linear regression.\n",
    "# A TrainValidationSplit requires an Estimator, a set of Estimator ParamMaps, and an Evaluator.\n",
    "tvs = TrainValidationSplit(estimator=lr,\n",
    "                           estimatorParamMaps=paramGrid,\n",
    "                           evaluator=evaluator,\n",
    "                           trainRatio=0.8)\n",
    "\n",
    "tuned_model = tvs.fit(vectorizer.transform(df_train))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(LinearRegression_41e591e8b658af913914,\n",
       " [5.039347932903683,\n",
       "  5.636217630762049,\n",
       "  5.107247882789134,\n",
       "  4.547646522666319,\n",
       "  4.545696090598205,\n",
       "  4.545352062329218,\n",
       "  5.038546732463139,\n",
       "  5.49137539732243,\n",
       "  5.628475999107409,\n",
       "  4.545755899303294,\n",
       "  4.5455883701354445,\n",
       "  4.545438239526134])"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "tuned_model.bestModel, tuned_model.validationMetrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+-------+-----+------+--------------------+------------------+\n",
      "|  AT|    V|     AP|   RH|    EP|            features|        prediction|\n",
      "+----+-----+-------+-----+------+--------------------+------------------+\n",
      "|1.81|39.42|1026.92|76.97|490.55|[1.81,39.42,1026....| 492.8349077427656|\n",
      "|1.81|39.42|1026.92|76.97|490.55|[1.81,39.42,1026....| 492.8349077427656|\n",
      "|1.81|39.42|1026.92|76.97|490.55|[1.81,39.42,1026....| 492.8349077427656|\n",
      "|1.81|39.42|1026.92|76.97|490.55|[1.81,39.42,1026....| 492.8349077427656|\n",
      "|2.34|39.42|1028.47|69.68|490.34|[2.34,39.42,1028....|492.94214880761615|\n",
      "|2.58|39.42|1028.68|69.03|488.69|[2.58,39.42,1028....|492.58206703142434|\n",
      "|2.58|39.42|1028.68|69.03|488.69|[2.58,39.42,1028....|492.58206703142434|\n",
      "| 2.8|39.64|1011.01|82.96|482.66|[2.8,39.64,1011.0...| 489.0569119382538|\n",
      "| 2.8|39.64|1011.01|82.96|482.66|[2.8,39.64,1011.0...| 489.0569119382538|\n",
      "|3.21|38.44| 1016.9|86.34|491.35|[3.21,38.44,1016....| 488.4152185498315|\n",
      "|3.21|38.44|1017.11|84.86|492.93|[3.21,38.44,1017....| 488.6394965835994|\n",
      "|3.26|41.31| 996.32|100.0|489.38|[3.26,41.31,996.3...| 484.4525561174197|\n",
      "|3.26|41.31| 996.32|100.0|489.38|[3.26,41.31,996.3...| 484.4525561174197|\n",
      "|3.31|39.42|1024.05|84.31|487.19|[3.31,39.42,1024....| 488.7036922378125|\n",
      "|3.31|39.42|1024.05|84.31|487.19|[3.31,39.42,1024....| 488.7036922378125|\n",
      "|3.38|39.64| 1011.0|81.22|488.92|[3.38,39.64,1011....| 488.1800909364742|\n",
      "|3.38|41.31| 998.79|97.76|489.11|[3.38,41.31,998.7...| 484.6874147567333|\n",
      "|3.38|41.31| 998.79|97.76|489.11|[3.38,41.31,998.7...| 484.6874147567333|\n",
      "| 3.4|39.64| 1011.1|83.43|459.86|[3.4,39.64,1011.1...| 487.8310275292321|\n",
      "| 3.4|39.64| 1011.1|83.43|459.86|[3.4,39.64,1011.1...| 487.8310275292321|\n",
      "+----+-----+-------+-----+------+--------------------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_test_pred = tuned_model.transform(vectorizer.transform(df_test))\n",
    "df_test_pred.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4.5683171800255895"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluator.evaluate(df_test_pred)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "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.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}