logistic_regression_with_spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-03T00:33:35.206861Z",
     "start_time": "2019-02-03T00:33:34.751973Z"
    }
   },
   "outputs": [],
   "source": [
    "import os\n",
    "from datetime import datetime\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession, Row\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator, RegressionEvaluator\n",
    "from pyspark.ml.classification import RandomForestClassifier, GBTClassifier, LogisticRegression\n",
    "from pyspark.ml.feature import StringIndexer, OneHotEncoder,  VectorAssembler, VectorIndexer, IndexToString\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.storagelevel import StorageLevel\n",
    "import pyspark.sql.functions as f\n",
    "from pyspark import SparkContext\n",
    "from pyspark.conf import SparkConf\n",
    "from pyspark.ml import Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-03T00:34:03.750134Z",
     "start_time": "2019-02-03T00:33:35.991919Z"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Unnamed: 0.1.1: integer (nullable = true)\n",
      " |-- _id: struct (nullable = true)\n",
      " |    |-- oid: string (nullable = true)\n",
      " |-- aqi_co: double (nullable = true)\n",
      " |-- aqi_no2: double (nullable = true)\n",
      " |-- aqi_o3: double (nullable = true)\n",
      " |-- aqi_pm10: double (nullable = true)\n",
      " |-- aqi_pm25_frm: double (nullable = true)\n",
      " |-- aqi_pm25_nonfrm: double (nullable = true)\n",
      " |-- aqi_so2: double (nullable = true)\n",
      " |-- arithmetic_mean_co: double (nullable = true)\n",
      " |-- arithmetic_mean_no2: double (nullable = true)\n",
      " |-- arithmetic_mean_o3: double (nullable = true)\n",
      " |-- arithmetic_mean_pm10: double (nullable = true)\n",
      " |-- arithmetic_mean_pm25_frm: double (nullable = true)\n",
      " |-- arithmetic_mean_pm25_nonfrm: double (nullable = true)\n",
      " |-- arithmetic_mean_pm25_speciation: double (nullable = true)\n",
      " |-- arithmetic_mean_pressure: double (nullable = true)\n",
      " |-- arithmetic_mean_so2: double (nullable = true)\n",
      " |-- arithmetic_mean_temp: double (nullable = true)\n",
      " |-- arithmetic_mean_wind: double (nullable = true)\n",
      " |-- city_name: string (nullable = true)\n",
      " |-- county_code: integer (nullable = true)\n",
      " |-- day: integer (nullable = true)\n",
      " |-- dow: integer (nullable = true)\n",
      " |-- first_max_value_co: double (nullable = true)\n",
      " |-- first_max_value_no2: double (nullable = true)\n",
      " |-- first_max_value_o3: double (nullable = true)\n",
      " |-- first_max_value_pm10: double (nullable = true)\n",
      " |-- first_max_value_pm25_frm: double (nullable = true)\n",
      " |-- first_max_value_pm25_nonfrm: double (nullable = true)\n",
      " |-- first_max_value_pm25_speciation: double (nullable = true)\n",
      " |-- first_max_value_pressure: double (nullable = true)\n",
      " |-- first_max_value_so2: double (nullable = true)\n",
      " |-- first_max_value_temp: double (nullable = true)\n",
      " |-- first_max_value_wind: double (nullable = true)\n",
      " |-- label: string (nullable = true)\n",
      " |-- latitude: double (nullable = true)\n",
      " |-- longitude: double (nullable = true)\n",
      " |-- max_aqi: double (nullable = true)\n",
      " |-- max_aqi_before_yesterday: double (nullable = true)\n",
      " |-- max_aqi_yesterday: double (nullable = true)\n",
      " |-- month: integer (nullable = true)\n",
      " |-- observation_count_co: double (nullable = true)\n",
      " |-- observation_count_no2: double (nullable = true)\n",
      " |-- observation_count_o3: double (nullable = true)\n",
      " |-- observation_count_pm10: double (nullable = true)\n",
      " |-- observation_count_pm25_frm: double (nullable = true)\n",
      " |-- observation_count_pm25_nonfrm: double (nullable = true)\n",
      " |-- observation_count_pm25_speciation: double (nullable = true)\n",
      " |-- observation_count_pressure: double (nullable = true)\n",
      " |-- observation_count_so2: double (nullable = true)\n",
      " |-- observation_count_temp: double (nullable = true)\n",
      " |-- observation_count_wind: double (nullable = true)\n",
      " |-- site_num: integer (nullable = true)\n",
      " |-- year: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pyspark_submit_args = '--packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.0 pyspark-shell'\n",
    "os.environ[\"PYSPARK_SUBMIT_ARGS\"] = pyspark_submit_args\n",
    "\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"aqi\") \\\n",
    "    .config(\"spark.mongodb.input.uri\", \"mongodb://34.221.229.103/mydb.air\")\\\n",
    "    .getOrCreate()\n",
    "spark.conf.set(\"spark.executor.memory\", '4g')\n",
    "spark.conf.set('spark.executor.cores', '4')\n",
    "spark.conf.set('spark.cores.max', '4')\n",
    "spark.conf.set(\"spark.driver.memory\",'4g')\n",
    "df = spark.read.format(\"com.mongodb.spark.sql.DefaultSource\").load()\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-03T00:34:03.776328Z",
     "start_time": "2019-02-03T00:34:03.752488Z"
    }
   },
   "outputs": [],
   "source": [
    "va = VectorAssembler(outputCol=\"features\", inputCols=[\n",
    " 'aqi_co',\n",
    " 'aqi_no2',\n",
    " 'aqi_o3',\n",
    " 'aqi_pm10',\n",
    " 'aqi_pm25_frm',\n",
    " 'aqi_pm25_nonfrm',\n",
    " 'aqi_so2',\n",
    " 'arithmetic_mean_co',\n",
    " 'arithmetic_mean_no2',\n",
    " 'arithmetic_mean_o3',\n",
    " 'arithmetic_mean_pm10',\n",
    " 'arithmetic_mean_pm25_frm',\n",
    " 'arithmetic_mean_pm25_nonfrm',\n",
    " 'arithmetic_mean_pm25_speciation',\n",
    " 'arithmetic_mean_pressure',\n",
    " 'arithmetic_mean_so2',\n",
    " 'arithmetic_mean_temp',\n",
    " 'arithmetic_mean_wind',\n",
    " 'county_code',\n",
    " 'day',\n",
    " 'dow',\n",
    " 'first_max_value_co',\n",
    " 'first_max_value_no2',\n",
    " 'first_max_value_o3',\n",
    " 'first_max_value_pm10',\n",
    " 'first_max_value_pm25_frm',\n",
    " 'first_max_value_pm25_nonfrm',\n",
    " 'first_max_value_pm25_speciation',\n",
    " 'first_max_value_pressure',\n",
    " 'first_max_value_so2',\n",
    " 'first_max_value_temp',\n",
    " 'first_max_value_wind',\n",
    " 'latitude',\n",
    " 'longitude',\n",
    " 'max_aqi',\n",
    " 'max_aqi_before_yesterday',\n",
    " 'max_aqi_yesterday',\n",
    " 'month',\n",
    " 'observation_count_co',\n",
    " 'observation_count_no2',\n",
    " 'observation_count_o3',\n",
    " 'observation_count_pm10',\n",
    " 'observation_count_pm25_frm',\n",
    " 'observation_count_pm25_nonfrm',\n",
    " 'observation_count_pm25_speciation',\n",
    " 'observation_count_pressure',\n",
    " 'observation_count_so2',\n",
    " 'observation_count_temp',\n",
    " 'observation_count_wind',\n",
    " 'site_num']) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-03T00:34:04.165063Z",
     "start_time": "2019-02-03T00:34:03.777674Z"
    }
   },
   "outputs": [],
   "source": [
    "df2 = va.transform(df).select(\"features\", \"label\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-03T00:34:14.165610Z",
     "start_time": "2019-02-03T00:34:04.167300Z"
    }
   },
   "outputs": [],
   "source": [
    "# Load and parse the data file, converting it to a DataFrame.\n",
    "data = df2\n",
    "\n",
    "# Index labels, adding metadata to the label column.\n",
    "# Fit on whole dataset to include all labels in index.\n",
    "labelIndexer = StringIndexer(inputCol=\"label\", outputCol=\"indexedLabel\").fit(data)\n",
    "dataIndexed = labelIndexer.transform(data)\n",
    "\n",
    "data = dataIndexed.drop('label').withColumnRenamed('indexedLabel','label')\n",
    "# Automatically identify categorical features, and index them.\n",
    "# Set maxCategories so features with > 4 distinct values are treated as continuous.\n",
    "#featureIndexer =\\\n",
    "#    VectorIndexer(inputCol=\"features\", outputCol=\"indexedFeatures\").fit(data)\n",
    "\n",
    "# Split the data into training and test sets (30% held out for testing)\n",
    "(trainingData, testData) = data.randomSplit([0.7, 0.3])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2019-02-03T00:45:59.570359Z",
     "start_time": "2019-02-03T00:34:14.167371Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+-----+--------------------+\n",
      "|predictedLabel|label|            features|\n",
      "+--------------+-----+--------------------+\n",
      "|          good|  0.0|[1.0,8.0,19.0,19....|\n",
      "|          good|  0.0|[1.0,8.0,28.0,11....|\n",
      "|          good|  0.0|[1.0,9.0,19.0,9.0...|\n",
      "|          good|  0.0|[1.0,9.0,33.0,19....|\n",
      "|          good|  1.0|[1.0,10.0,30.0,19...|\n",
      "+--------------+-----+--------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Test Error = 0.285293\n",
      "LogisticRegression_4df29b5663d8161e56fb\n",
      "CPU times: user 89.4 ms, sys: 58.1 ms, total: 147 ms\n",
      "Wall time: 11min 45s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)\n",
    "\n",
    "# Convert indexed labels back to original labels.\n",
    "labelConverter = IndexToString(inputCol=\"prediction\", outputCol=\"predictedLabel\",\n",
    "                               labels=labelIndexer.labels)\n",
    "\n",
    "# Chain indexers and forest in a Pipeline\n",
    "pipeline = Pipeline(stages=[lr, labelConverter])\n",
    "\n",
    "# Fit the model\n",
    "# Train model.  This also runs the indexers.\n",
    "model = pipeline.fit(trainingData)\n",
    "# Make predictions.\n",
    "predictions = model.transform(testData)\n",
    "\n",
    "# Select example rows to display.\n",
    "predictions.select(\"predictedLabel\", \"label\", \"features\").show(5)\n",
    "\n",
    "# Select (prediction, true label) and compute test error\n",
    "evaluator = MulticlassClassificationEvaluator(labelCol=\"label\", predictionCol=\"prediction\", metricName=\"accuracy\")\n",
    "accuracy = evaluator.evaluate(predictions)\n",
    "print(\"Test Error = %g\" % (1.0 - accuracy))\n",
    "\n",
    "lrModel = model.stages[0]\n",
    "print(lrModel)  # summary only"
   ]
  },
  {
   "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.6.7"
  },
  "latex_envs": {
   "LaTeX_envs_menu_present": true,
   "autoclose": false,
   "autocomplete": true,
   "bibliofile": "biblio.bib",
   "cite_by": "apalike",
   "current_citInitial": 1,
   "eqLabelWithNumbers": true,
   "eqNumInitial": 1,
   "hotkeys": {
    "equation": "Ctrl-E",
    "itemize": "Ctrl-I"
   },
   "labels_anchors": false,
   "latex_user_defs": false,
   "report_style_numbering": false,
   "user_envs_cfg": false
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}