ddl_nyiso_hdi.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Distributed Training of RNN Models on an HDInsight Spark Cluster\n",
    "\n",
    "This notebook demonstrates energy consumption prediction using Distributed Keras on an HDInsight Spark cluster. It uses distkeras package to train a GRU model and an LSTM model in a distrbuted manner on the Spark cluster. The data used is the NYISO data which describes the hourly energy consumption of New York City.\n",
    "\n",
    "This example requires the following packages\n",
    "- distkeras\n",
    "- keras\n",
    "- tensorflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>155</td><td>application_1526382914088_0018</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://hn0-walmar.y11pdt2eof4e5evgwhstr0mncg.gx.internal.cloudapp.net:8088/proxy/application_1526382914088_0018/\">Link</a></td><td><a target=\"_blank\" href=\"http://10.0.0.6:30060/node/containerlogs/container_e14_1526382914088_0018_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n",
      "Using TensorFlow backend."
     ]
    }
   ],
   "source": [
    "import os\n",
    "import time\n",
    "import pyspark\n",
    "\n",
    "import datetime as dt\n",
    "import numpy as np\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "from distkeras.evaluators import *\n",
    "from distkeras.predictors import *\n",
    "from distkeras.trainers import *\n",
    "from distkeras.transformers import *\n",
    "from distkeras.utils import *\n",
    "\n",
    "from keras.models import Sequential\n",
    "from keras.layers import Dense\n",
    "from keras.layers import LSTM\n",
    "from keras.layers import GRU\n",
    "\n",
    "from pyspark.ml.feature import VectorAssembler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of desired executors: 2\n",
      "Number of desired processes per executor: 2\n",
      "Total number of workers: 4"
     ]
    }
   ],
   "source": [
    "## Setup the pyspark environment\n",
    "\n",
    "# First, setup the Spark variables. You can modify them according to your need, e.g. increasing num_executors to reduce training time.\n",
    "application_name = 'Distributed Keras NYISO'\n",
    "master = 'yarn-client'\n",
    "num_processes = 2 \n",
    "num_executors = 2 #4, 3, 2, 1\n",
    "\n",
    "# This variable is derived from the number of cores and executors, and will be used to assign the number of model trainers.\n",
    "num_workers = num_executors * num_processes\n",
    "\n",
    "print('Number of desired executors: ' + str(num_executors))\n",
    "print('Number of desired processes per executor: ' + str(num_processes))\n",
    "print('Total number of workers: ' + str(num_workers))\n",
    "\n",
    "# Use the DataBricks CSV reader, which has some nice functionality regarding invalid values.\n",
    "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages com.databricks:spark-csv_2.10:1.4.0 pyspark-shell'\n",
    "\n",
    "# Modify the Spark configuration\n",
    "conf = pyspark.SparkConf()\n",
    "conf.set('spark.app.name', application_name)\n",
    "conf.set('spark.master', master)\n",
    "conf.set('spark.executor.cores', num_processes)\n",
    "conf.set('spark.executor.instances', num_executors)\n",
    "conf.set('spark.executor.memory', '2g')\n",
    "conf.set('spark.locality.wait', '0')\n",
    "conf.set('spark.serializer', 'org.apache.spark.serializer.KryoSerializer');\n",
    "conf.set('spark.local.dir', '/tmp/' + get_os_username() + '/dist-keras');\n",
    "\n",
    "# Create the Spark context\n",
    "sc.stop()\n",
    "sc = pyspark.SparkContext(conf=conf)\n",
    "sqlc = pyspark.sql.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## Define variables\n",
    "LEN_SEQ_IN = 24\n",
    "LEN_SEQ_OUT = 1\n",
    "LEN_EXTRA_IN = LEN_SEQ_OUT\n",
    "LEN_TEST_DATA = 120\n",
    "N_UNITS = 128\n",
    "\n",
    "BATCH_SIZE = 32\n",
    "COM_WINDOW = 5\n",
    "N_EPOCHS = 20\n",
    "\n",
    "input_container = 'nyiso'\n",
    "storage_account = 'publicdat'\n",
    "storage_key = 'GNDru9kic+4jXle1roq4klFzwQ3Jy9CvsvMJl8o5b8W/DoX6v14PE8aPuX48V3r9yc3tuKP3NKCSUHqFobib4A=='\n",
    "data_path = 'wasb://{}@{}.blob.core.windows.net/NYISO_data_1region_small.csv'.format(input_container, storage_account)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load and Prepare Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "## Load data\n",
    "\n",
    "# Attach the blob storage to the spark cluster  \n",
    "def attach_storage_container(spark, account, key):\n",
    "    config = spark._sc._jsc.hadoopConfiguration()\n",
    "    setting = 'fs.azure.account.key.' + account + '.blob.core.windows.net'\n",
    "    if not config.get(setting):\n",
    "        config.set(setting, key)\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()\n",
    "spark.sparkContext.setLogLevel('ERROR')\n",
    "attach_storage_container(spark, storage_account, storage_key)\n",
    "time.sleep(4) \n",
    "\n",
    "# Load the input dataset\n",
    "raw_df = sqlc.read.format('com.databricks.spark.csv') \\\n",
    "                  .options(header='true', inferSchema='true') \\\n",
    "                  .load(data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|\n",
      "+--------------------+------+-----------+-----------+\n",
      "|2016-01-02 00:00:...|N.Y.C.|   4836.967|      34.44|\n",
      "|2016-01-02 01:00:...|N.Y.C.|   4613.692|      33.79|\n",
      "|2016-01-02 02:00:...|N.Y.C.|   4448.508|      33.13|\n",
      "+--------------------+------+-----------+-----------+\n",
      "only showing top 3 rows"
     ]
    }
   ],
   "source": [
    "# Convert data type and select columns\n",
    "func = F.udf (lambda x: dt.datetime.strptime(x[:19], '%m/%d/%Y %H:%M:%S'), TimestampType())\n",
    "raw_df = raw_df.withColumn('TimeStamp', func(F.col('TimeStamp')))\n",
    "df = raw_df.select(['TimeStamp', 'Name', 'HourAvgLoad', 'temperature'])\n",
    "df.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+\n",
      "|2016-01-02 00:00:...|N.Y.C.|   4836.967|      34.44| 0.3608378292605522|0.13131163940862922|\n",
      "|2016-01-02 01:00:...|N.Y.C.|   4613.692|      33.79| 0.3539617052787475|0.09975432572999143|\n",
      "|2016-01-02 02:00:...|N.Y.C.|   4448.508|      33.13|0.34697979477414576|0.07640749647148892|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+\n",
      "only showing top 3 rows"
     ]
    }
   ],
   "source": [
    "# Normalize the data\n",
    "orig_min = df.select(F.min('temperature')).collect()[0][0]\n",
    "orig_max = df.select(F.max('temperature')).collect()[0][0]\n",
    "transformer = MinMaxTransformer(n_min=0.0, n_max=1.0, \\\n",
    "                                o_min=orig_min, o_max=orig_max, \\\n",
    "                                input_col='temperature', \\\n",
    "                                output_col='NormTemp', \\\n",
    "                                is_vector=False)\n",
    "df = transformer.transform(df)\n",
    "orig_min = df.select(F.min('HourAvgLoad')).collect()[0][0]\n",
    "orig_max = df.select(F.max('HourAvgLoad')).collect()[0][0]\n",
    "transformer = MinMaxTransformer(n_min=0.0, n_max=1.0, \\\n",
    "                                o_min=orig_min, o_max=orig_max, \\\n",
    "                                input_col='HourAvgLoad', \\\n",
    "                                output_col='NormLoad', \\\n",
    "                                is_vector=False)\n",
    "df = transformer.transform(df)\n",
    "df.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create input features and output targets\n",
    "wSpec = Window.partitionBy('Name').orderBy('TimeStamp')\n",
    "for n_lag in range(LEN_SEQ_IN, 0, -1):\n",
    "    df = df.withColumn('NormLoad_lag'+str(n_lag), F.lag(F.col('NormLoad'), count = n_lag).over(wSpec))\n",
    "for n_lag in range(1, LEN_SEQ_OUT+1):\n",
    "    df = df.withColumn('NormLoad_next'+str(n_lag), F.lead(F.col('NormLoad'), count = n_lag).over(wSpec))\n",
    "    df = df.withColumn('OrigNormLoad_next'+str(n_lag), F.lead(F.col('HourAvgLoad'), count = n_lag).over(wSpec))\n",
    "    df = df.withColumn('NormTemp_next'+str(n_lag), F.lead(F.col('NormTemp'), count = n_lag).over(wSpec))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|      NormLoad_lag21|     NormLoad_lag20|     NormLoad_lag19|     NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|     NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|     NormLoad_lag9|     NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|      NormLoad_lag5|     NormLoad_lag4|      NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+\n",
      "|2016-01-03 00:00:...|N.Y.C.|   4919.133|      34.61|0.36263620014810116|0.14292484391302507|0.13131163940862922|0.09975432572999143|0.07640749647148892|0.061212496229800406|0.05648359867718655|0.06364351535541934|0.08511167564777489|0.10788707407343567|0.14099910928589954|0.17776120664482353|0.2031515053520584|0.21917107335995967|0.22299427494995927|0.21947367870577084|0.21895072691712025|0.2188965943400788|0.2383776791738833|0.2769630408770214|0.28345174186760513|0.27544365392350956|0.2617397729710814|0.24436222637254368|0.21753381024651963|0.18268571643405696|0.10714151442880526|          4665.958|0.3609436157833492|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "# Drop null values\n",
    "df = df.na.drop()\n",
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|      NormLoad_lag21|     NormLoad_lag20|     NormLoad_lag19|     NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|     NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|     NormLoad_lag9|     NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|      NormLoad_lag5|     NormLoad_lag4|      NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|            features|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+\n",
      "|2016-01-03 00:00:...|N.Y.C.|   4919.133|      34.61|0.36263620014810116|0.14292484391302507|0.13131163940862922|0.09975432572999143|0.07640749647148892|0.061212496229800406|0.05648359867718655|0.06364351535541934|0.08511167564777489|0.10788707407343567|0.14099910928589954|0.17776120664482353|0.2031515053520584|0.21917107335995967|0.22299427494995927|0.21947367870577084|0.21895072691712025|0.2188965943400788|0.2383776791738833|0.2769630408770214|0.28345174186760513|0.27544365392350956|0.2617397729710814|0.24436222637254368|0.21753381024651963|0.18268571643405696|0.10714151442880526|          4665.958|0.3609436157833492|[0.13131163940862...|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "# Assemble all the features\n",
    "features = ['NormLoad_lag'+str(n) for n in range(LEN_SEQ_IN, 0, -1)] + ['NormTemp_next'+str(n) for n in range(1, LEN_SEQ_OUT+1)]\n",
    "vector_assembler = VectorAssembler(inputCols=features, outputCol='features')\n",
    "df = vector_assembler.transform(df)\n",
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|      NormLoad_lag21|     NormLoad_lag20|     NormLoad_lag19|     NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|     NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|     NormLoad_lag9|     NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|      NormLoad_lag5|     NormLoad_lag4|      NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|            features|      feature_matrix|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+\n",
      "|2016-01-03 00:00:...|N.Y.C.|   4919.133|      34.61|0.36263620014810116|0.14292484391302507|0.13131163940862922|0.09975432572999143|0.07640749647148892|0.061212496229800406|0.05648359867718655|0.06364351535541934|0.08511167564777489|0.10788707407343567|0.14099910928589954|0.17776120664482353|0.2031515053520584|0.21917107335995967|0.22299427494995927|0.21947367870577084|0.21895072691712025|0.2188965943400788|0.2383776791738833|0.2769630408770214|0.28345174186760513|0.27544365392350956|0.2617397729710814|0.24436222637254368|0.21753381024651963|0.18268571643405696|0.10714151442880526|          4665.958|0.3609436157833492|[0.13131163940862...|[WrappedArray(0.1...|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "# Reshape the vectors into the format that Keras requires\n",
    "reshape_transformer = ReshapeTransformer('features', 'feature_matrix', (LEN_SEQ_IN+LEN_EXTRA_IN, 1))\n",
    "df = reshape_transformer.transform(df)\n",
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|      NormLoad_lag21|     NormLoad_lag20|     NormLoad_lag19|     NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|     NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|     NormLoad_lag9|     NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|      NormLoad_lag5|     NormLoad_lag4|      NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|            features|      feature_matrix|              labels|   labels2|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+\n",
      "|2016-01-03 00:00:...|N.Y.C.|   4919.133|      34.61|0.36263620014810116|0.14292484391302507|0.13131163940862922|0.09975432572999143|0.07640749647148892|0.061212496229800406|0.05648359867718655|0.06364351535541934|0.08511167564777489|0.10788707407343567|0.14099910928589954|0.17776120664482353|0.2031515053520584|0.21917107335995967|0.22299427494995927|0.21947367870577084|0.21895072691712025|0.2188965943400788|0.2383776791738833|0.2769630408770214|0.28345174186760513|0.27544365392350956|0.2617397729710814|0.24436222637254368|0.21753381024651963|0.18268571643405696|0.10714151442880526|          4665.958|0.3609436157833492|[0.13131163940862...|[WrappedArray(0.1...|[0.10714151442880...|[4665.958]|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "# Assemble all the target variables\n",
    "targets = ['NormLoad_next'+str(n) for n in range(1, LEN_SEQ_OUT+1)]\n",
    "vector_assembler = VectorAssembler(inputCols=targets, outputCol='labels')\n",
    "df = vector_assembler.transform(df)\n",
    "\n",
    "targets = ['OrigNormLoad_next'+str(n) for n in range(1, LEN_SEQ_OUT+1)]\n",
    "vector_assembler = VectorAssembler(inputCols=targets, outputCol='labels2')\n",
    "df = vector_assembler.transform(df)\n",
    "\n",
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|      NormLoad_lag21|     NormLoad_lag20|     NormLoad_lag19|     NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|     NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|     NormLoad_lag9|     NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|      NormLoad_lag5|     NormLoad_lag4|      NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|            features|      feature_matrix|              labels|   labels2|        label_matrix|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "|2016-01-03 00:00:...|N.Y.C.|   4919.133|      34.61|0.36263620014810116|0.14292484391302507|0.13131163940862922|0.09975432572999143|0.07640749647148892|0.061212496229800406|0.05648359867718655|0.06364351535541934|0.08511167564777489|0.10788707407343567|0.14099910928589954|0.17776120664482353|0.2031515053520584|0.21917107335995967|0.22299427494995927|0.21947367870577084|0.21895072691712025|0.2188965943400788|0.2383776791738833|0.2769630408770214|0.28345174186760513|0.27544365392350956|0.2617397729710814|0.24436222637254368|0.21753381024651963|0.18268571643405696|0.10714151442880526|          4665.958|0.3609436157833492|[0.13131163940862...|[WrappedArray(0.1...|[0.10714151442880...|[4665.958]|[WrappedArray(0.1...|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "# Reshape the labels\n",
    "reshape_transformer = ReshapeTransformer('labels', 'label_matrix', (LEN_SEQ_OUT, 1))\n",
    "df = reshape_transformer.transform(df)\n",
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Partition the data into training and testing sets\n",
    "df_train = df.limit(df.count() - LEN_TEST_DATA)\n",
    "df2 = df\n",
    "df2 = df2.orderBy('TimeStamp', ascending=False)\n",
    "df_test = df2.limit(LEN_TEST_DATA).orderBy('TimeStamp', ascending=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|           NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|      NormLoad_lag21|     NormLoad_lag20|     NormLoad_lag19|     NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|     NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|     NormLoad_lag9|     NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|      NormLoad_lag5|     NormLoad_lag4|      NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|            features|      feature_matrix|              labels|   labels2|        label_matrix|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "|2016-01-03 00:00:...|N.Y.C.|   4919.133|      34.61|0.36263620014810116|0.14292484391302507|0.13131163940862922|0.09975432572999143|0.07640749647148892|0.061212496229800406|0.05648359867718655|0.06364351535541934|0.08511167564777489|0.10788707407343567|0.14099910928589954|0.17776120664482353|0.2031515053520584|0.21917107335995967|0.22299427494995927|0.21947367870577084|0.21895072691712025|0.2188965943400788|0.2383776791738833|0.2769630408770214|0.28345174186760513|0.27544365392350956|0.2617397729710814|0.24436222637254368|0.21753381024651963|0.18268571643405696|0.10714151442880526|          4665.958|0.3609436157833492|[0.13131163940862...|[WrappedArray(0.1...|[0.10714151442880...|[4665.958]|[WrappedArray(0.1...|\n",
      "+--------------------+------+-----------+-----------+-------------------+-------------------+-------------------+-------------------+-------------------+--------------------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+------------------+-------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "df_train.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------+-----------+-----------+------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+--------------------+--------------------+-------------------+-------------------+-------------------+------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+------------------+-------------------+------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "|           TimeStamp|  Name|HourAvgLoad|temperature|          NormTemp|           NormLoad|     NormLoad_lag24|     NormLoad_lag23|     NormLoad_lag22|     NormLoad_lag21|    NormLoad_lag20|      NormLoad_lag19|      NormLoad_lag18|     NormLoad_lag17|     NormLoad_lag16|     NormLoad_lag15|    NormLoad_lag14|     NormLoad_lag13|    NormLoad_lag12|     NormLoad_lag11|     NormLoad_lag10|      NormLoad_lag9|      NormLoad_lag8|     NormLoad_lag7|      NormLoad_lag6|     NormLoad_lag5|      NormLoad_lag4|     NormLoad_lag3|      NormLoad_lag2|      NormLoad_lag1|     NormLoad_next1|OrigNormLoad_next1|    NormTemp_next1|            features|      feature_matrix|              labels|   labels2|        label_matrix|\n",
      "+--------------------+------+-----------+-----------+------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+--------------------+--------------------+-------------------+-------------------+-------------------+------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+------------------+-------------------+------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "|2017-04-26 23:00:...|N.Y.C.|   4914.725|      58.49|0.6152544165873268|0.14230182459292462|0.25150687851208076|0.20217386818392424|0.14004507561741542|0.08370889280930005|0.0504542472306877|0.029099298933659146|0.019397553885941776|0.02007838057943645|0.04888892532276745|0.12240732516944353|0.2052009110102837|0.26132282492337355|0.2922452468629255|0.30659815338656526|0.31344740843467545|0.31123758378182353|0.31299498446833207|0.3109173111458553|0.31198794892937654|0.3003664054640266|0.29014524208569015|0.2808851792919008|0.24669543937985283|0.20098436487222615|0.08508100523206219|          4509.875|0.6065799217179731|[0.25150687851208...|[WrappedArray(0.2...|[0.08508100523206...|[4509.875]|[WrappedArray(0.0...|\n",
      "+--------------------+------+-----------+-----------+------------------+-------------------+-------------------+-------------------+-------------------+-------------------+------------------+--------------------+--------------------+-------------------+-------------------+-------------------+------------------+-------------------+------------------+-------------------+-------------------+-------------------+-------------------+------------------+-------------------+------------------+-------------------+------------------+-------------------+-------------------+-------------------+------------------+------------------+--------------------+--------------------+--------------------+----------+--------------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "df_test.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Select the desired columns to reduce network usage\n",
    "df_train = df_train.select('features', 'feature_matrix', 'labels', 'labels2', 'label_matrix')\n",
    "df_test = df_test.select('features', 'feature_matrix', 'labels', 'labels2', 'label_matrix')\n",
    "\n",
    "# Repartition the data\n",
    "df_train = df_train.repartition(num_workers)\n",
    "df_test = df_test.repartition(num_workers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DataFrame[features: vector, feature_matrix: array<array<double>>, labels: vector, labels2: vector, label_matrix: array<array<double>>]"
     ]
    }
   ],
   "source": [
    "# Cache the data\n",
    "df_train.cache()\n",
    "df_test.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "## Train and Evaluate GRU Model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "_________________________________________________________________\n",
      "Layer (type)                 Output Shape              Param #   \n",
      "=================================================================\n",
      "gru_1 (GRU)                  (None, 128)               49920     \n",
      "_________________________________________________________________\n",
      "dense_1 (Dense)              (None, 1)                 129       \n",
      "=================================================================\n",
      "Total params: 50,049\n",
      "Trainable params: 50,049\n",
      "Non-trainable params: 0\n",
      "_________________________________________________________________"
     ]
    }
   ],
   "source": [
    "# Create the GRU network\n",
    "gru_model = Sequential()\n",
    "gru_model.add(GRU(N_UNITS, input_shape=(LEN_SEQ_IN+LEN_EXTRA_IN, 1))) \n",
    "gru_model.add(Dense(LEN_SEQ_OUT, activation='linear')) \n",
    "gru_model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Fit the model\n",
    "gru_trainer = ADAG(keras_model=gru_model, worker_optimizer='adagrad', loss='mean_squared_error',\n",
    "                   num_workers=num_workers, batch_size=BATCH_SIZE, communication_window=COM_WINDOW, \n",
    "                   num_epoch=N_EPOCHS, features_col='feature_matrix', label_col='labels')\n",
    "trained_gru_model = gru_trainer.train(df_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# gru_trainer = DynSGD(keras_model=gru_model, worker_optimizer='adagrad', loss='mean_squared_error',\n",
    "#                    num_workers=num_workers, batch_size=BATCH_SIZE, communication_window=COM_WINDOW, \n",
    "#                    num_epoch=N_EPOCHS, features_col='feature_matrix', label_col='labels')\n",
    "# trained_gru_model = gru_trainer.train(df_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of parameter updates 1425\n",
      "Total training time in seconds 88.4753541946"
     ]
    }
   ],
   "source": [
    "print('Number of parameter updates ' + str(gru_trainer.parameter_server.num_updates))\n",
    "print('Total training time in seconds ' + str(gru_trainer.get_training_time()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+\n",
      "|            features|      feature_matrix|              labels|   labels2|        label_matrix|          prediction|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+\n",
      "|[0.20217386818392...|[WrappedArray(0.2...|[0.04815170463909...|[4248.592]|[WrappedArray(0.0...|[0.06627381592988...|\n",
      "|[0.02909929893365...|[WrappedArray(0.0...|[0.04732840891776...|[4242.767]|[WrappedArray(0.0...| [0.088986337184906]|\n",
      "|[0.12240732516944...|[WrappedArray(0.1...|[0.30439454761984...|[6061.567]|[WrappedArray(0.3...|[0.34112250804901...|\n",
      "|[0.30659815338656...|[WrappedArray(0.3...|[0.32994851610309...|[6242.367]|[WrappedArray(0.3...|[0.3110193908214569]|\n",
      "|[0.31091731114585...|[WrappedArray(0.3...|[0.32862573075445...|[6233.008]|[WrappedArray(0.3...|[0.3101566433906555]|\n",
      "|[0.28088517929190...|[WrappedArray(0.2...|[0.24665544063493...| [5653.05]|[WrappedArray(0.2...|[0.2449420541524887]|\n",
      "|[0.08508100523206...|[WrappedArray(0.0...|[0.04970642617291...|[4259.592]|[WrappedArray(0.0...|[0.07105318456888...|\n",
      "|[0.01904420808279...|[WrappedArray(0.0...|[0.04890899536438...| [4253.95]|[WrappedArray(0.0...|[0.0876825749874115]|\n",
      "|[0.26512369505861...|[WrappedArray(0.2...|[0.3075934578448565]|  [6084.2]|[WrappedArray(0.3...|[0.3308871388435364]|\n",
      "|[0.32930684012459...|[WrappedArray(0.3...|[0.3858736870729994]| [6638.05]|[WrappedArray(0.3...| [0.362085223197937]|\n",
      "|[0.33330289848148...|[WrappedArray(0.3...|[0.3963185890138855]| [6711.95]|[WrappedArray(0.3...|[0.3790944516658783]|\n",
      "|[0.27702791516647...|[WrappedArray(0.2...|[0.2867272574627342]|[5936.567]|[WrappedArray(0.2...|[0.2548230290412903]|\n",
      "|[0.08535195079391...|[WrappedArray(0.0...|[0.10251395645253...|[4633.217]|[WrappedArray(0.1...|[0.1238204687833786]|\n",
      "|[0.02187337725939...|[WrappedArray(0.0...|[0.0592538297738221]|[4327.142]|[WrappedArray(0.0...|[0.10758449882268...|\n",
      "|[0.26322622809574...|[WrappedArray(0.2...|[0.21015128571230...|[5394.775]|[WrappedArray(0.2...|[0.19438238441944...|\n",
      "|[0.37266562095154...|[WrappedArray(0.3...|[0.3198357592171668]|[6170.817]|[WrappedArray(0.3...|[0.3415073752403259]|\n",
      "|[0.40547024531527...|[WrappedArray(0.4...|[0.3255458273959462]|[6211.217]|[WrappedArray(0.3...|[0.31900525093078...|\n",
      "|[0.31375114448705...|[WrappedArray(0.3...|[0.29787531755187...|[6015.442]|[WrappedArray(0.2...|[0.2738603949546814]|\n",
      "|[0.13941032521665...|[WrappedArray(0.1...|[0.11792096417610...|[4742.225]|[WrappedArray(0.1...|[0.15239261090755...|\n",
      "|[0.05466386779100...|[WrappedArray(0.0...|[0.04536493695886...|[4228.875]|[WrappedArray(0.0...|[0.07289811968803...|\n",
      "|[0.16371217751188...|[WrappedArray(0.1...|[0.12760504193366...|[4810.742]|[WrappedArray(0.1...|[0.13266727328300...|\n",
      "|[0.31214101833129...|[WrappedArray(0.3...|[0.17845970062847...| [5170.55]|[WrappedArray(0.1...|[0.2104790210723877]|\n",
      "|[0.33483641926712...|[WrappedArray(0.3...|[0.17592027501045...|[5152.583]|[WrappedArray(0.1...|[0.16473780572414...|\n",
      "|[0.31234242543908...|[WrappedArray(0.3...|[0.18121339514152...|[5190.033]|[WrappedArray(0.1...|[0.19772745668888...|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+"
     ]
    }
   ],
   "source": [
    "predictor = ModelPredictor(keras_model=trained_gru_model, features_col='feature_matrix')\n",
    "df_pred = predictor.predict(df_test.limit(24))\n",
    "df_pred.show(24)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+--------------------+\n",
      "|            features|      feature_matrix|              labels|   labels2|        label_matrix|          prediction|         prediction2|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+--------------------+\n",
      "|[0.14004507561741...|[WrappedArray(0.1...|[0.02862694626401...| [4110.45]|[WrappedArray(0.0...|[0.01342243980616...|[4002.8747414102418]|\n",
      "|[0.01939755388594...|[WrappedArray(0.0...|[0.11942508659092...|[4752.867]|[WrappedArray(0.1...|[0.1151493564248085]| [4722.615259862646]|\n",
      "|[0.20520091101028...|[WrappedArray(0.2...|[0.3104531560988476]|[6104.433]|[WrappedArray(0.3...|[0.3623005151748657]| [6471.264575576543]|\n",
      "|[0.31344740843467...|[WrappedArray(0.3...|[0.32840439494336...|[6231.442]|[WrappedArray(0.3...|[0.3151266276836395]| [6137.498848973095]|\n",
      "|[0.31198794892937...|[WrappedArray(0.3...|[0.3032673745078247]|[6053.592]|[WrappedArray(0.3...|[0.30896949768066...| [6093.935787319183]|\n",
      "|[0.24669543937985...|[WrappedArray(0.2...|[0.19863461528132...|[5313.292]|[WrappedArray(0.1...|[0.23976676166057...|[5604.3110669696625]|\n",
      "|[0.04815170463909...|[WrappedArray(0.0...|[0.02853974051980...|[4109.833]|[WrappedArray(0.0...|[0.01929968595504...|[4044.4575626622436]|\n",
      "|[0.04732840891776...|[WrappedArray(0.0...|[0.11889860134424...|[4749.142]|[WrappedArray(0.1...|[0.11785231530666...| [4741.739294008642]|\n",
      "|[0.30439454761984...|[WrappedArray(0.3...|[0.33253147392406...|[6260.642]|[WrappedArray(0.3...|[0.36142104864120...| [6465.042154609322]|\n",
      "|[0.32994851610309...|[WrappedArray(0.3...|[0.39107606800182...|[6674.858]|[WrappedArray(0.3...|[0.37049973011016...| [6529.275841469525]|\n",
      "|[0.32862573075445...|[WrappedArray(0.3...|[0.35676746256159...|[6432.117]|[WrappedArray(0.3...|[0.38239574432373...| [6613.442782945633]|\n",
      "|[0.24665544063493...|[WrappedArray(0.2...|[0.24569900421499...|[5646.283]|[WrappedArray(0.2...|[0.2590906023979187]| [5741.031530079006]|\n",
      "|[0.04970642617291...|[WrappedArray(0.0...|[0.0769458541371566]|[4452.317]|[WrappedArray(0.0...|[0.07341497391462...| [4427.335238570168]|\n",
      "|[0.04890899536438...|[WrappedArray(0.0...|[0.07627562781775...|[4447.575]|[WrappedArray(0.0...|[0.12370164692401...|  [4783.12461375305]|\n",
      "|[0.30759345784485...|[WrappedArray(0.3...|[0.2619305797047783]|[5761.125]|[WrappedArray(0.2...|  [0.25412717461586]| [5705.914176639973]|\n",
      "|[0.38587368707299...|[WrappedArray(0.3...|[0.32584489928372...|[6213.333]|[WrappedArray(0.3...|[0.32432347536087...| [6202.568587989687]|\n",
      "|[0.39631858901388...|[WrappedArray(0.3...|[0.31762381448949...|[6155.167]|[WrappedArray(0.3...|[0.3144816756248474]| [6132.935669977784]|\n",
      "|[0.28672725746273...|[WrappedArray(0.2...|[0.26529202899923...|[5784.908]|[WrappedArray(0.2...|[0.2911844253540039]| [5968.102452322006]|\n",
      "|[0.10251395645253...|[WrappedArray(0.1...|[0.08365348818736...|[4499.775]|[WrappedArray(0.0...|  [0.08798848092556]| [4530.446035991102]|\n",
      "|[0.05925382977382...|[WrappedArray(0.0...|[0.04410236173508...|[4219.942]|[WrappedArray(0.0...|[0.09805912524461...| [4601.698080231472]|\n",
      "|[0.21015128571230...|[WrappedArray(0.2...|[0.15446638988854...|[5000.792]|[WrappedArray(0.1...|[0.17073442041873...| [5115.891927503913]|\n",
      "|[0.31983575921716...|[WrappedArray(0.3...|[0.17271189511792...|[5129.883]|[WrappedArray(0.1...|[0.19983722269535...| [5321.800714433043]|\n",
      "|[0.32554582739594...|[WrappedArray(0.3...|[0.18244303853645...|[5198.733]|[WrappedArray(0.1...|[0.17319385707378...| [5133.292987833291]|\n",
      "|[0.29787531755187...|[WrappedArray(0.2...|[0.14529466354553...|  [4935.9]|[WrappedArray(0.1...|[0.2081742137670517]| [5380.786777077346]|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+--------------------+"
     ]
    }
   ],
   "source": [
    "# Convert the values to the original range\n",
    "inverse_transformer = MinMaxTransformer(n_min=orig_min, n_max=orig_max, \\\n",
    "                                        o_min=0.0, o_max=1.0, \\\n",
    "                                        input_col='prediction', \\\n",
    "                                        output_col='prediction2', \\\n",
    "                                        is_vector=True)\n",
    "df_pred = inverse_transformer.transform(df_pred)\n",
    "df_pred.show(24)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2.8088257141182944"
     ]
    }
   ],
   "source": [
    "# Compute Mean-Absolute-Percentage Errors (MAPEs)\n",
    "def get_MAPE(actual, pred): \n",
    "    \"\"\"\n",
    "    Compute the mean-absolute-percentage-error (MAPE)\n",
    "    \"\"\"\n",
    "    actual, pred = np.array(actual), np.array(pred)\n",
    "    mape = np.mean(np.abs((actual - pred) / actual)) * 100\n",
    "    if mape == np.inf:\n",
    "        mape = np.nan\n",
    "    return mape\n",
    "\n",
    "actual = df_pred.select('labels2').rdd.map(lambda x: list(x[0])).collect()\n",
    "pred = df_pred.select('prediction2').rdd.map(lambda x: list(x[0])).collect()\n",
    "get_MAPE(actual, pred)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train and Evaluate LSTM Model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "_________________________________________________________________\n",
      "Layer (type)                 Output Shape              Param #   \n",
      "=================================================================\n",
      "lstm_1 (LSTM)                (None, 128)               66560     \n",
      "_________________________________________________________________\n",
      "dense_2 (Dense)              (None, 1)                 129       \n",
      "=================================================================\n",
      "Total params: 66,689\n",
      "Trainable params: 66,689\n",
      "Non-trainable params: 0\n",
      "_________________________________________________________________"
     ]
    }
   ],
   "source": [
    "# Create the LSTM network\n",
    "lstm_model = Sequential()\n",
    "lstm_model.add(LSTM(N_UNITS, input_shape=(LEN_SEQ_IN+LEN_EXTRA_IN, 1))) \n",
    "lstm_model.add(Dense(LEN_SEQ_OUT, activation='linear')) \n",
    "lstm_model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Fit the model\n",
    "lstm_trainer = ADAG(keras_model=lstm_model, worker_optimizer='adam', loss='mean_squared_error',\n",
    "                    num_workers=num_workers, batch_size=BATCH_SIZE, communication_window=COM_WINDOW, \n",
    "                    num_epoch=N_EPOCHS, features_col='feature_matrix', label_col='labels')\n",
    "trained_lstm_model = lstm_trainer.train(df_train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of parameter updates 1425\n",
      "Total training time in seconds 99.2543179989"
     ]
    }
   ],
   "source": [
    "print('Number of parameter updates ' + str(lstm_trainer.parameter_server.num_updates))\n",
    "print('Total training time in seconds ' + str(lstm_trainer.get_training_time()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+\n",
      "|            features|      feature_matrix|              labels|   labels2|        label_matrix|          prediction|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+\n",
      "|[0.14004507561741...|[WrappedArray(0.1...|[0.02862694626401...| [4110.45]|[WrappedArray(0.0...|[-0.0090088704600...|\n",
      "|[0.01939755388594...|[WrappedArray(0.0...|[0.11942508659092...|[4752.867]|[WrappedArray(0.1...|[0.1277359277009964]|\n",
      "|[0.20520091101028...|[WrappedArray(0.2...|[0.3104531560988476]|[6104.433]|[WrappedArray(0.3...|[0.3263910710811615]|\n",
      "|[0.31344740843467...|[WrappedArray(0.3...|[0.32840439494336...|[6231.442]|[WrappedArray(0.3...|[0.3154873549938202]|\n",
      "|[0.31198794892937...|[WrappedArray(0.3...|[0.3032673745078247]|[6053.592]|[WrappedArray(0.3...|[0.3446936011314392]|\n",
      "|[0.24669543937985...|[WrappedArray(0.2...|[0.19863461528132...|[5313.292]|[WrappedArray(0.1...|[0.1838761866092682]|\n",
      "|[0.04815170463909...|[WrappedArray(0.0...|[0.02853974051980...|[4109.833]|[WrappedArray(0.0...|[0.02234012633562...|\n",
      "|[0.04732840891776...|[WrappedArray(0.0...|[0.11889860134424...|[4749.142]|[WrappedArray(0.1...|[0.13599708676338...|\n",
      "|[0.30439454761984...|[WrappedArray(0.3...|[0.33253147392406...|[6260.642]|[WrappedArray(0.3...|[0.31306833028793...|\n",
      "|[0.32994851610309...|[WrappedArray(0.3...|[0.39107606800182...|[6674.858]|[WrappedArray(0.3...|[0.3611087203025818]|\n",
      "|[0.32862573075445...|[WrappedArray(0.3...|[0.35676746256159...|[6432.117]|[WrappedArray(0.3...|[0.4186570942401886]|\n",
      "|[0.24665544063493...|[WrappedArray(0.2...|[0.24569900421499...|[5646.283]|[WrappedArray(0.2...|[0.20829132199287...|\n",
      "|[0.04970642617291...|[WrappedArray(0.0...|[0.0769458541371566]|[4452.317]|[WrappedArray(0.0...|[0.05747207254171...|\n",
      "|[0.04890899536438...|[WrappedArray(0.0...|[0.07627562781775...|[4447.575]|[WrappedArray(0.0...| [0.151891827583313]|\n",
      "|[0.30759345784485...|[WrappedArray(0.3...|[0.2619305797047783]|[5761.125]|[WrappedArray(0.2...|[0.21469919383525...|\n",
      "|[0.38587368707299...|[WrappedArray(0.3...|[0.32584489928372...|[6213.333]|[WrappedArray(0.3...|[0.3192870020866394]|\n",
      "|[0.39631858901388...|[WrappedArray(0.3...|[0.31762381448949...|[6155.167]|[WrappedArray(0.3...|[0.35652151703834...|\n",
      "|[0.28672725746273...|[WrappedArray(0.2...|[0.26529202899923...|[5784.908]|[WrappedArray(0.2...|[0.2566663324832916]|\n",
      "|[0.10251395645253...|[WrappedArray(0.1...|[0.08365348818736...|[4499.775]|[WrappedArray(0.0...|[0.08822667598724...|\n",
      "|[0.05925382977382...|[WrappedArray(0.0...|[0.04410236173508...|[4219.942]|[WrappedArray(0.0...|[0.09207779169082...|\n",
      "|[0.21015128571230...|[WrappedArray(0.2...|[0.15446638988854...|[5000.792]|[WrappedArray(0.1...|[0.16198866069316...|\n",
      "|[0.31983575921716...|[WrappedArray(0.3...|[0.17271189511792...|[5129.883]|[WrappedArray(0.1...|[0.19730426371097...|\n",
      "|[0.32554582739594...|[WrappedArray(0.3...|[0.18244303853645...|[5198.733]|[WrappedArray(0.1...|[0.19240206480026...|\n",
      "|[0.29787531755187...|[WrappedArray(0.2...|[0.14529466354553...|  [4935.9]|[WrappedArray(0.1...|[0.1871282160282135]|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+"
     ]
    }
   ],
   "source": [
    "predictor = ModelPredictor(keras_model=trained_lstm_model, features_col='feature_matrix')\n",
    "df_pred = predictor.predict(df_test.limit(24))\n",
    "df_pred.show(24)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+--------------------+\n",
      "|            features|      feature_matrix|              labels|   labels2|        label_matrix|          prediction|         prediction2|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+--------------------+\n",
      "|[0.14004507561741...|[WrappedArray(0.1...|[0.02862694626401...| [4110.45]|[WrappedArray(0.0...|[-0.0090088704600...| [3844.168241525599]|\n",
      "|[0.01939755388594...|[WrappedArray(0.0...|[0.11942508659092...|[4752.867]|[WrappedArray(0.1...|[0.1277359277009964]| [4811.668045860499]|\n",
      "|[0.20520091101028...|[WrappedArray(0.2...|[0.3104531560988476]|[6104.433]|[WrappedArray(0.3...|[0.3263910710811615]| [6217.197286716997]|\n",
      "|[0.31344740843467...|[WrappedArray(0.3...|[0.32840439494336...|[6231.442]|[WrappedArray(0.3...|[0.3154873549938202]|[6140.0510747740855]|\n",
      "|[0.31198794892937...|[WrappedArray(0.3...|[0.3032673745078247]|[6053.592]|[WrappedArray(0.3...|[0.3446936011314392]| [6346.691749984383]|\n",
      "|[0.24669543937985...|[WrappedArray(0.2...|[0.19863461528132...|[5313.292]|[WrappedArray(0.1...|[0.1838761866092682]| [5208.872840773999]|\n",
      "|[0.04815170463909...|[WrappedArray(0.0...|[0.02853974051980...|[4109.833]|[WrappedArray(0.0...|[0.02234012633562...|[4065.9693533325635]|\n",
      "|[0.04732840891776...|[WrappedArray(0.0...|[0.11889860134424...|[4749.142]|[WrappedArray(0.1...|[0.13599708676338...| [4870.117580204188]|\n",
      "|[0.30439454761984...|[WrappedArray(0.3...|[0.33253147392406...|[6260.642]|[WrappedArray(0.3...|[0.31306833028793...| [6122.935937956452]|\n",
      "|[0.32994851610309...|[WrappedArray(0.3...|[0.39107606800182...|[6674.858]|[WrappedArray(0.3...|[0.3611087203025818]| [6462.832362276673]|\n",
      "|[0.32862573075445...|[WrappedArray(0.3...|[0.35676746256159...|[6432.117]|[WrappedArray(0.3...|[0.4186570942401886]| [6869.999883624255]|\n",
      "|[0.24665544063493...|[WrappedArray(0.2...|[0.24569900421499...|[5646.283]|[WrappedArray(0.2...|[0.20829132199287...| [5381.615343773066]|\n",
      "|[0.04970642617291...|[WrappedArray(0.0...|[0.0769458541371566]|[4452.317]|[WrappedArray(0.0...|[0.05747207254171...| [4314.535672032728]|\n",
      "|[0.04890899536438...|[WrappedArray(0.0...|[0.07627562781775...|[4447.575]|[WrappedArray(0.0...| [0.151891827583313]| [4982.576400137662]|\n",
      "|[0.30759345784485...|[WrappedArray(0.3...|[0.2619305797047783]|[5761.125]|[WrappedArray(0.2...|[0.21469919383525...| [5426.952459605484]|\n",
      "|[0.38587368707299...|[WrappedArray(0.3...|[0.32584489928372...|[6213.333]|[WrappedArray(0.3...|[0.3192870020866394]| [6166.934421477436]|\n",
      "|[0.39631858901388...|[WrappedArray(0.3...|[0.31762381448949...|[6155.167]|[WrappedArray(0.3...|[0.35652151703834...| [6430.376880823075]|\n",
      "|[0.28672725746273...|[WrappedArray(0.2...|[0.26529202899923...|[5784.908]|[WrappedArray(0.2...|[0.2566663324832916]| [5723.879282245099]|\n",
      "|[0.10251395645253...|[WrappedArray(0.1...|[0.08365348818736...|[4499.775]|[WrappedArray(0.0...|[0.08822667598724...| [4532.131318931817]|\n",
      "|[0.05925382977382...|[WrappedArray(0.0...|[0.04410236173508...|[4219.942]|[WrappedArray(0.0...|[0.09207779169082...| [4559.378817482352]|\n",
      "|[0.21015128571230...|[WrappedArray(0.2...|[0.15446638988854...|[5000.792]|[WrappedArray(0.1...|[0.16198866069316...|[5054.0137358868415]|\n",
      "|[0.31983575921716...|[WrappedArray(0.3...|[0.17271189511792...|[5129.883]|[WrappedArray(0.1...|[0.19730426371097...|[5303.8794673016955]|\n",
      "|[0.32554582739594...|[WrappedArray(0.3...|[0.18244303853645...|[5198.733]|[WrappedArray(0.1...|[0.19240206480026...| [5269.195321720242]|\n",
      "|[0.29787531755187...|[WrappedArray(0.2...|[0.14529466354553...|  [4935.9]|[WrappedArray(0.1...|[0.1871282160282135]| [5231.881670863569]|\n",
      "+--------------------+--------------------+--------------------+----------+--------------------+--------------------+--------------------+"
     ]
    }
   ],
   "source": [
    "# Convert the values to the original range\n",
    "df_pred = inverse_transformer.transform(df_pred)\n",
    "df_pred.show(24)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.5870739850117217"
     ]
    }
   ],
   "source": [
    "# Compute MAPE\n",
    "actual = df_pred.select('labels2').rdd.map(lambda x: list(x[0])).collect()\n",
    "pred = df_pred.select('prediction2').rdd.map(lambda x: list(x[0])).collect()\n",
    "get_MAPE(actual, pred)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 2
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}