spark_esri.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ArcGIS Notebook to be demonstrate the usage of the built-in [Spark](https://spark.apache.org/) engine.\n",
    "\n",
    "\n",
    "This notebook will demonstrate the spatial binning of AIS data around the port of Miami using Apache Spark.\n",
    "\n",
    "The AIS broadcast data is in a FileGeodatabase that can be download from [here](https://marinecadastre.gov/ais). Note only data from 2010-2014 is in FileGeodatabase format."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import the required modules."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import arcpy\n",
    "\n",
    "from spark_esri import spark_start, spark_stop"
   ]
  },
  {
   "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": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "config = {\n",
    "    \"spark.driver.memory\":\"16G\",\n",
    "    \"spark.executor.memory\":\"16G\"\n",
    "}\n",
    "\n",
    "spark_stop()\n",
    "spark = spark_start(config=config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let's check the spark version and some env vars."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.3.0-esri-6'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\"CONDA_DEFAULT_ENV\" in os.environ # Should be False, this was for Pro pre 2.8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "os.getenv(\"CONDA_DEFAULT_ENV\") # if set, what is the value ? Spark will use the python.exe in that folder."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'c:\\\\program files\\\\arcgis\\\\pro\\\\Java\\\\runtime\\\\jre'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "os.environ[\"JAVA_HOME\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'C:\\\\Program Files\\\\ArcGIS\\\\Pro\\\\java\\\\runtime\\\\hadoop'"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "os.environ[\"HADOOP_HOME\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'C:\\\\Program Files\\\\ArcGIS\\\\Pro\\\\java\\\\runtime\\\\spark'"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "os.environ[\"SPARK_HOME\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'C:\\\\Users\\\\mraad\\\\AppData\\\\Local\\\\ESRI\\\\conda\\\\envs\\\\sparkgeo\\\\python.exe'"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "os.environ[\"PYSPARK_PYTHON\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read the selected Broadcast feature shapes in WebMercator SR.\n",
    "\n",
    "It is assumed that you added to the map the `Broadcast` point feature class from the download `miami.gdb`.\n",
    "\n",
    "Note that the `SearchCursor` is subject to the user selected features, and to an active query definition in the layer properties. For Example, set the query definition to `Stauts = 0: Under way using engine` to get the location of all moving ships, in such that we get a \"heat map\" of the port movement."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sp_ref = arcpy.SpatialReference(3857)\n",
    "data = arcpy.da.SearchCursor(\"Broadcast\", [\"SHAPE@X\",\"SHAPE@Y\"], spatial_reference=sp_ref)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a Spark data frame of the read data, and create a view named 'v0'."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark\\\n",
    "    .createDataFrame(data, \"x double,y double\")\\\n",
    "    .createOrReplaceTempView(\"v0\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregate the data at 200x200 meters bins.\n",
    "\n",
    "The aggregation is performed by Spark as a SQL statement in a parallel share-nothing way and the resulting bins are collected back in the `rows` array variable.\n",
    "\n",
    "This is a nested SQL expression, where the inner expression is mapping the input `x` and `y` into `q` and `r` cell locations given a user defined bin size, and the outter expression is aggreating as a sum the `q` and `r` pairs. Finally, `q` and `r` are mapped back to `x` and `y` to enble the placement on a map. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cell0 = 200.0 # meters\n",
    "cell1 = cell0 * 0.5\n",
    "\n",
    "rows = spark\\\n",
    "    .sql(f\"\"\"\n",
    "with t as (select cast(x/{cell0} as long) q,cast(y/{cell0} as long) r from v0)\n",
    "\n",
    "select q*{cell0}+{cell1} x,r*{cell0}+{cell1} y,least(count(1),400) as pop\n",
    "from t\n",
    "group by q,r\n",
    "\"\"\")\\\n",
    "    .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Report the number of bins"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(rows)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an in-memory point feature class of the collected bins.\n",
    "\n",
    "The variable `rows` is an array of form `[[x0,y0,pop0],[x1,y1,pop1],...,[xN,yN,popN]]`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ws = \"memory\"\n",
    "nm = \"Bins\"\n",
    "\n",
    "fc = os.path.join(ws,nm)\n",
    "\n",
    "arcpy.management.Delete(fc)\n",
    "\n",
    "sp_ref = arcpy.SpatialReference(3857)\n",
    "arcpy.management.CreateFeatureclass(ws,nm,\"POINT\",spatial_reference=sp_ref)\n",
    "arcpy.management.AddField(fc, \"POP\", \"LONG\")\n",
    "\n",
    "with arcpy.da.InsertCursor(fc, [\"SHAPE@X\",\"SHAPE@Y\", \"POP\"]) as cursor:\n",
    "    for row in rows:\n",
    "        cursor.insertRow(row)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Apply a graduated colors symbology to highlight the bins."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# _ = arcpy.ApplySymbologyFromLayer_management(fc, f\"{nm}.lyrx\")"
   ]
  },
  {
   "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 (ipykernel)",
   "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.9.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}