CADS_cogni_pyspark_competition.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "sc = pyspark.SparkContext()\n",
    "from pyspark.sql import SQLContext\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['state05_CO.txt.gz',\n",
       " 'state06_CT.txt.gz',\n",
       " 'state03_AR.txt.gz',\n",
       " 'state04_CA.txt.gz',\n",
       " 'state01_AL.txt.gz',\n",
       " 'state02_AZ.txt.gz']"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from statistics import mean\n",
    "import numpy as np\n",
    "import re\n",
    "import os\n",
    "files = os.listdir('WeatherDatasets/')\n",
    "files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def datacleaner(row):\n",
    "    row = row.replace('-9999','')\n",
    "    return row[:16]+re.sub(r'[A-Za-z]','',row[16:])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['CO', 1893, 10, 'TMAX', 35.0],\n",
       " ['CO', 1893, 10, 'TMIN', 22.532258064516128],\n",
       " ['CO', 1893, 10, 'PRCP', 4.870967741935484],\n",
       " ['CO', 1893, 11, 'TMAX', 28.316666666666666],\n",
       " ['CO', 1893, 11, 'TMIN', 17.15]]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "combinedRdd = sc.emptyRDD()\n",
    "for file in files:\n",
    "    textRdd = sc.textFile(os.path.join('WeatherDatasets',files[0])).map(datacleaner)\n",
    "    def getRow(X):\n",
    "        return [file[8:10],int(X[6:10]),int(X[10:12]),X[12:16],mean(map(float,X[16:].split()))]\n",
    "    newRdd = textRdd.map(getRow)\n",
    "    combinedRdd = combinedRdd.union(newRdd)\n",
    "combinedRdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----+-----+------+------------------+\n",
      "|STATE|YEAR|MONTH|METRIC|             VALUE|\n",
      "+-----+----+-----+------+------------------+\n",
      "|   CO|1893|   10|  TMAX|              35.0|\n",
      "|   CO|1893|   10|  TMIN|22.532258064516128|\n",
      "|   CO|1893|   10|  PRCP| 4.870967741935484|\n",
      "|   CO|1893|   11|  TMAX|28.316666666666666|\n",
      "|   CO|1893|   11|  TMIN|             17.15|\n",
      "|   CO|1893|   11|  PRCP| 4.333333333333333|\n",
      "|   CO|1893|   11|  SNOW|               4.5|\n",
      "|   CO|1893|   12|  TMAX|27.366666666666667|\n",
      "|   CO|1893|   12|  TMIN|              17.0|\n",
      "|   CO|1893|   12|  PRCP| 3.629032258064516|\n",
      "|   CO|1893|   12|  SNOW|11.333333333333334|\n",
      "|   CO|1894|    1|  TMAX|24.541666666666668|\n",
      "|   CO|1894|    1|  TMIN|13.977272727272727|\n",
      "|   CO|1894|    1|  PRCP|3.2758620689655173|\n",
      "|   CO|1894|    1|  SNOW|3.3225806451612905|\n",
      "|   CO|1894|    2|  TMAX|              21.5|\n",
      "|   CO|1894|    2|  TMIN|             11.25|\n",
      "|   CO|1894|    2|  PRCP| 4.464285714285714|\n",
      "|   CO|1894|    2|  SNOW|              16.0|\n",
      "|   CO|1894|    3|  TMAX|              30.0|\n",
      "+-----+----+-----+------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_data = sqlContext.createDataFrame(combinedRdd, ['STATE','YEAR',\"MONTH\",\"METRIC\",\"VALUE\"])\n",
    "df_data.show(20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+----+-----+------------------+------------------+------------------+------------------+------------------+\n",
      "|STATE|YEAR|MONTH|              PRCP|              SNOW|              SNWD|              TMAX|              TMIN|\n",
      "+-----+----+-----+------------------+------------------+------------------+------------------+------------------+\n",
      "|   AR|1923|    4| 3.163492063492064|            1.9625|3.9681972789115645|31.694614729672203|15.805469544139493|\n",
      "|   CA|1948|    1|  2.86231884057971| 4.699169110459432| 4.790610405126535| 20.47539738195418| 5.215109179281327|\n",
      "|   AL|1987|    1|1.0657504937458855| 1.818440352596359|1.0299999999999998|20.021582181259603|  5.32347670250896|\n",
      "|   CA|1973|    1|0.6438172043010754| 1.207807386629266|2.8924541434135422|17.141129032258068|3.2896505376344085|\n",
      "|   AR|1896|    1|3.8064516129032255| 7.204637096774194|               0.0| 25.82459677419355| 9.578629032258064|\n",
      "|   AZ|1940|    9| 5.613079918374147|              1.25|               1.2| 39.69184946217367| 24.76135585481069|\n",
      "|   CO|1939|    6|2.7285997806471944|              1.25|               1.2| 42.46923196801071| 24.04147433558353|\n",
      "|   CO|1933|    6|2.6335914218816523|              1.25|               1.0|43.749008828918875|26.028621800211003|\n",
      "|   AR|1903|    6| 8.658059210526316|               1.2|               3.0| 40.16149425287357|26.781193555008205|\n",
      "|   AR|1906|    3| 6.311535058550261|5.2874324205024985| 3.835547875064004| 25.53016726403823|12.331589008363201|\n",
      "+-----+----+-----+------------------+------------------+------------------+------------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import rand\n",
    "data = df_data.groupBy(['STATE','YEAR','MONTH']).pivot('METRIC').avg('VALUE').fillna(0)\n",
    "data.orderBy(rand()).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1893\n",
      "+-----+----+-----+------------------+-------------------+--------------------+------------------+------------------+\n",
      "|STATE|YEAR|MONTH|              PRCP|               SNOW|                SNWD|              TMAX|              TMIN|\n",
      "+-----+----+-----+------------------+-------------------+--------------------+------------------+------------------+\n",
      "|   CA|  49|   10| 6.141084296561395| 1.8779803646563815|  2.7615865701119158|33.491367219615995| 17.91606616046816|\n",
      "|   AZ| 102|    9|2.9816228790366717| 0.2616666666666666|0.015892857142857143|37.771653373993274| 21.46551724137931|\n",
      "|   CA| 107|   10|2.5755904607072573|0.11303123399897592|0.001536098310291...|32.155666790260774|16.682343256175294|\n",
      "|   AR| 100|    2| 2.178679261559697| 2.7653059903059902|  3.0509509009509004|19.093890484515484| 6.775484931734931|\n",
      "|   CT| 117|    3|2.5027697441601777| 1.8899481566820282|   1.391853689228828|25.633870967741935| 11.10725806451613|\n",
      "+-----+----+-----+------------------+-------------------+--------------------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "firstYear = data.agg({\"YEAR\": \"min\"}).collect()[0][0]\n",
    "data = data.withColumn('YEAR',data.YEAR-firstYear)\n",
    "print(firstYear)\n",
    "data.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import CountVectorizer\n",
    "from pyspark.sql.functions import col, split\n",
    "data = data.withColumn(\"STATE\", split(col(\"STATE\"),\" \"))\n",
    "stateVectorizer = CountVectorizer(inputCol=\"STATE\", outputCol=\"STATE_OHE\", vocabSize=6, minDF=1.0)\n",
    "stateVectorizer_model = stateVectorizer.fit(data)\n",
    "ohe_data = stateVectorizer_model.transform(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+\n",
      "|            features|              TMAX|              TMIN|\n",
      "+--------------------+------------------+------------------+\n",
      "|(11,[3,6,7,8,9,10...|33.491367219615995| 17.91606616046816|\n",
      "|(11,[2,6,7,8,9,10...|37.771653373993274| 21.46551724137931|\n",
      "|(11,[3,6,7,8,9,10...|32.155666790260774|16.682343256175294|\n",
      "|(11,[1,6,7,8,9,10...|19.093890484515484| 6.775484931734931|\n",
      "|(11,[5,6,7,8,9,10...|25.633870967741935| 11.10725806451613|\n",
      "+--------------------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "vectorAssembler = VectorAssembler(inputCols=['STATE_OHE','YEAR','MONTH','PRCP','SNOW','SNWD'],\n",
    "                                  outputCol='features')\n",
    "X = vectorAssembler.transform(ohe_data)\n",
    "X = X.select(['features','TMAX','TMIN'])\n",
    "X.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "#train test split\n",
    "(X_train,X_val) = X.randomSplit([0.7,0.3])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train on Maximum temperature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+------------------+\n",
      "|            features|              TMAX|              TMIN|        prediction|\n",
      "+--------------------+------------------+------------------+------------------+\n",
      "|(11,[0,6,7,8,9,10...|26.820138888888895|10.126173371647512|25.715864729074827|\n",
      "|(11,[1,6,7,8,9,10...| 36.82249351863658|  21.3342431545713| 39.61775247674409|\n",
      "|(11,[1,6,7,8,9,10...|22.096496106785313| 7.093614116695317|24.102587754612358|\n",
      "|(11,[2,6,7,8,9,10...| 39.67297851074463|21.075180966064586|37.302877916617085|\n",
      "|(11,[3,6,7,8,9,10...| 34.18022977758433|17.782702499556674|36.598142006856385|\n",
      "+--------------------+------------------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.regression import RandomForestRegressor\n",
    "rf_max = RandomForestRegressor(featuresCol='features',labelCol='TMAX')\n",
    "rf_max_model = rf_max.fit(X_train)\n",
    "predictions = rf_max_model.transform(X_val)\n",
    "predictions.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2.1904977078212995"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "lr_evaluator = RegressionEvaluator(predictionCol=\"prediction\",labelCol=\"TMAX\",metricName=\"mae\")\n",
    "lr_evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train on Minimum temperature"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------+------------------+------------------+\n",
      "|            features|              TMAX|              TMIN|        prediction|\n",
      "+--------------------+------------------+------------------+------------------+\n",
      "|(11,[0,6,7,8,9,10...|26.820138888888895|10.126173371647512| 9.990956619569086|\n",
      "|(11,[1,6,7,8,9,10...| 36.82249351863658|  21.3342431545713|23.099525048749722|\n",
      "|(11,[1,6,7,8,9,10...|22.096496106785313| 7.093614116695317|  9.61731307202829|\n",
      "|(11,[2,6,7,8,9,10...| 39.67297851074463|21.075180966064586|20.727313527185125|\n",
      "|(11,[3,6,7,8,9,10...| 34.18022977758433|17.782702499556674| 19.74247482058521|\n",
      "+--------------------+------------------+------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.regression import RandomForestRegressor\n",
    "rf_min = RandomForestRegressor(featuresCol='features',labelCol='TMIN')\n",
    "rf_min_model = rf_min.fit(X_train)\n",
    "predictions = rf_min_model.transform(X_val)\n",
    "predictions.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.7878727616183605"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "lr_evaluator = RegressionEvaluator(predictionCol=\"prediction\",labelCol=\"TMIN\",metricName=\"mae\")\n",
    "lr_evaluator.evaluate(predictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Results\n",
    "#### Using Year, State, Month, PRCP, SNOW, SNWD as independent variables\n",
    "1. Mean Absolute Error on TMAX : **2.1904**\n",
    "2. Mean Absolute Error on TMIN : **1.7878**"
   ]
  },
  {
   "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.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}