H2O_AutoML_IPYNB/ChicagoCrimeDemo.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "#Initiate H2OContext on top of Spark\n",
    "from pysparkling import *\n",
    "hc = H2OContext.getOrCreate(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# This is just helper function returning relative path to data files within sparkling-water project directories\n",
    "def _locate(example_name): \n",
    "    return \"examples/smalldata/\" + example_name \n",
    "\n",
    "\n",
    "# Define file names\n",
    "chicagoAllWeather = \"chicagoAllWeather.csv\"\n",
    "chicagoCensus = \"chicagoCensus.csv\"\n",
    "chicagoCrimes10k = \"chicagoCrimes10k.csv\"\n",
    "\n",
    "# Add files to Spark Cluster\n",
    "sc.addFile(_locate(chicagoAllWeather))\n",
    "sc.addFile(_locate(chicagoCensus))\n",
    "sc.addFile(_locate(chicagoCrimes10k))\n",
    "\n",
    "# And import them into H2O\n",
    "from pyspark import SparkFiles\n",
    "import h2o\n",
    "# Since we have already loaded files into spark, we have to use h2o.upload_file instead of \n",
    "# h2o.import_file since h2o.import_file expects cluster-relative path (ie. the file on this\n",
    "# path can be accessed from all the machines on the cluster) but SparkFiles.get(..) already\n",
    "# give us relative path to the file on a current node which h2o.upload_file can handle ( it\n",
    "# uploads file located on current node and distributes it to the H2O cluster)\n",
    "f_weather = h2o.upload_file(SparkFiles.get(chicagoAllWeather))\n",
    "f_census = h2o.upload_file(SparkFiles.get(chicagoCensus))\n",
    "f_crimes = h2o.upload_file(SparkFiles.get(chicagoCrimes10k), col_types = {\"Date\": \"string\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "f_weather.show()\n",
    "f_census.show()\n",
    "f_crimes.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Set time zone to UTC for date manipulation\n",
    "h2o.set_timezone(\"Etc/UTC\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Transform weather table\n",
    "## Remove 1st column (date)\n",
    "f_weather = f_weather[1:]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Transform census table\n",
    "## Remove all spaces from column names (causing problems in Spark SQL)\n",
    "col_names = map(lambda s: s.strip().replace(' ', '_').replace('+','_'), f_census.col_names)\n",
    "\n",
    "## Update column names in the table\n",
    "#f_weather.names = col_names\n",
    "f_census.names = col_names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Transform crimes table\n",
    "\n",
    "## Drop useless columns\n",
    "f_crimes = f_crimes[2:]\n",
    "\n",
    "## Replace ' ' by '_' in column names\n",
    "col_names = map(lambda s: s.replace(' ', '_'), f_crimes.col_names)\n",
    "f_crimes.names = col_names\n",
    "\n",
    "## Refine date column\n",
    "def refine_date_col(data, col, pattern):\n",
    "    data[col]         = data[col].as_date(pattern)\n",
    "    data[\"Day\"]       = data[col].day()\n",
    "    data[\"Month\"]     = data[col].month()\n",
    "    data[\"Year\"]      = data[col].year()\n",
    "    data[\"WeekNum\"]   = data[col].week()\n",
    "    data[\"WeekDay\"]   = data[col].dayOfWeek()\n",
    "    data[\"HourOfDay\"] = data[col].hour()\n",
    "    \n",
    "    data.describe() # HACK: Force evaluation before ifelse and cut. See PUBDEV-1425.\n",
    "        \n",
    "    # Create weekend and season cols\n",
    "    data[\"Weekend\"] = ((data[\"WeekDay\"] == \"Sun\") | (data[\"WeekDay\"] == \"Sat\"))\n",
    "    data[\"Season\"] = data[\"Month\"].cut([0, 2, 5, 7, 10, 12], [\"Winter\", \"Spring\", \"Summer\", \"Autumn\", \"Winter\"])\n",
    "    \n",
    "refine_date_col(f_crimes, \"Date\", \"%m/%d/%Y %I:%M:%S %p\")\n",
    "f_crimes = f_crimes.drop(\"Date\")\n",
    "f_crimes.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Expose H2O frames as Spark DataFrame\n",
    "\n",
    "df_weather = hc.as_spark_frame(f_weather)\n",
    "df_census = hc.as_spark_frame(f_census)\n",
    "df_crimes = hc.as_spark_frame(f_crimes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Use Spark SQL to join datasets\n",
    "\n",
    "## Register DataFrames as tables in SQL context\n",
    "sqlContext.registerDataFrameAsTable(df_weather, \"chicagoWeather\")\n",
    "sqlContext.registerDataFrameAsTable(df_census, \"chicagoCensus\")\n",
    "sqlContext.registerDataFrameAsTable(df_crimes, \"chicagoCrime\")\n",
    "\n",
    "\n",
    "crimeWithWeather = sqlContext.sql(\"\"\"SELECT\n",
    "a.Year, a.Month, a.Day, a.WeekNum, a.HourOfDay, a.Weekend, a.Season, a.WeekDay,\n",
    "a.IUCR, a.Primary_Type, a.Location_Description, a.Community_Area, a.District,\n",
    "a.Arrest, a.Domestic, a.Beat, a.Ward, a.FBI_Code,\n",
    "b.minTemp, b.maxTemp, b.meanTemp,\n",
    "c.PERCENT_AGED_UNDER_18_OR_OVER_64, c.PER_CAPITA_INCOME, c.HARDSHIP_INDEX,\n",
    "c.PERCENT_OF_HOUSING_CROWDED, c.PERCENT_HOUSEHOLDS_BELOW_POVERTY,\n",
    "c.PERCENT_AGED_16__UNEMPLOYED, c.PERCENT_AGED_25__WITHOUT_HIGH_SCHOOL_DIPLOMA\n",
    "FROM chicagoCrime a\n",
    "JOIN chicagoWeather b\n",
    "ON a.Year = b.year AND a.Month = b.month AND a.Day = b.day\n",
    "JOIN chicagoCensus c\n",
    "ON a.Community_Area = c.Community_Area_Number\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "crimeWithWeather.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Publish Spark DataFrame as H2OFrame with given name\n",
    "crimeWithWeatherHF = hc.as_h2o_frame(crimeWithWeather, \"crimeWithWeatherTable\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Transform selected String columns to categoricals\n",
    "crimeWithWeatherHF[\"Arrest\"] = crimeWithWeatherHF[\"Arrest\"].asfactor()\n",
    "crimeWithWeatherHF[\"Season\"] = crimeWithWeatherHF[\"Season\"].asfactor()\n",
    "crimeWithWeatherHF[\"WeekDay\"] = crimeWithWeatherHF[\"WeekDay\"].asfactor()\n",
    "crimeWithWeatherHF[\"Primary_Type\"] = crimeWithWeatherHF[\"Primary_Type\"].asfactor()\n",
    "crimeWithWeatherHF[\"Location_Description\"] = crimeWithWeatherHF[\"Location_Description\"].asfactor()\n",
    "crimeWithWeatherHF[\"Domestic\"] = crimeWithWeatherHF[\"Domestic\"].asfactor()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Split frame into two - we use one as the training frame and the second one as the validation frame\n",
    "splits = crimeWithWeatherHF.split_frame(ratios=[0.8])\n",
    "train = splits[0]\n",
    "test = splits[1]\n",
    "\n",
    "# Prepare column names\n",
    "predictor_columns = train.drop(\"Arrest\").col_names\n",
    "response_column = \"Arrest\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Create and train GBM model\n",
    "from h2o.estimators.gbm import H2OGradientBoostingEstimator\n",
    "\n",
    "# Prepare model based on the given set of parameters\n",
    "gbm_model = H2OGradientBoostingEstimator(ntrees       = 50,\n",
    "                                         max_depth    = 3,\n",
    "                                         distribution = \"bernoulli\")\n",
    "\n",
    "# Train the model\n",
    "gbm_model.train(x                = predictor_columns,\n",
    "                y                = response_column,\n",
    "                training_frame   = train,\n",
    "                validation_frame = test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Show GBM model performance\n",
    "gbm_model.model_performance(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Create and train deeplearning model\n",
    "from h2o.estimators.deeplearning import H2ODeepLearningEstimator\n",
    "\n",
    "# Prepare model based on the given set of parameters\n",
    "dl_model = H2ODeepLearningEstimator(variable_importances = True)\n",
    "\n",
    "# Train the model\n",
    "dl_model.train(x                = predictor_columns,\n",
    "               y                = response_column,\n",
    "               training_frame   = train,\n",
    "               validation_frame = test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Show deeplearning model performance\n",
    "dl_model.model_performance(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Create crime class which is used as a data holder on which prediction is done\n",
    "from datetime import datetime\n",
    "from pytz import timezone\n",
    "from pyspark.sql import Row\n",
    "\n",
    "def get_season(dt):\n",
    "    if (dt >= 3 and dt <= 5):\n",
    "        return \"Spring\"\n",
    "    elif (dt >= 6 and dt <= 8):\n",
    "        return \"Summer\"\n",
    "    elif (dt >= 9 and dt <= 10):\n",
    "        return \"Autumn\"\n",
    "    else:       \n",
    "        return \"Winter\"\n",
    "    \n",
    "def crime(date,\n",
    "          iucr,\n",
    "          primaryType,\n",
    "          locationDescr,\n",
    "          domestic,\n",
    "          beat,\n",
    "          district,\n",
    "          ward,\n",
    "          communityArea,\n",
    "          fbiCode,\n",
    "          datePattern = \"%m/%d/%Y %I:%M:%S %p\",\n",
    "          dateTimeZone = \"Etc/UTC\"):\n",
    "\n",
    "    dt = datetime.strptime(date, datePattern)\n",
    "    dt.replace(tzinfo=timezone(dateTimeZone))\n",
    "\n",
    "    crime = Row(\n",
    "        Year = dt.year,\n",
    "        Month = dt.month,\n",
    "        Day = dt.day,\n",
    "        WeekNum = dt.isocalendar()[1],\n",
    "        HourOfDay = dt.hour,\n",
    "        Weekend = 1 if dt.weekday() == 5 or dt.weekday() == 6 else 0,\n",
    "        Season = get_season(dt.month),\n",
    "        WeekDay = dt.strftime('%a'),  #gets the day of week in short format - Mon, Tue ...\n",
    "        IUCR = iucr,\n",
    "        Primary_Type = primaryType,\n",
    "        Location_Description = locationDescr,\n",
    "        Domestic = 'true' if domestic else 'false',\n",
    "        Beat = beat,\n",
    "        District = district,\n",
    "        Ward = ward,\n",
    "        Community_Area = communityArea,\n",
    "        FBI_Code = fbiCode\n",
    "    )\n",
    "    return crime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# Create crime examples\n",
    "crime_examples = [\n",
    "  crime(\"02/08/2015 11:43:58 PM\", 1811, \"NARCOTICS\", \"STREET\",False, 422, 4, 7, 46, 18),\n",
    "  crime(\"02/08/2015 11:00:39 PM\", 1150, \"DECEPTIVE PRACTICE\", \"RESIDENCE\",False, 923, 9, 14, 63, 11)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# For given crime and model return probability of crime.\n",
    "def score_event(crime, model, censusTable):\n",
    "    rdd = sc.parallelize([crime])\n",
    "    crime_frame = sqlContext.createDataFrame(rdd)\n",
    "    # Join table with census data\n",
    "    df_row = censusTable.join(crime_frame).where(\"Community_Area = Community_Area_Number\")  \n",
    "    row = hc.as_h2o_frame(df_row)\n",
    "    row[\"Season\"] = row[\"Season\"].asfactor()\n",
    "    row[\"WeekDay\"] = row[\"WeekDay\"].asfactor()\n",
    "    row[\"Primary_Type\"] = row[\"Primary_Type\"].asfactor()\n",
    "    row[\"Location_Description\"] = row[\"Location_Description\"].asfactor()\n",
    "    row[\"Domestic\"] = row[\"Domestic\"].asfactor()\n",
    "\n",
    "    predictTable = model.predict(row)\n",
    "    probOfArrest = predictTable[\"true\"][0,0]\n",
    "    return probOfArrest\n",
    "\n",
    "for crime in crime_examples:\n",
    "    arrestProbGBM = 100*score_event(crime, gbm_model, df_census)\n",
    "    arrestProbDLM = 100*score_event(crime, dl_model, df_census)\n",
    "\n",
    "    print(\"\"\"\n",
    "       |Crime: \"\"\"+str(crime)+\"\"\"\n",
    "       |  Probability of arrest best on DeepLearning: \"\"\"+str(arrestProbDLM)+\"\"\"\n",
    "       |  Probability of arrest best on GBM: \"\"\"+str(arrestProbGBM)+\"\"\"\n",
    "        \"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "# stop H2O and Spark services\n",
    "h2o.shutdown(prompt = False)\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [python2]",
   "language": "python",
   "name": "Python [python2]"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}