full_dataset_execution.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import urllib\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "import numpy as np\n",
    "import pyspark.sql.functions as f\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import lit\n",
    "from pyspark.sql import Row\n",
    "from functools import reduce\n",
    "from pyspark.sql import DataFrame\n",
    "import matplotlib.pyplot as plt\n",
    "ACCESS_KEY = \"XXXXXXXXXXXXXXXXXXXX\"\n",
    "SECRET_KEY = \"XXXXXXXXXXXXXXXXXXXX\"\n",
    "ENCODED_SECRET_KEY = urllib.quote(SECRET_KEY, \"\")\n",
    "AWS_BUCKET_NAME = \"telematicsdata\"\n",
    "MOUNT_NAME = \"telefonica\"\n",
    "#dbutils.fs.mount(\"s3n://%s:%s@%s\" % (ACCESS_KEY, ENCODED_SECRET_KEY, AWS_BUCKET_NAME), \"/mnt/%s\" % MOUNT_NAME)\n",
    "sc=SparkContext.getOrCreate()\n",
    "sqlContext = SQLContext.getOrCreate(sc)\n",
    "\n",
    "def preparePlot(xticks, yticks, figsize=(10.5, 6), hideLabels=False, gridColor='#999999',\n",
    "                gridWidth=1.0):\n",
    "    \"\"\"Template for generating the plot layout.\"\"\"\n",
    "    plt.close()\n",
    "    fig, ax = plt.subplots(figsize=figsize, facecolor='white', edgecolor='white')\n",
    "    ax.axes.tick_params(labelcolor='#999999', labelsize='10')\n",
    "    for axis, ticks in [(ax.get_xaxis(), xticks), (ax.get_yaxis(), yticks)]:\n",
    "        axis.set_ticks_position('none')\n",
    "        axis.set_ticks(ticks)\n",
    "        axis.label.set_color('#999999')\n",
    "        if hideLabels: axis.set_ticklabels([])\n",
    "    plt.grid(color=gridColor, linewidth=gridWidth, linestyle='-')\n",
    "    map(lambda position: ax.spines[position].set_visible(False), ['bottom', 'top', 'left', 'right'])\n",
    "    return fig, ax\n",
    "pass\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "display(dbutils.fs.ls(\"/mnt/\"+MOUNT_NAME+\"/data/1\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "DATA_BASE_DIR=\"/mnt/\"+MOUNT_NAME+\"/data/\"\n",
    "trip_df_list=[]\n",
    "#cnt=0\n",
    "for d in dbutils.fs.ls(DATA_BASE_DIR):\n",
    "  driver=int(d.path.split(DATA_BASE_DIR,1)[1][:-1])\n",
    "  for f in dbutils.fs.ls(DATA_BASE_DIR+str(driver)+\"/\"):\n",
    "    trip=int(f.path.split(DATA_BASE_DIR+str(driver)+\"/\",1)[1][:-4])\n",
    "    df = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load(DATA_BASE_DIR+str(driver)+\"/\"+str(trip)+\".csv\")\n",
    "    schema  = StructType(df.schema.fields[:] + [StructField(\"t\", IntegerType(), False),StructField(\"driver\", IntegerType(), True),StructField(\"trip\", IntegerType(), True),StructField(\"driver_trip\", StringType(), True),StructField(\"x_y\", ArrayType(DoubleType()), True)])\n",
    "    df = (df.rdd # Extract rdd\n",
    "      .zipWithIndex() # Add index\n",
    "      .map(lambda ri: Row(*list(ri[0]) + [ri[1],driver,trip,str(driver)+\"_\"+str(trip),[ri[0].x,ri[0].y]])) # Map to rows\n",
    "      .toDF(schema))\n",
    "    trip_df_list.append(df)\n",
    "    \"\"\"if cnt==0:\n",
    "      df.saveAsTable(\"ALLCSVS\")\n",
    "    else:\n",
    "      df.write.insertInto(\"ALLCSVS\", overwrite=False)\n",
    "    cnt=cnt+1\"\"\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def unionAll(dfs):\n",
    "    return reduce(DataFrame.unionAll, dfs)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "print(len(DF_list))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "DF_list=[]\n",
    "for i in range(len(trip_df_list)/1000):\n",
    "  DF_list.append(unionAll(trip_df_list[i*1000:((i+1)*1000)]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "DF=unionAll(DF_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "DF=DF.repartition(4,[\"driver\"])\n",
    "DF=DF.sortWithinPartitions([\"driver\",\"trip\",'t'],ascending=True)\n",
    "DF.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "DF.write.saveAsTable(\"ALLCSVS\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "DF=sqlContext.sql(\"SELECT * FROM ALLCSVS\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "display(DF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\n",
    "import pyspark.sql.functions as f\n",
    "counts=DF.groupBy(\"driver\").agg(f.countDistinct(DF.trip))\n",
    "display(counts)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "display(counts[counts['count(trip)']<200])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import Row\n",
    "from pyspark.sql.functions import struct\n",
    "from pyspark.sql import DataFrame\n",
    "from collections import OrderedDict\n",
    "\n",
    "def reduce_by(self, by, cols, f, schema=None):\n",
    "    \"\"\"\n",
    "    :param self DataFrame\n",
    "    :param by a list of grouping columns \n",
    "    :param cols a list of columns to aggregate\n",
    "    :param aggregation function Row => Row\n",
    "    :return DataFrame\n",
    "    \"\"\"\n",
    "    def merge_kv(kv):\n",
    "        key, value = kv\n",
    "        return Row(**OrderedDict(zip(\n",
    "            key.__fields__ + value.__fields__, key + value)\n",
    "        ))\n",
    "\n",
    "    return (self\n",
    "        .select(struct(*by), struct(*cols))\n",
    "        .rdd\n",
    "        .reduceByKey(f)\n",
    "        .map(merge_kv)\n",
    "        .toDF(schema))\n",
    "\n",
    "DataFrame.reduce_by = reduce_by"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "def foo(row1, row2):\n",
    "    \"\"\" A dummy function\n",
    "    >>> foo(Row(x=1, y=None), Row(x=None, y=2))\n",
    "    Row(x=1, y=2)\n",
    "    \"\"\"\n",
    "    return Row(**OrderedDict(zip(\n",
    "      row1.__fields__, (str(x)+\",\"+str(y) for (x, y) in zip(row1, row2))\n",
    "    )))\n",
    "field = [StructField(\"driver\", IntegerType(), True),StructField(\"trip\", IntegerType(), True),StructField(\"trip_array\", StringType(), True)]\n",
    "schema = StructType(field)\n",
    "\n",
    "grouped=DF.reduce_by(by=[\"driver\",\"trip\"], cols=[\"x_y\"], f=foo,schema=schema)\n",
    "\n",
    "def makeMatrix(x):\n",
    "  return eval(\"[\"+x+\"]\")\n",
    "\n",
    "mMat=f.udf(lambda x : makeMatrix(x),ArrayType(ArrayType(DoubleType())))\n",
    "grouped=grouped.withColumn(\"trip_array\",mMat(grouped.trip_array))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "grouped.write.format(\"com.databricks.spark.avro\").save(\"/mnt/all-output\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "grouped=sqlContext.read.format(\"com.databricks.spark.avro\").load(\"/mnt/all-output\")\n",
    "display(grouped)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Calculate Distances \n",
    "from scipy.spatial import distance\n",
    "\n",
    "def calcDistance(trip_array):\n",
    "  distances=[]\n",
    "  for i,x_y in enumerate(trip_array):\n",
    "    if i>0:\n",
    "      distances.append(distance.euclidean(trip_array[i-1],trip_array[i]))\n",
    "  return distances\n",
    "\n",
    "get_distances=f.udf(calcDistance,ArrayType(DoubleType()))\n",
    "get_abs_distance=f.udf(lambda arr:distance.euclidean(arr[0],arr[len(arr)-1]),DoubleType())\n",
    "grouped=grouped.withColumn(\"trip_distances\",get_distances(grouped.trip_array))\n",
    "grouped=grouped.withColumn(\"trip_absolute_distance\",get_abs_distance(grouped.trip_array))\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.trip_distances))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Smoothed Speed (with MA)\n",
    "def running_mean(x,N=10):\n",
    "    cumsum = np.cumsum(np.insert(x, 0, 0)) \n",
    "    return list(map(float,(cumsum[N:] - cumsum[:-N]) / N))\n",
    "\n",
    "calc_smoothed_speed = f.udf(lambda x:running_mean(x),ArrayType(FloatType()))\n",
    "\n",
    "grouped=grouped.withColumn(\"smoothed_speed\",calc_smoothed_speed(grouped.trip_distances))\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.smoothed_speed))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Plot speed & smoothed speed\n",
    "import matplotlib.pyplot as plt\n",
    "cols=grouped.select(grouped.driver,grouped.trip,grouped.trip_distances.alias(\"speed_per_sec\"),grouped.smoothed_speed).take(1)[0]\n",
    "speed_per_sec=cols.speed_per_sec\n",
    "smoothed_speed=list(np.zeros(9))+cols.smoothed_speed\n",
    "t=range(len(speed_per_sec))\n",
    "driver=str(cols.driver)\n",
    "trip=str(cols.trip)\n",
    "fig, ax = plt.subplots()\n",
    "fig.set_figheight(5)\n",
    "ax.plot(t, speed_per_sec,label=\"speed per second\")\n",
    "ax.plot(t, smoothed_speed,label=\"smoothed speed\")\n",
    "ax.set_title('Speed & 10 periods MA Smoothed Speed vs. time for driver '+driver+\" trip \"+trip)\n",
    "ax.set_xlabel('time in seconds')\n",
    "ax.set_ylabel('speed (m/s)')\n",
    "ax.legend()\n",
    "display(fig)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Calculate Durations per Trip\n",
    "\n",
    "get_length = f.udf(lambda x:len(x),IntegerType())\n",
    "grouped=grouped.withColumn(\"trip_duration\",get_length(grouped.trip_array))\n",
    "#display(grouped.select(grouped.driver,grouped.trip,grouped.trip_duration))\n",
    "#STOPs\n",
    "def stops(bits):   \n",
    "  # make sure all runs of ones are well-bounded\n",
    "  bounded = np.hstack(([1], bits, [1]))\n",
    "  log = (bounded<0+0.5)*1\n",
    "  # get 1 at run starts and -1 at run ends\n",
    "  diffs = np.diff(log)    \n",
    "\n",
    "  # get indices if starts and ends\n",
    "  run_starts = np.where(diffs > 0)[0]\n",
    "  run_ends = np.where(diffs < 0)[0]\n",
    "  return np.array([run_starts,run_ends,run_ends-run_starts]).T.tolist()\n",
    "get_info_array = f.udf(stops,ArrayType(ArrayType(IntegerType())))\n",
    "\n",
    "get_stop_duration = f.udf(lambda x:sum([i[2] for i in x]),IntegerType())\n",
    "grouped=grouped.withColumn(\"stop_info_array\",get_info_array(grouped.smoothed_speed))\n",
    "grouped=grouped.withColumn(\"stop_no\",get_length(grouped.stop_info_array))\n",
    "grouped=grouped.withColumn(\"total_stop_duration\",get_stop_duration(grouped.stop_info_array))\n",
    "grouped=grouped.withColumn(\"stop_ratio\",grouped.total_stop_duration/grouped.trip_duration)\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.stop_no,grouped.total_stop_duration,grouped.trip_duration,grouped.stop_ratio))\n",
    "  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Acceleration\n",
    "def get_accel(x):\n",
    "    return list(map(float,(np.diff(x))))\n",
    "def get_neg_accel(accel_s):\n",
    "  accel_s=np.array(accel_s)\n",
    "  return list(map(float,(accel_s[accel_s<0])))\n",
    "def get_pos_accel(accel_s):\n",
    "  accel_s=np.array(accel_s)\n",
    "  return list(map(float,(accel_s[accel_s>0])))\n",
    "from scipy.signal import savgol_filter\n",
    "def get_circular_acceleration(ride):\n",
    "  ride = np.array(ride)\n",
    "  ride = savgol_filter(ride.T, 7, 3).T\n",
    "\n",
    "  # http://stackoverflow.com/questions/28269379/curve-curvature-in-numpy\n",
    "  dx_dt = np.gradient(ride[:, 0])\n",
    "  dy_dt = np.gradient(ride[:, 1])\n",
    "  velocity = np.vstack((dx_dt, dy_dt)).T\n",
    "  ds_dt = np.linalg.norm(velocity, axis=1)\n",
    "  np.seterr(all='ignore')\n",
    "  tangent = np.array([1/ds_dt] * 2).T\n",
    "  np.seterr(all='print')\n",
    "  tangent = np.nan_to_num(tangent)\n",
    "  tangent = tangent * velocity\n",
    "  tangent_x = tangent[:, 0]\n",
    "  tangent_y = tangent[:, 1]\n",
    "\n",
    "  deriv_tangent_x = np.gradient(tangent_x)\n",
    "  deriv_tangent_y = np.gradient(tangent_y)\n",
    "  dT_dt = np.vstack((deriv_tangent_x, deriv_tangent_y)).T\n",
    "  length_dT_dt = np.linalg.norm(dT_dt, axis=1)\n",
    "\n",
    "  np.seterr(all='ignore')\n",
    "  normal = np.array([1/length_dT_dt] * 2).T\n",
    "  np.seterr(all='print')\n",
    "  normal = np.nan_to_num(normal)\n",
    "  normal = normal * dT_dt\n",
    "  d2s_dt2 = np.gradient(ds_dt)\n",
    "  d2x_dt2 = np.gradient(dx_dt)\n",
    "  d2y_dt2 = np.gradient(dy_dt)\n",
    "\n",
    "  np.seterr(all='ignore')\n",
    "  curvature = np.abs(d2x_dt2 * dy_dt - dx_dt * d2y_dt2) / (dx_dt * dx_dt + dy_dt * dy_dt)**1.5\n",
    "  np.seterr(all='print')\n",
    "  curvature = np.nan_to_num(curvature)\n",
    "\n",
    "  t_comp = d2s_dt2\n",
    "  n_comp = curvature * ds_dt * ds_dt\n",
    "  t_component = np.array([t_comp] * 2).T\n",
    "  n_component = np.array([n_comp] * 2).T\n",
    "  \n",
    "  #acceleration = [float(np.linalg.norm(v,2)) for v in (t_component * tangent + n_component * normal)]\n",
    "  #Calculating magnitude of the acceleration vectors!!!\n",
    "  acceleration = [float(np.linalg.norm(v)) for v in (t_component * tangent + n_component * normal)]\n",
    "  return acceleration\n",
    "calc_acceleration = f.udf(get_accel,ArrayType(FloatType()))\n",
    "calc_neg_acceleration = f.udf(get_neg_accel,ArrayType(FloatType()))\n",
    "calc_pos_acceleration = f.udf(get_pos_accel,ArrayType(FloatType()))\n",
    "calc_circular_acceleration = f.udf(get_circular_acceleration,ArrayType(FloatType()))\n",
    "grouped=grouped.withColumn(\"accelerations\",calc_acceleration(grouped.smoothed_speed))\n",
    "grouped=grouped.withColumn(\"pos_accelerations\",calc_pos_acceleration(grouped.accelerations))\n",
    "grouped=grouped.withColumn(\"neg_accelerations\",calc_neg_acceleration(grouped.accelerations))\n",
    "grouped=grouped.withColumn(\"circular_accelerations\",calc_circular_acceleration(grouped.trip_array))\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.circular_accelerations))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Speed Descriptive Statistics\n",
    "import numpy as np\n",
    "\n",
    "avg=udf(lambda xs: float(np.mean(xs)) if len(xs)>0 else 0, FloatType())\n",
    "median=udf(lambda xs: float(np.median(xs)) if len(xs)>0 else 0, FloatType())\n",
    "max_udf = udf(lambda xs: float(np.max(xs))if len(xs)>0 else 0, FloatType())\n",
    "min_udf = udf(lambda xs: float(np.min(xs))if len(xs)>0 else 0, FloatType())\n",
    "std_udf = udf(lambda xs: float(np.std(xs))if len(xs)>0 else 0, FloatType())\n",
    "calc_total_trip_length = udf(lambda xs: float(np.sum(xs)), FloatType())\n",
    "\n",
    "grouped=grouped.select(grouped.driver,grouped.trip,grouped.trip_array,grouped.trip_distances\n",
    "                       ,grouped.accelerations,grouped.pos_accelerations,grouped.neg_accelerations,grouped.circular_accelerations\n",
    "                       ,grouped.smoothed_speed,grouped.trip_absolute_distance,grouped.trip_duration\n",
    "                       ,grouped.stop_info_array,grouped.stop_no,grouped.total_stop_duration,grouped.stop_ratio\n",
    "                       ,avg(grouped.smoothed_speed).alias(\"average_trip_speed\")\n",
    "                       ,median(grouped.smoothed_speed).alias(\"median_trip_speed\")\n",
    "                       ,max_udf(grouped.trip_distances).alias(\"max_trip_speed\")\n",
    "                      ,std_udf(grouped.trip_distances).alias(\"std_trip_speed_per_sec\")\n",
    "                      ,calc_total_trip_length(grouped.trip_distances).alias(\"total_trip_distance\")\n",
    "                       ,avg(grouped.pos_accelerations).alias(\"average_acceleration\")\n",
    "                       ,median(grouped.pos_accelerations).alias(\"median_acceleration\")\n",
    "                       ,max_udf(grouped.pos_accelerations).alias(\"max_acceleration\")\n",
    "                       ,std_udf(grouped.pos_accelerations).alias(\"std_acceleration\")\n",
    "                       ,avg(grouped.neg_accelerations).alias(\"average_breaking\")\n",
    "                       ,median(grouped.neg_accelerations).alias(\"median_breaking\")\n",
    "                       ,min_udf(grouped.neg_accelerations).alias(\"max_breaking\")\n",
    "                       ,std_udf(grouped.neg_accelerations).alias(\"std_breaking\")\n",
    "                      ,avg(grouped.circular_accelerations).alias(\"average_circular_acceleration\")\n",
    "                       ,median(grouped.circular_accelerations).alias(\"median_circular_acceleration\")\n",
    "                       ,max_udf(grouped.circular_accelerations).alias(\"max_circular_acceleration\")\n",
    "                       ,std_udf(grouped.circular_accelerations).alias(\"std_circular_acceleration\"))\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.average_trip_speed,grouped.median_trip_speed\n",
    "                       ,grouped.max_trip_speed,grouped.std_trip_speed_per_sec,grouped.total_trip_distance\n",
    "                      ,grouped.average_acceleration,grouped.median_acceleration,grouped.max_acceleration,grouped.std_acceleration\n",
    "                      ,grouped.average_breaking,grouped.median_breaking,grouped.max_breaking,grouped.std_breaking\n",
    "                      ,grouped.average_circular_acceleration, grouped.median_circular_acceleration, grouped.max_circular_acceleration,\n",
    "                       grouped.std_circular_acceleration))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#ANGLES (Changed)\n",
    "\n",
    "import math\n",
    "def get_angle(p1, p2, p3):\n",
    "  dot_product = (p1[0] - p2[0]) * (p3[0] - p2[0]) + (p1[1] - p2[1]) * (p3[1] - p2[1])\n",
    "  denominator = max(distance.euclidean(p1, p2) * distance.euclidean(p2, p3), 0.1)\n",
    "\n",
    "  # just in case dot_product is infinitesimaly larger than denominator\n",
    "  ratio = dot_product / denominator\n",
    "  if ratio > 1:\n",
    "    ratio = 1\n",
    "  if ratio < -1:\n",
    "    ratio = -1\n",
    "  angle = math.acos(ratio)\n",
    "\n",
    "  return angle * 180 / math.pi\n",
    "\n",
    "def calcAngles(trip_array):\n",
    "  angles=[]\n",
    "  for i,x_y in enumerate(trip_array):\n",
    "    if i>1:\n",
    "      angles.append(get_angle(trip_array[i-2],trip_array[i-1],trip_array[i]))\n",
    "  return angles\n",
    "\n",
    "get_angles=f.udf(calcAngles,ArrayType(DoubleType()))\n",
    "get_angle_changes=f.udf(lambda x:np.abs(np.diff(x)).tolist(),ArrayType(DoubleType()))\n",
    "def getTurn(x):\n",
    "  x=np.array(x[10:-10])\n",
    "  return len(x[x>15])\n",
    "\n",
    "get_turn_no=f.udf(getTurn,IntegerType())\n",
    "grouped=grouped.withColumn(\"trip_angles\",get_angles(grouped.trip_array))\n",
    "grouped=grouped.withColumn(\"trip_angle_changes\",get_angle_changes(grouped.trip_angles))\n",
    "grouped=grouped.withColumn(\"avg_trip_angle_changes\",avg(grouped.trip_angles))\n",
    "grouped=grouped.withColumn(\"turn_no\",get_turn_no(grouped.trip_angle_changes))\n",
    "grouped=grouped.withColumn(\"turn_ratio\",grouped.turn_no.cast(\"float\")/grouped.trip_duration)\n",
    "\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.trip_angle_changes,grouped.turn_no,grouped.turn_ratio))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Calculate RDP Smoothed Distances to match the Trips, \n",
    "#it might be better to keep EPSILON THRESHOLD HIGHER\n",
    "EPSILON=10\n",
    "from rdp import rdp \n",
    "get_distances=f.udf(lambda arr:rdp(arr,epsilon=EPSILON),ArrayType(ArrayType(DoubleType())))\n",
    "grouped=grouped.withColumn(\"rdp_smt_trip_array\",get_distances(grouped.trip_array))\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.rdp_smt_trip_array))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#ROTATE RDP smoothed Routes\n",
    "def removeRotation(XY):\n",
    "    \"\"\" change of basis matrix so that the horizontal (x) axis is the vector between the first\n",
    "        and last point\n",
    "\n",
    "        Param: XY must be an N x 2 numpy array\n",
    "        Return: Nx2 array of vectors in new basis\n",
    "\n",
    "        Assumes all XY vectors start at origin (obvious from fn name)\n",
    "    \"\"\"\n",
    "    # calc the unit vectors of the new basis\n",
    "    xdash = XY[-1]\n",
    "\n",
    "    ydash = np.array( [-xdash[1], xdash[0] ])\n",
    "\n",
    "    normXdash = np.linalg.norm(xdash)\n",
    "    normYdash = np.linalg.norm(ydash)\n",
    "\n",
    "    # adapt for round trip!!! \n",
    "    if normXdash > 0:\n",
    "        u = xdash /normXdash\n",
    "    else:\n",
    "        u = np.array([1,0])\n",
    "    if normYdash > 0:\n",
    "        v = ydash / normYdash\n",
    "    else:\n",
    "        v = np.array([0,1])\n",
    "\n",
    "    # change of basis 'matrix' - (x',y') = M(inv)(x,y)\n",
    "    # Minv is just transpose of the new basis matrix M since rotn about origin\n",
    "    Mdash = np.array([[u[0],u[1]],[v[0],v[1]]])\n",
    "\n",
    "    # now transform aall the points t the new basis\n",
    "    # Mdash * XY -> 2x2 x (2xN) hence transpose \n",
    "    XYnew = np.dot(Mdash, np.array(XY).T)\n",
    "\n",
    "    # return it back as Nx2\n",
    "    return (XYnew.T).tolist()\n",
    "def rotate_path( route, angle_to_rotate=90): \n",
    "  rotation_matrix = [   [ np.cos(angle_to_rotate), -1 * np.sin(angle_to_rotate) ], \n",
    "                            [ np.sin(angle_to_rotate),      np.cos(angle_to_rotate) ]  ]\n",
    "  return np.dot( route, rotation_matrix).tolist()\n",
    "get_rotated_array=f.udf(lambda x:removeRotation(np.array(x)),ArrayType(ArrayType(DoubleType())))\n",
    "grouped=grouped.withColumn(\"rotated_rdp_smt_trip_array\",get_rotated_array(grouped.rdp_smt_trip_array))\n",
    "display(grouped.select(grouped.driver,grouped.trip,grouped.rotated_rdp_smt_trip_array))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "grouped.write.format(\"com.databricks.spark.avro\").save(\"/mnt/all-feat-output2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "featurized=sqlContext.read.format(\"com.databricks.spark.avro\").load(\"/mnt/all-feat-output2\")\n",
    "display(featurized)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#CORRELATION ANALYSIS BETWEEN FEATURES\n",
    "\n",
    "drop=['rdp_smt_trip_array',\"trip_distances\",\"trip_angles\", \"trip_array\",\"accelerations\",\"pos_accelerations\",\"neg_accelerations\",\"circular_accelerations\",\"smoothed_speed\",'stop_info_array',\"trip_angle_changes\",'rotated_rdp_smt_trip_array']\n",
    "drop.append(\"driver\")\n",
    "drop.append(\"trip\")\n",
    "drop.append(\"driver_trip\")\n",
    "drop.append(\"dbucket\")\n",
    "drop.append(\"target\")\n",
    "keep = [c for c in featurized.columns if c not in drop]\n",
    "onlyfeatureDF=featurized.select(keep)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Write Features\n",
    "onlyfeatureDF.write.saveAsTable(\"all_feats2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "drop=['rdp_smt_trip_array',\"trip_distances\",\"trip_angles\", \"trip_array\",\"accelerations\",\"pos_accelerations\",\"neg_accelerations\",\"circular_accelerations\",\"smoothed_speed\",'stop_info_array',\"trip_angle_changes\",'rotated_rdp_smt_trip_array']\n",
    "keep = [c for c in featurized.columns if c not in drop]\n",
    "featurized=featurized.select(keep)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#CREATING TRAINING AND TEST SETS\n",
    "import itertools\n",
    "#SEED=123\n",
    "driver_trip_count=200\n",
    "zeros_sample_count=200\n",
    "train_ratio=0.9\n",
    "all_drivers_trip_count=featurized.select([\"trip\"]).count()\n",
    "other_drivers_trip_count=all_drivers_trip_count-driver_trip_count\n",
    "trip_arr=range(1,driver_trip_count+1)\n",
    "makeStr=f.udf(lambda x,y:str(x)+\"_\"+str(y),StringType())\n",
    "train_df_list=[]\n",
    "test_df_list=[]\n",
    "driver_list=[d.driver for d in featurized.select([\"driver\"]).distinct().collect()]\n",
    "for driver in driver_list:\n",
    "  driverdf=(featurized.filter(featurized.driver==driver).withColumn(\"driver_trip\",makeStr(featurized.driver,featurized.trip))\n",
    "            .withColumn(\"dbucket\",f.lit(driver)).withColumn(\"target\",f.lit(1)))\n",
    "  train_trips=list(np.random.choice(trip_arr, int(train_ratio*driver_trip_count),replace=False))\n",
    "  test_trips=list(set(trip_arr)-set(train_trips))\n",
    "  train_onesdf=driverdf.filter(driverdf.trip.isin(train_trips))\n",
    "  test_onesdf=driverdf.filter(driverdf.trip.isin(test_trips))\n",
    "  #test_onesdf.take(1)\n",
    "  other_driv_trips=[str(e[0])+\"_\"+str(e[1]) for e in itertools.product(*[driver_list,trip_arr]) if e[0]!=driver]\n",
    "  \n",
    "  random_other_driv_trips=list(np.random.choice(other_driv_trips, int(driver_trip_count),replace=False))\n",
    "  train_other_driv_trips=list(np.random.choice(random_other_driv_trips, int(driver_trip_count*train_ratio),replace=False))\n",
    "  test_other_driv_trips=list(set(random_other_driv_trips)-set(train_other_driv_trips))\n",
    "  zerosdf=(featurized.filter(featurized.driver!=driver).withColumn(\"driver_trip\",makeStr(featurized.driver,featurized.trip))\n",
    "           .withColumn(\"dbucket\",f.lit(driver)).withColumn(\"target\",f.lit(0)))\n",
    "  train_zerosdf=zerosdf.filter(zerosdf.driver_trip.isin(train_other_driv_trips))\n",
    "  test_zerosdf=zerosdf.filter(zerosdf.driver_trip.isin(test_other_driv_trips))\n",
    "  train_df_list.append(train_onesdf)\n",
    "  train_df_list.append(train_zerosdf)\n",
    "  test_df_list.append(test_onesdf)\n",
    "  test_df_list.append(test_zerosdf)\n",
    "\n",
    "from pyspark.sql import DataFrame\n",
    "def unionAll(dfs):\n",
    "    return reduce(DataFrame.unionAll, dfs)\n",
    "TRAIN=unionAll(train_df_list)\n",
    "TEST=unionAll(test_df_list)\n",
    "#TRAIN.cache()\n",
    "#TEST.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "TRAIN.saveAsTable(\"TRAIN\")\n",
    "TEST.saveAsTable(\"TEST\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "TRAIN=sqlContext.sql(\"SELECT * FROM TRAIN\")\n",
    "TEST=sqlContext.sql(\"SELECT * FROM TEST\")\n",
    "#IMPORTANT HERE I used table caching in Databricks for these two tables!!!!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# CHECK AGAIN\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "from pyspark.sql.functions import col\n",
    "\n",
    "feature_cols=TRAIN.columns\n",
    "#REMOVING non-feature columns\n",
    "feature_cols.remove(\"driver\")\n",
    "feature_cols.remove(\"trip\")\n",
    "feature_cols.remove(\"driver_trip\")\n",
    "feature_cols.remove(\"dbucket\")\n",
    "feature_cols.remove(\"target\")\n",
    "\n",
    "dbucket_list=[d.dbucket for d in TRAIN.select([\"dbucket\"]).distinct().collect()]\n",
    "LabelledTRAIN_DF_list=[]\n",
    "LabelledTEST_DF_list=[]\n",
    "for dbucket in dbucket_list:\n",
    "  #dbucket_TRAIN=TRAIN.filter(TRAIN.dbucket==dbucket)\n",
    "  dbucket_TRAIN=sqlContext.sql(\"SELECT * FROM TRAIN WHERE dbucket=\"+str(dbucket))\n",
    "  #dbucket_TEST=TEST.filter(TEST.dbucket==dbucket)\n",
    "  dbucket_TEST=sqlContext.sql(\"SELECT * FROM TEST WHERE dbucket=\"+str(dbucket))\n",
    "  assembler = VectorAssembler(\n",
    "      inputCols=feature_cols,\n",
    "      outputCol=\"features\")\n",
    "\n",
    "  transformedTRAIN = assembler.transform(dbucket_TRAIN)\n",
    "  LabelledTRAIN=(transformedTRAIN.select(col(\"target\").alias(\"label\"), col(\"features\"))\n",
    "    .map(lambda row: LabeledPoint(row.label, row.features)))\n",
    "  LabelledTRAIN_DF_list.append((dbucket,LabelledTRAIN.toDF()))\n",
    "  #LabelledTRAIN.take(1)\n",
    "\n",
    "  transformedTEST = assembler.transform(dbucket_TEST)\n",
    "  LabelledTEST=(transformedTEST.select(col(\"target\").alias(\"label\"), col(\"features\"))\n",
    "    .map(lambda row: LabeledPoint(row.label, row.features)))\n",
    "  LabelledTEST_DF_list.append((dbucket,LabelledTEST.toDF()))\n",
    "  #LabelledTEST.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#RANDOM FOREST CLASSIFIER\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.classification import RandomForestClassifier,DecisionTreeClassifier,GBTClassifier,GBTClassificationModel\n",
    "from pyspark.mllib.tree import GradientBoostedTrees, GradientBoostedTreesModel\n",
    "#from pyspark.mllib.tree import RandomForest\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.ml.feature import StringIndexer, VectorIndexer\n",
    "\n",
    "def train_driver_model(LabelledTRAIN_DF):\n",
    "  numFolds = 10\n",
    "  #http://stackoverflow.com/questions/28818692/pyspark-mllib-class-probabilities-of-random-forest-predictions\n",
    "  #http://spark.apache.org/docs/latest/ml-classification-regression.html#output-columns-predictions\n",
    "  \"\"\"rf = RandomForest.trainClassifier(trainingData, numClasses=2, categoricalFeaturesInfo={},\n",
    "                                       numTrees=3, featureSubsetStrategy=\"auto\",\n",
    "                                       impurity='gini', maxDepth=4, maxBins=32)\"\"\"\n",
    "  # Index labels, adding metadata to the label column\n",
    "  labelIndexer = StringIndexer(inputCol='label',\n",
    "                               outputCol='indexedLabel').fit(LabelledTRAIN_DF)\n",
    "\n",
    "  # Automatically identify categorical features and index them\n",
    "  featureIndexer = VectorIndexer(inputCol='features',\n",
    "                                 outputCol='indexedFeatures',\n",
    "                                 maxCategories=2).fit(LabelledTRAIN_DF)\n",
    "  rf = RandomForestClassifier(labelCol=\"indexedLabel\", featuresCol=\"indexedFeatures\",impurity='gini')\n",
    "  #dTree = DecisionTreeClassifier(labelCol='indexedLabel', featuresCol='indexedFeatures',impurity='gini')\n",
    "\n",
    "  paramGrid = (ParamGridBuilder()\n",
    "               .addGrid(rf.maxDepth, [5,6,7]).addGrid(rf.numTrees, range(9,15,2)).addGrid(rf.maxBins,[100])\n",
    "               #.addGrid(dTree.maxDepth,[3,4,5,6]).addGrid(dTree.maxBins,[100])\n",
    "               .build())\n",
    "\n",
    "  #\n",
    "  #paramGrid.build()\n",
    "  #https://www.mapr.com/blog/churn-prediction-pyspark-using-mllib-and-ml-packages\n",
    "  #evaluator = MulticlassClassificationEvaluator(predictionCol=\"prediction\",labelCol=\"indexedLabel\",metricName=\"precision\") # + other params as in Scala    \n",
    "  evaluator = BinaryClassificationEvaluator(labelCol='indexedLabel', metricName='areaUnderROC')   \n",
    "  #pipeline = Pipeline(stages=[labelIndexer, featureIndexer, dTree])\n",
    "  pipeline = Pipeline(stages=[labelIndexer, featureIndexer, rf])\n",
    "  crossval = CrossValidator(\n",
    "      estimator=pipeline,\n",
    "     estimatorParamMaps=paramGrid,\n",
    "      evaluator=evaluator,\n",
    "      numFolds=numFolds)\n",
    "\n",
    "  cv_model = crossval.fit(LabelledTRAIN_DF)\n",
    "  best_model = cv_model.bestModel.stages[2]\n",
    "  print(best_model)\n",
    "  return cv_model,evaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#MULTI THREAD TRAINING-->try AGAIN\n",
    "from multiprocessing.pool import ThreadPool\n",
    "N_THREADS=5\n",
    "tpool = ThreadPool(processes=N_THREADS)\n",
    "def execute_training_thread(DF_tuple):\n",
    "  cv_model,evaluator=train_driver_model(DF_tuple[1])\n",
    "  return (DF_tuple[0],(cv_model,evaluator))\n",
    "\n",
    "CV_MODELS_LIST = tpool.map(execute_training_thread,LabelledTRAIN_DF_list)\n",
    "CV_MODELS_LIST"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#EVALUATION -->Try this\n",
    "#vectorized_test_data = vectorizeData(final_test_data)\n",
    "def validate_driver_model(cv_model,LabelledTEST_DF,dbucket,evaluator):\n",
    "  #vectorized_data=vectorizeData(LabelledTEST_DF)\n",
    "  transformed_data = cv_model.transform(LabelledTEST_DF)\n",
    "  auc=evaluator.evaluate(transformed_data)\n",
    "  print \"Driver Bucket:\",dbucket,' ROC_AUC:', auc\n",
    "  getOneProb=f.udf(lambda x:x[\"values\"][1],StringType())\n",
    "  #predictions = transformed_data.select('indexedLabel', 'prediction', getOneProb(transformed_data.probability).alias(\"probability\"))\n",
    "  predictions = transformed_data.select( 'prediction', \"probability\")\n",
    "  predictions=predictions.toPandas()\n",
    "  predictions[\"probability\"] =predictions.apply(lambda x:x[1][1], axis=1)\n",
    "  predictions=sqlContext.createDataFrame(predictions)\n",
    "  print(predictions.toPandas().head(5))\n",
    "  return auc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#MULTI THREADED VALIDATION-->Try This\n",
    "from multiprocessing.pool import ThreadPool\n",
    "N_THREADS=5\n",
    "tpool = ThreadPool(processes=N_THREADS)\n",
    "def execute_validation_thread(CV_MODELS_LIST,DF_tuple):\n",
    "  model_tuple=[v for k,v  in CV_MODELS_LIST if k==DF_tuple[0]][0]\n",
    "  cv_model=model_tuple[0]\n",
    "  evaluator=model_tuple[1]\n",
    "  auc=validate_driver_model(cv_model,DF_tuple[1],DF_tuple[0],evaluator)\n",
    "  return (DF_tuple[0],auc)\n",
    "AUC_LIST = tpool.map(lambda x:execute_validation_thread(CV_MODELS_LIST,x),LabelledTEST_DF_list)\n",
    "print(AUC_LIST)\n",
    "AVG_AUC=np.mean([v for k,v in AUC_LIST])\n",
    "print(\"Average AUC\",str(AVG_AUC))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#PREDICTING ALL PROBABILITIES AND CREATING OUTPUT DF\n",
    "\n",
    "from pyspark.sql import DataFrame\n",
    "assembler = VectorAssembler(\n",
    "      inputCols=feature_cols,\n",
    "      outputCol=\"features\")\n",
    "driver_list=[d.driver for d in featurized.select([\"driver\"]).distinct().collect()]\n",
    "driver_final_df_list=[]\n",
    "for driver in driver_list:\n",
    "  driver_featurized=featurized.filter(featurized.driver==driver)\n",
    "  driver_transformed = assembler.transform(driver_featurized)\n",
    "  labelled_driver=(driver_transformed.select(f.lit(1).alias(\"label\"), col(\"features\"))\n",
    "      .map(lambda row: LabeledPoint(row.label, row.features)))\n",
    "  \n",
    "  cv_model=[v[0] for k,v  in CV_MODELS_LIST if k==driver][0]\n",
    "  print(cv_model)\n",
    "  transformed_data=cv_model.transform(labelled_driver.toDF())\n",
    "  print(transformed_data)\n",
    "  predictions = transformed_data.select('indexedLabel', 'prediction', \"probability\")\n",
    "  print(transformed_data.columns)\n",
    "  predictions=predictions.toPandas()\n",
    "  predictions[\"probability\"] =predictions.apply(lambda x:str(x[2][1]), axis=1)\n",
    "  #predictions=sqlContext.createDataFrame(predictions).select(['indexedLabel',\"probability\"])\n",
    "  pan_driver_featurized=driver_featurized.toPandas()\n",
    "  pan_driver_featurized[\"driver_trip\"] =pan_driver_featurized[[\"driver\",\"trip\"]].apply(lambda x: str(x[0])+'_'+str(x[1]), axis=1)\n",
    "  driver_final=sqlContext.createDataFrame(pan_driver_featurized.join(predictions))\n",
    "  #driver_final=joined.withColumn(\"driver_trip\",makeStr(featurized.driver,featurized.trip))\n",
    "  driver_final=driver_final.select([\"driver_trip\",\"probability\"])\n",
    "  driver_final_df_list.append(driver_final)\n",
    "\n",
    "def unionAll(dfs):\n",
    "  return reduce(DataFrame.unionAll, dfs)\n",
    "finalDF=unionAll(driver_final_df_list)\n",
    "finalDF.cache()\n",
    "display(finalDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#TRAIN.repartition(4,[\"dbucket\"])\n",
    "#TRAIN.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "dbucket_list=[d.dbucket for d in TRAIN.select([\"dbucket\"]).distinct().collect()]\n",
    "for i in range(len(dbucket_list)/4):\n",
    "  print(\"Start Driver Bucket Batch no:\",str(i+1))\n",
    "  print(dbucket_list[i*4:((i+1)*4)])\n",
    " \n",
    "  print(\"End Driver Bucket Batch no:\",str(i+1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#MULTI THREAD TRAINING\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.mllib.regression import LabeledPoint\n",
    "from pyspark.sql.functions import col\n",
    "import itertools\n",
    "feature_cols=TRAIN.columns\n",
    "#REMOVING non-feature columns\n",
    "feature_cols.remove(\"driver\")\n",
    "feature_cols.remove(\"trip\")\n",
    "feature_cols.remove(\"driver_trip\")\n",
    "feature_cols.remove(\"dbucket\")\n",
    "feature_cols.remove(\"target\")\n",
    "\n",
    "#REMOVING some features with high correlation\n",
    "feature_cols.remove(\"max_breaking\")\n",
    "feature_cols.remove(\"average_circular_acceleration\")\n",
    "feature_cols.remove(\"average_trip_speed\")\n",
    "feature_cols.remove(\"max_circular_acceleration\")\n",
    "#feature_cols.remove(\"std_circular_acceleration\")\n",
    "\n",
    "assembler = VectorAssembler(\n",
    "      inputCols=feature_cols,\n",
    "      outputCol=\"features\")\n",
    "########################################\n",
    "def execute_training_thread(dbucket):\n",
    "  dbucket_TRAIN=sqlContext.sql(\"SELECT * FROM TRAIN WHERE dbucket=\"+str(dbucket))\n",
    "  transformedTRAIN = assembler.transform(dbucket_TRAIN)\n",
    "  LabelledTRAIN=(transformedTRAIN.select(col(\"target\").alias(\"label\"), col(\"features\"))\n",
    "    .map(lambda row: LabeledPoint(row.label, row.features))).toDF()\n",
    "  cv_model,evaluator=train_driver_model(LabelledTRAIN)\n",
    "  return (dbucket,(cv_model,evaluator))\n",
    "#########################################\n",
    "\n",
    "dbucket_list=[d.dbucket for d in TRAIN.select([\"dbucket\"]).distinct().collect()]\n",
    "models_agg_list=[]\n",
    "from multiprocessing.pool import ThreadPool\n",
    "N_THREADS=4\n",
    "N=4\n",
    "for i in range(len(dbucket_list)/N):\n",
    "  print(\"Start Driver Bucket Batch no:\",str(i+1))\n",
    "  tpool = ThreadPool(processes=N_THREADS)\n",
    "  models_sub_list=tpool.map(execute_training_thread,dbucket_list[i*N:((i+1)*N)])\n",
    "  models_agg_list.append(models_sub_list)\n",
    "  print(\"End Driver Bucket Batch no:\",str(i+1))\n",
    "CV_MODELS_LIST = [e for e in itertools.chain(*models_agg_list)]\n",
    "CV_MODELS_LIST\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#MULTI THREADED VALIDATION\n",
    "def execute_validation_thread(CV_MODELS_LIST,dbucket):\n",
    "  dbucket_TEST=sqlContext.sql(\"SELECT * FROM TEST WHERE dbucket=\"+str(dbucket))\n",
    "  transformedTEST = assembler.transform(dbucket_TEST)\n",
    "  LabelledTEST=(transformedTEST.select(col(\"target\").alias(\"label\"), col(\"features\"))\n",
    "    .map(lambda row: LabeledPoint(row.label, row.features))).toDF()\n",
    "  model_tuple=[v for k,v  in CV_MODELS_LIST if k==dbucket][0]\n",
    "  cv_model=model_tuple[0]\n",
    "  evaluator=model_tuple[1]\n",
    "  auc=validate_driver_model(cv_model,LabelledTEST,dbucket,evaluator)\n",
    "  return (dbucket,auc)\n",
    "AUC_LIST = tpool.map(lambda x:execute_validation_thread(CV_MODELS_LIST,x),dbucket_list)\n",
    "print(AUC_LIST)\n",
    "AVG_AUC=np.mean([v for k,v in AUC_LIST])\n",
    "print(AVG_AUC)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#Correlation Matrix to observe the correlated features\n",
    "from pyspark.mllib.stat import Statistics\n",
    "import pandas as pd\n",
    "featTRAIN=LabelledTRAIN.map(lambda lp:lp.features)\n",
    "correlation_matrix = Statistics.corr(featTRAIN, method=\"spearman\")\n",
    "#display(correlation_matrix)\n",
    "\n",
    "pd.set_option('display.max_columns', 50)\n",
    "\n",
    "corr_df = pd.DataFrame(correlation_matrix, index=feature_cols, columns=feature_cols)\n",
    "corr_disp_df=corr_df\n",
    "corr_disp_df.insert(0, 'features',corr_disp_df.index)\n",
    "display(sqlContext.createDataFrame(corr_disp_df))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# get a boolean dataframe where true means that a pair of variables is highly correlated\n",
    "highly_correlated_df = (abs(corr_df[feature_cols]) > .8) & (corr_df[feature_cols] < 1.0)\n",
    "# get the names of the variables so we can use them to slice the dataframe\n",
    "correlated_vars_index = (highly_correlated_df==True).any()\n",
    "correlated_var_names = correlated_vars_index[correlated_vars_index==True].index\n",
    "# slice it\n",
    "highly_correlated_df.loc[correlated_var_names,correlated_var_names]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#PREDICTING ALL PROBABILITIES AND CREATING OUTPUT DF\n",
    "\n",
    "from pyspark.sql import DataFrame\n",
    "assembler = VectorAssembler(\n",
    "      inputCols=feature_cols,\n",
    "      outputCol=\"features\")\n",
    "driver_list=[d.driver for d in featurized.select([\"driver\"]).distinct().collect()]\n",
    "driver_final_df_list=[]\n",
    "for driver in driver_list:\n",
    "  driver_featurized=featurized.filter(featurized.driver==driver)\n",
    "  driver_transformed = assembler.transform(driver_featurized)\n",
    "  labelled_driver=(driver_transformed.select(f.lit(1).alias(\"label\"), col(\"features\"))\n",
    "      .map(lambda row: LabeledPoint(row.label, row.features)))\n",
    "  \n",
    "  cv_model=[v[0] for k,v  in CV_MODELS_LIST if k==driver][0]\n",
    "  print(cv_model)\n",
    "  transformed_data=cv_model.transform(labelled_driver.toDF())\n",
    "  print(transformed_data)\n",
    "  predictions = transformed_data.select('indexedLabel', 'prediction', \"probability\")\n",
    "  print(transformed_data.columns)\n",
    "  predictions=predictions.toPandas()\n",
    "  predictions[\"probability\"] =predictions.apply(lambda x:str(x[2][1]), axis=1)\n",
    "  #predictions=sqlContext.createDataFrame(predictions).select(['indexedLabel',\"probability\"])\n",
    "  pan_driver_featurized=driver_featurized.toPandas()\n",
    "  pan_driver_featurized[\"driver_trip\"] =pan_driver_featurized[[\"driver\",\"trip\"]].apply(lambda x: str(x[0])+'_'+str(x[1]), axis=1)\n",
    "  driver_final=sqlContext.createDataFrame(pan_driver_featurized.join(predictions))\n",
    "  #driver_final=joined.withColumn(\"driver_trip\",makeStr(featurized.driver,featurized.trip))\n",
    "  driver_final=driver_final.select([\"driver_trip\",\"probability\"])\n",
    "  driver_final_df_list.append(driver_final)\n",
    "\n",
    "def unionAll(dfs):\n",
    "  return reduce(DataFrame.unionAll, dfs)\n",
    "finalDF=unionAll(driver_final_df_list)\n",
    "finalDF.cache()\n",
    "display(finalDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": true
   },
   "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.5.1"
  },
  "name": "telefonica",
  "notebookId": 4981
 },
 "nbformat": 4,
 "nbformat_minor": 0
}