taxi_trips_duration_error.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Notebook to draw the errors of the predicted trip durations from a selected location.\n",
    "\n",
    "### Import the required modules."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import arcpy\n",
    "\n",
    "from spark_esri import spark_start, spark_stop\n",
    "\n",
    "from pyspark.sql.functions import col, lit"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start a Spark instance.\n",
    "\n",
    "Note the `config` argument to [configure the Spark instance](https://spark.apache.org/docs/latest/configuration.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "spark_stop()\n",
    "\n",
    "config = {\"spark.driver.memory\":\"2G\"}\n",
    "spark = spark_start(config=config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a Spark data frame of the selected predictions features, and create a view named 'v0'.\n",
    "\n",
    "A new column (`error`) is added which is the square of the difference between `duration` and `duration_predicted`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fields = ['plon','plat','dlon','dlat','duration','duration_predicted']\n",
    "\n",
    "schema = \",\".join([f\"{f} double\" for f in fields])\n",
    "\n",
    "with arcpy.da.SearchCursor(\"Predictions\",fields) as data:\n",
    "    spark\\\n",
    "        .createDataFrame(data,schema)\\\n",
    "        .withColumn(\"delta\",col(\"duration\")-col(\"duration_predicted\"))\\\n",
    "        .withColumn(\"error\", col(\"delta\")*col(\"delta\"))\\\n",
    "        .drop(\"delta\")\\\n",
    "        .createOrReplaceTempView(\"v0\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Calculate the average of the pickup locations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rows = spark\\\n",
    "    .sql(\"\"\"select avg(plon) plon,avg(plat) plat from v0\"\"\")\\\n",
    "    .collect()\n",
    "\n",
    "plon,plat = rows[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregate the dropoff location at bin locations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cell1 = 0.05\n",
    "cell2 = cell1 * 0.5\n",
    "\n",
    "spark\\\n",
    "    .sql(f\"\"\"\n",
    "select\n",
    "cast(dlon/{cell1} as long) dq,\n",
    "cast(dlat/{cell1} as long) dr,\n",
    "error\n",
    "from v0\n",
    "\"\"\")\\\n",
    "    .createOrReplaceTempView('v1')\n",
    "\n",
    "rows = spark\\\n",
    "    .sql(f\"\"\"\n",
    "select\n",
    "dq*{cell1}+{cell2} dlon,\n",
    "dr*{cell1}+{cell2} dlat,\n",
    "avg(error) mse\n",
    "from v1\n",
    "group by dq,dr\n",
    "\"\"\")\\\n",
    "    .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an in-memory linestring features between the avg pickup location and the dropoff bins."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ws = \"memory\"\n",
    "nm = \"Trips\"\n",
    "\n",
    "fc = os.path.join(ws,nm)\n",
    "\n",
    "arcpy.management.Delete(fc)\n",
    "\n",
    "sp_ref = arcpy.SpatialReference(4326)\n",
    "arcpy.management.CreateFeatureclass(ws,nm,\"POLYLINE\",spatial_reference=sp_ref)\n",
    "arcpy.management.AddField(fc, \"MSE\", \"DOUBLE\")\n",
    "\n",
    "with arcpy.da.InsertCursor(fc, [\"SHAPE@WKT\",\"MSE\"]) as cursor:\n",
    "    for dlon,dlat,mse in rows:\n",
    "        wkt = f\"LINESTRING({plon} {plat},{dlon} {dlat})\"\n",
    "        cursor.insertRow((wkt,mse))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Stop the spark instance."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_stop()"
   ]
  }
 ],
 "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}