spark_dbconnect.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ArcGIS Pro, Jupyter Notebook and Databricks\n",
    "\n",
    "This notebook demonstrates the authoring of a Spark Job in a local Jupyter notebook in ArcGIS Pro. However, the execution of the Job is performed remotely on a predefined [Databricks](https://databricks.com/) cluster.\n",
    "\n",
    "In this example, we started a Spark cluster on [Microsoft Azure using Databricks](https://databricks.com/product/azure).  It is a small cluster consisting of:\n",
    "\n",
    "- 1 driver machine with 14GB of Memory and 4 cores\n",
    "- 2-8 worker machines, where each has 14GB of Memory and 4 cores.\n",
    "\n",
    "![](media/Cluster.png)\n",
    "\n",
    "To connect the local notebook to the cluster, we use the [Databricks Connect API](https://docs.databricks.com/dev-tools/databricks-connect.html).\n",
    "\n",
    "### Setting Up an ArcGIS Pro Conda Env.\n",
    "\n",
    "Using the ArcGIS Pro Python Command Prompt, the following are the steps to create an ArcGIS Pro conda environment to enable the local notebook to execute remotely on a cluster with a Databricks Runtime version 6.6.  \n",
    "\n",
    "```\n",
    "proswap arcgispro-py3\n",
    "conda remove --yes --all --name dbconnect\n",
    "conda create --yes --name dbconnect4 --clone arcgispro-py3\n",
    "activate dbconnect4\n",
    "proswap dbconnect4\n",
    "conda install --yes -c conda-forge untangle\n",
    "pip install -U databricks-connect==6.6 databricks-cli pyarrow\n",
    "clone https://github.com/mraad/https://github.com/mraad/spark-esri.git.git\n",
    "cd https://github.com/mraad/spark-esri.git\n",
    "python setup.py install\n",
    "```\n",
    "\n",
    "Next, get a [personal access token](https://docs.databricks.com/dev-tools/api/latest/authentication.html) and save it somewhere safe.\n",
    "\n",
    "And finally, [configure](https://docs.databricks.com/dev-tools/databricks-connect.html#set-up-client) the Databricks connection and test it.\n",
    "\n",
    "```\n",
    "databricks-connect configure\n",
    "powershell\n",
    "$env:DATABRICKS_TOKEN=xxx-xxx-xxx-xxx\n",
    "$env:SPARK_HOME=\"$env:LOCALAPPDATA\\esri\\conda\\envs\\dbconnect\\lib\\site-packages\\pyspark\".ToLower()\n",
    "\n",
    "$env:SPARK_HOME=\"$env:LOCALAPPDATA\\esri\\conda\\envs\\spark_esri2\\lib\\site-packages\\pyspark\".ToLower()\n",
    "\n",
    "databricks-connect test\n",
    "```\n",
    "\n",
    "Note that 2 inlined environment variables were defined to perform the test.  However, it is **HIGHLY** advisable to define them in the System Properties and make sure to relaunch Pro to take effect."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import required modules."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from spark_dbconnect import spark_start, spark_stop"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Configure and start a remote Spark instance.\n",
    "\n",
    "Note that the personal access token in retrieved from the `DATABRICKS_TOKEN` env variable.  The other values can be derived from the cluster url.  For example:\n",
    "\n",
    "```\n",
    "https://adb-2740165739887726.6.azuredatabricks.net/?o=2740165739887726#/setting/clusters/0802-221206-sols809/configuration\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "config = {\n",
    "    \"spark.databricks.service.server.enabled\": True,\n",
    "    \"spark.databricks.service.address\": \"https://adb-2740165739887726.6.azuredatabricks.net\",\n",
    "    \"spark.databricks.service.token\": os.environ[\"DATABRICKS_TOKEN\"],\n",
    "    \"spark.databricks.service.clusterId\": \"0802-221206-sols809\",\n",
    "    \"spark.databricks.service.orgId\": \"2740165739887726\",\n",
    "    \"spark.databricks.service.port\": \"15001\"\n",
    "}\n",
    "spark = spark_start(config=config)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For speed purposes and bandwidth reduction, it is best to colocate the data storage and the execution engine.\n",
    "\n",
    "Databricks provides a [distributed file system](https://docs.databricks.com/data/databricks-file-system.html) that is accessable from any running cluster.\n",
    "\n",
    "Using the [ExportToFile toolbox](tools/ParquetToolkit.pyt), we exported the Broadcast point features in the [Miami AIS sample geodatabase](ftp://ftp.coast.noaa.gov/pub/MSP/AIS/AIS.SampleData.zip) to a local [parquet](https://parquet.apache.org/) file.\n",
    "\n",
    "![](media/ExportToFile.png)\n",
    "\n",
    "We configured the [Databricks CLI](https://docs.databricks.com/dev-tools/cli/index.html#set-up-the-cli):\n",
    "\n",
    "```\n",
    "dbfs configure --token\n",
    "```\n",
    "\n",
    "And then, we uploaded the local parquet file to DBFS.\n",
    "\n",
    "```\n",
    "dbfs cp BroadcastXY.parq dbfs:/FileStore/tables\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a Spark dataframe from a parquet file.\n",
    "\n",
    "In addition to storing the data in a columar format, Parquet files hold also the column types enabling the dataframe schema generation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.parquet(\"dbfs:/FileStore/tables/BroadcastXY.parq\")\n",
    "df.createOrReplaceTempView(\"v0\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Display the dataframe schema."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Perform spatial binning at 100x100 meters.\n",
    "\n",
    "Note that due to the @ sign in the field name, the field is escaped with back quotes in the inner select statement."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cell0 = 100.0 # meters\n",
    "cell1 = cell0 * 0.5\n",
    "\n",
    "rows = spark.sql(f\"\"\"\n",
    "select q*{cell0}+{cell1} x,r*{cell0}+{cell1} y,least(count(1),1000) as pop\n",
    "from\n",
    "(select cast(`Shape@X`/{cell0} as long) q,cast(`Shape@Y`/{cell0} as long) r from v0)\n",
    "group by q,r\n",
    "\"\"\")\\\n",
    "    .collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an ephemeral feature class in the TOC of the collected bins."
   ]
  },
  {
   "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": [
    "### 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
}