bike_dataframe.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'local[*]'"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.master"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.types \n",
    "from pyspark import SparkConf, SparkContext\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import  StringIndexer, OneHotEncoder,VectorAssembler\n",
    "from pyspark.ml.classification import DecisionTreeClassifier\n",
    "from pyspark.sql.functions import udf,col\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.sql import SQLContext\n",
    "\n",
    "global Path    \n",
    "Path=\"file:/home/vivid/Downloads/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def CreateSparkContext():\n",
    "    def SetLogger( sc ):\n",
    "        logger = sc._jvm.org.apache.log4j\n",
    "        logger.LogManager.getLogger(\"org\"). setLevel( logger.Level.ERROR )\n",
    "        logger.LogManager.getLogger(\"akka\").setLevel( logger.Level.ERROR )\n",
    "        logger.LogManager.getRootLogger().setLevel(logger.Level.ERROR)    \n",
    "\n",
    "    sparkConf = SparkConf().setAppName(\"RunDecisionTreeBinary\").set(\"spark.ui.showConsoleProgress\", \"false\") \n",
    "    sc = SparkContext(conf = sparkConf)\n",
    "    print((\"master=\"+sc.master))    \n",
    "    SetLogger(sc)\n",
    "    return (sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "master=local[*]\n"
     ]
    }
   ],
   "source": [
    "sc = CreateSparkContext()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "read data\n"
     ]
    }
   ],
   "source": [
    "print(\"read data\")\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "row_df = sqlContext.read.format(\"csv\").option(\"header\", \"true\").load(Path+\"data/hour.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[instant: string, dteday: string, season: string, yr: string, mnth: string, hr: string, holiday: string, weekday: string, workingday: string, weathersit: string, temp: string, atemp: string, hum: string, windspeed: string, casual: string, registered: string, cnt: string]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "row_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['instant', 'dteday', 'season', 'yr', 'mnth', 'hr', 'holiday', 'weekday', 'workingday', 'weathersit', 'temp', 'atemp', 'hum', 'windspeed', 'casual', 'registered', 'cnt']\n"
     ]
    }
   ],
   "source": [
    "print(row_df.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "df= row_df.drop(\"instant\").drop(\"dteday\") \\\n",
    "                            .drop('yr').drop(\"casual\").drop(\"registered\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "hour_df= df.select([ col(column).cast(\"double\").alias(column) for column in df.columns])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[season: double, mnth: double, hr: double, holiday: double, weekday: double, workingday: double, weathersit: double, temp: double, atemp: double, hum: double, windspeed: double, cnt: double]"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "hour_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----+---+-------+-------+----------+----------+----+------+----+---------+----+\n",
      "|season|mnth| hr|holiday|weekday|workingday|weathersit|temp| atemp| hum|windspeed| cnt|\n",
      "+------+----+---+-------+-------+----------+----------+----+------+----+---------+----+\n",
      "|   1.0| 1.0|0.0|    0.0|    6.0|       0.0|       1.0|0.24|0.2879|0.81|      0.0|16.0|\n",
      "|   1.0| 1.0|1.0|    0.0|    6.0|       0.0|       1.0|0.22|0.2727| 0.8|      0.0|40.0|\n",
      "|   1.0| 1.0|2.0|    0.0|    6.0|       0.0|       1.0|0.22|0.2727| 0.8|      0.0|32.0|\n",
      "|   1.0| 1.0|3.0|    0.0|    6.0|       0.0|       1.0|0.24|0.2879|0.75|      0.0|13.0|\n",
      "|   1.0| 1.0|4.0|    0.0|    6.0|       0.0|       1.0|0.24|0.2879|0.75|      0.0| 1.0|\n",
      "+------+----+---+-------+-------+----------+----------+----+------+----+---------+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "hour_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<bound method DataFrame.cache of DataFrame[season: double, mnth: double, hr: double, holiday: double, weekday: double, workingday: double, weathersit: double, temp: double, atemp: double, hum: double, windspeed: double, cnt: double]>"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "\n",
    "train_df, test_df = hour_df.randomSplit([0.7, 0.3])\n",
    "train_df.cache()\n",
    "test_df.cache"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import  StringIndexer,  VectorIndexer,VectorAssembler\n",
    "from pyspark.ml.regression import DecisionTreeRegressor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "setup pipeline\n"
     ]
    }
   ],
   "source": [
    "print(\"setup pipeline\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['season', 'mnth', 'hr', 'holiday', 'weekday', 'workingday', 'weathersit', 'temp', 'atemp', 'hum', 'windspeed']\n"
     ]
    }
   ],
   "source": [
    "featuresCols = hour_df.columns[:-1]\n",
    "print(featuresCols) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [],
   "source": [
    "vectorAssembler = VectorAssembler(inputCols=featuresCols, outputCol=\"aFeatures\")\n",
    "vectorIndexer = VectorIndexer(inputCol=\"aFeatures\", outputCol=\"features\", maxCategories=24)\n",
    "dt = DecisionTreeRegressor(labelCol=\"cnt\",featuresCol= 'features',maxDepth=10,impurity=\"variance\",maxBins=100)\n",
    "dt_pipeline = Pipeline(stages=[vectorAssembler,vectorIndexer ,dt])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[VectorAssembler_41929154221977c48ed6,\n",
       " VectorIndexer_49399091be0dea07b623,\n",
       " DecisionTreeRegressor_4e3491e00fd98965b2b6]"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dt_pipeline.getStages()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "train model\n"
     ]
    }
   ],
   "source": [
    "print(\"train model\")\n",
    "dt_pipelineModel = dt_pipeline.fit(train_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DecisionTreeRegressionModel (uid=DecisionTreeRegressor_4e3491e00fd98965b2b6) of depth 10 with 1791 nodes"
      ]
     },
     "execution_count": 59,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dt_pipelineModel.stages[2]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DecisionTreeRegressionModel (uid=DecisionTreeRegressor_4e3491e00fd98965b2b6) of depth 10 with 1791 nodes\n",
      "  If (feature 2 in {0.0,1.0,2.0,3.0,4.0,5.0,6.0,22.0,23.0})\n",
      "   If (feature 2 in {0.0,1.0,2.0,3.0,4.0,5.0})\n",
      "    If (feature 2 in {2.0,3.0,4.0,5.0})\n",
      "     If (feature 4 in {1.0,2.0,3.0,4.0,5.0})\n",
      "      If (feature 2 in {2.0,3.0,4.0})\n",
      "       If (feature 2 in {3.0,4.0})\n",
      "        If (feature 1 in {0.0,1.0,2.0,3.0,11.0})\n",
      "         If (feature 0 in {0.0})\n",
      "          If (feature 7 <= 0.33)\n",
      "           If (feature 1 in {1.0})\n",
      "            Predict: 1.8611111111111112\n",
      "           Else (feature 1 not in {1.0})\n",
      "            Predict: 2.5789473684210527\n",
      "          Else (feature 7 > 0.33)\n",
      "           If (feature 2 in {4.0})\n",
      "            Predict: 2.3333333333333335\n",
      "           Else (feature 2 not in {4.0})\n",
      "            Predict: 4.105263157894737\n",
      "         Else (feature 0 not in {0.0})\n",
      "          If (feature 7 <= 0.47)\n",
      "           If (feature 0 in {1.0})\n",
      "            Predict: 3.403225806451613\n",
      "    \n"
     ]
    }
   ],
   "source": [
    "print(dt_pipelineModel.stages[2].toDebugString[:980])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "predict\n",
      "+------+----+---+-------+-------+----------+----------+----+------+----+---------+----+------------------+\n",
      "|season|mnth| hr|holiday|weekday|workingday|weathersit|temp| atemp| hum|windspeed| cnt|        prediction|\n",
      "+------+----+---+-------+-------+----------+----------+----+------+----+---------+----+------------------+\n",
      "|   1.0| 1.0|0.0|    0.0|    0.0|       0.0|       1.0|0.04|0.0758|0.57|   0.1045|22.0|              15.0|\n",
      "|   1.0| 1.0|0.0|    0.0|    0.0|       0.0|       1.0|0.26| 0.303|0.56|      0.0|39.0|28.764705882352942|\n",
      "|   1.0| 1.0|0.0|    0.0|    0.0|       0.0|       1.0|0.38|0.3939| 0.4|   0.2836|91.0|             56.25|\n",
      "|   1.0| 1.0|0.0|    0.0|    1.0|       1.0|       1.0|0.12|0.1212| 0.5|   0.2836| 5.0|             15.44|\n",
      "|   1.0| 1.0|0.0|    0.0|    1.0|       1.0|       1.0|0.22| 0.197|0.44|   0.3582| 5.0|             15.44|\n",
      "|   1.0| 1.0|0.0|    0.0|    3.0|       1.0|       2.0|0.22|0.2273|0.69|   0.1343|17.0|10.090909090909092|\n",
      "|   1.0| 1.0|0.0|    0.0|    4.0|       1.0|       3.0|0.34|0.3333|0.93|    0.194| 3.0|               8.2|\n",
      "|   1.0| 1.0|0.0|    0.0|    5.0|       1.0|       2.0| 0.2|0.2121|0.75|   0.1343| 9.0|             17.25|\n",
      "|   1.0| 1.0|0.0|    0.0|    5.0|       1.0|       2.0|0.24|0.2273| 0.7|   0.2537|21.0|             17.25|\n",
      "|   1.0| 1.0|0.0|    0.0|    5.0|       1.0|       2.0|0.26|0.2273|0.48|   0.2985|27.0|33.833333333333336|\n",
      "+------+----+---+-------+-------+----------+----------+----+------+----+---------+----+------------------+\n",
      "only showing top 10 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "print(\"predict\")\n",
    "predicted = dt_pipelineModel.transform(test_df).select('season', 'mnth', 'hr', 'holiday', 'weekday', 'workingday', \\\n",
    "                     'weathersit', 'temp', 'atemp', 'hum', 'windspeed','cnt','prediction').show(10)\n",
    "print(predicted)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "eval model\n"
     ]
    }
   ],
   "source": [
    "print(\"eval model\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluator = RegressionEvaluator(labelCol='cnt', predictionCol='prediction', metricName=\"rmse\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "79.97113484730437\n"
     ]
    }
   ],
   "source": [
    "predictions = dt_pipelineModel.transform(test_df)\n",
    "auc = evaluator.evaluate(predictions)\n",
    "print(auc)"
   ]
  },
  {
   "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.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}