Analysis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "toc": "true"
   },
   "source": [
    "# Table of Contents\n",
    " <p><div class=\"lev1\"><a href=\"#The-Data\"><span class=\"toc-item-num\">1&nbsp;&nbsp;</span>The Data</a></div><div class=\"lev2\"><a href=\"#EuroStat\"><span class=\"toc-item-num\">1.1&nbsp;&nbsp;</span>EuroStat</a></div><div class=\"lev3\"><a href=\"#Exercise:-Getting-Spatial-Boundary-Data\"><span class=\"toc-item-num\">1.1.1&nbsp;&nbsp;</span>Exercise: Getting Spatial Boundary Data</a></div><div class=\"lev2\"><a href=\"#OpenStreetMap/Overpass-API\"><span class=\"toc-item-num\">1.2&nbsp;&nbsp;</span>OpenStreetMap/Overpass API</a></div><div class=\"lev3\"><a href=\"#Exercise:-Getting-to-know-OSM-Data\"><span class=\"toc-item-num\">1.2.1&nbsp;&nbsp;</span>Exercise: Getting to know OSM Data</a></div><div class=\"lev1\"><a href=\"#What-are-we-going-to-do?\"><span class=\"toc-item-num\">2&nbsp;&nbsp;</span>What are we going to do?</a></div><div class=\"lev2\"><a href=\"#What-do-we-mean-by-culture/cultural-score?\"><span class=\"toc-item-num\">2.1&nbsp;&nbsp;</span>What do we mean by culture/cultural score?</a></div><div class=\"lev1\"><a href=\"#GeoPandas\"><span class=\"toc-item-num\">3&nbsp;&nbsp;</span>GeoPandas</a></div><div class=\"lev2\"><a href=\"#OGC-SimpleFeatures\"><span class=\"toc-item-num\">3.1&nbsp;&nbsp;</span>OGC SimpleFeatures</a></div><div class=\"lev3\"><a href=\"#WKT/WKB:-Well-Known-Text-and-Well-Known-Binary\"><span class=\"toc-item-num\">3.1.1&nbsp;&nbsp;</span>WKT/WKB: Well Known Text and Well Known Binary</a></div><div class=\"lev3\"><a href=\"#Implications-for-PySpark-data-de/serialization-&amp;-un/marshalling\"><span class=\"toc-item-num\">3.1.2&nbsp;&nbsp;</span>Implications for PySpark data de/serialization &amp; un/marshalling</a></div><div class=\"lev2\"><a href=\"#Exercise:\"><span class=\"toc-item-num\">3.2&nbsp;&nbsp;</span>Exercise:</a></div><div class=\"lev2\"><a href=\"#Create-a-pyspark-dataframe-from-GeoPanadas\"><span class=\"toc-item-num\">3.3&nbsp;&nbsp;</span>Create a pyspark dataframe from GeoPanadas</a></div><div class=\"lev3\"><a href=\"#Exercise\"><span class=\"toc-item-num\">3.3.1&nbsp;&nbsp;</span>Exercise</a></div><div class=\"lev1\"><a href=\"#Lets-look-at-the-POI-data-next\"><span class=\"toc-item-num\">4&nbsp;&nbsp;</span>Lets look at the POI data next</a></div><div class=\"lev1\"><a href=\"#Count-the-number-of-museums-in-Zurich\"><span class=\"toc-item-num\">5&nbsp;&nbsp;</span>Count the number of museums in Zurich</a></div><div class=\"lev2\"><a href=\"#Exercise\"><span class=\"toc-item-num\">5.1&nbsp;&nbsp;</span>Exercise</a></div><div class=\"lev1\"><a href=\"#SQL-is-nice-but...\"><span class=\"toc-item-num\">6&nbsp;&nbsp;</span>SQL is nice but...</a></div><div class=\"lev1\"><a href=\"#Joins,-Spatial-Predicate,-Broadcast-Variables,-UDFs\"><span class=\"toc-item-num\">7&nbsp;&nbsp;</span>Joins, Spatial Predicate, Broadcast Variables, UDFs</a></div><div class=\"lev2\"><a href=\"#Grouping\"><span class=\"toc-item-num\">7.1&nbsp;&nbsp;</span>Grouping</a></div><div class=\"lev2\"><a href=\"#Adding-population-column\"><span class=\"toc-item-num\">7.2&nbsp;&nbsp;</span>Adding population column</a></div><div class=\"lev2\"><a href=\"#Calculating-the-score\"><span class=\"toc-item-num\">7.3&nbsp;&nbsp;</span>Calculating the score</a></div><div class=\"lev1\"><a href=\"#Spatial-Spark\"><span class=\"toc-item-num\">8&nbsp;&nbsp;</span>Spatial Spark</a></div><div class=\"lev2\"><a href=\"#Spatial-Partition\"><span class=\"toc-item-num\">8.1&nbsp;&nbsp;</span>Spatial Partition</a></div><div class=\"lev2\"><a href=\"#Spatial-Range-Query\"><span class=\"toc-item-num\">8.2&nbsp;&nbsp;</span>Spatial Range Query</a></div><div class=\"lev1\"><a href=\"#Perform-Analysis-on-All-Features-and-All-Cities\"><span class=\"toc-item-num\">9&nbsp;&nbsp;</span>Perform Analysis on All Features and All Cities</a></div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import notebook\n",
    "\n",
    "import os.path, json, io, pandas\n",
    "import matplotlib.pyplot as plt\n",
    "import matplotlib\n",
    "matplotlib.rcParams['figure.figsize'] = (16, 20)\n",
    "\n",
    "from retrying import retry # for exponential back down when calling TurboOverdrive API\n",
    "\n",
    "import pyspark.sql.functions as func # resuse as func.coalace for example\n",
    "from pyspark.sql.types import StringType, IntegerType, FloatType, DoubleType,DecimalType\n",
    "\n",
    "import pandas as pandas\n",
    "from geopandas import GeoDataFrame # Loading Boundries Data\n",
    "from shapely.geometry import Point, Polygon, shape # creating geospatial data\n",
    "from shapely import wkb, wkt # creating and parsing geospatial data\n",
    "import overpy # OpenStreetMap API\n",
    "\n",
    "%matplotlib inline\n",
    "\n",
    "# make sure nbextensions are installed\n",
    "notebook.nbextensions.check_nbextension('usability/codefolding', user=True)\n",
    "\n",
    "try:\n",
    "    sc\n",
    "except NameError:\n",
    "    import pyspark\n",
    "    sc = pyspark.SparkContext('local[*]')\n",
    "    sqlContext = pyspark.sql.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# The Data\n",
    "\n",
    "We are going to use 2 main datasets:\n",
    "\n",
    "1. Boundaries/Urban footprints of some of the largest cities in Europe along with the total population for 2013 for those urban areas.\n",
    "2. OpenStreetMap points of interests that have culturally relevant tags\n",
    "\n",
    "## EuroStat\n",
    "\n",
    "European Commission's agency \"Eurostat\" conducted a Cities (Urban Audit) which is a Data Database containing detailed year-by-year statistics based on objective measurements. They have an incredibly rich and valuable database. The data set we are using defines cities in terms of two distinct areas (with possibly a third area - \"Kernel\" - for large cities like London and Paris). The regions are \"Core City\" and \"Large Urban Zone\". There is a great article that explains this in detail (Cities explained by EuroStat)[http://ec.europa.eu/eurostat/statistics-explained/index.php/European_cities_-_spatial_dimension] This is extremely important for our analysis as we want to do spatial analysis bounded by a well defined region whose population can be reliable defined. The EuroStat Database will give us this.\n",
    "\n",
    "### Exercise: Getting Spatial Boundary Data\n",
    "\n",
    "Just spend 5 minutes looking at the database for Urban Audit and try to download 2013 population data for European cities. http://bit.ly/eurostat-urban-audit-database WARNING: you can get lost on their website and emerge 2 hours later wondering where all the time went.\n",
    "\n",
    "## OpenStreetMap/Overpass API\n",
    "\n",
    "For the Cultural data we are going to use OpenStreetMap. Now there are many ways to obtain OSM data. we will use the Overpass Turbo. Overpass turbo is a web based data mining tool for OpenStreetMap. You find it at http://overpass-turbo.eu. It runs any kind of Overpass API query and shows the results on an interactive map. The source code is found on https://github.com/tyrasd/overpass-turbo. A huge thanks goes to Roland who created the Overpass API, without which this tool wouldn't exist. The Overpass API (formerly known as OSM Server Side Scripting, or OSM3S) is a read-only API that serves up custom selected parts of the OSM map data. It acts as a database over the web: the client sends a query to the API and gets back the data set that corresponds to the query. Unlike the main API, which is optimized for editing, Overpass API is optimized for data consumers that need a few elements within a glimpse or up to roughly 100 million elements in some minutes, both selected by search criteria like e.g. location, type of objects, tag properties, proximity, or combinations of them. It acts as a database backend for various services. See [Overpass API Wiki page](http://wiki.openstreetmap.org/wiki/Overpass_API) for more details.\n",
    "\n",
    "We will use Overpass API to download POI data of cultural relevance. To understand the Overpass API have a look at the quick Overpass query language guide here: http://wiki.openstreetmap.org/wiki/Overpass_API/Language_Guide \n",
    "\n",
    "### Exercise: Getting to know OSM Data\n",
    "\n",
    "Just spend 5-10 minutes looking at Overpass API. Try to Query datasets like all the museums. To gain a better understanding of the OSM Tags and meanings behind them look at the Tag Features Wiki pages via Tag Info Website: http://taginfo.openstreetmap.org/keys If you are interested in reading more about a key you can click on it then go to wiki tab an click on the wiki page in your language of choice. *Hint* there is a link to the Turbo Overpass page on the site too.\n",
    "\n",
    "\n",
    "# What are we going to do?\n",
    "\n",
    "We will build a data processing pipeline that takes our data sources performs aggregations by precise boundaries. Then run a weighted algorithm to calculate a \"cultural\" score for each city normalized by its total population. So the pipeline is something like this:\n",
    "\n",
    "![](images/sketch.png)\n",
    "\n",
    "## What do we mean by culture/cultural score?\n",
    "\n",
    "This is a completely subjective and our algorithm is therefore only meant as an excuse to have fun learning PySpark and Spatial Analysis. We collect data from OpenStreetMap that seems culturally relevant. These include the following tags:\n",
    "\n",
    "    tourism\"=\"artwork\"\n",
    "    tourism\"=\"gallery\"\n",
    "    amenity\"=\"theatre\"\n",
    "    amenity\"=\"arts_centre\"\n",
    "    tourism\"=\"museum\"\n",
    "\n",
    "We will also assign different weights to each tag. From 1 to 5. The tags above are listed from highest (artwork) to lowest (museum). This should make our analysis interesting. We will calculate a score by performing a weighted count for each of the sites found per city. Finally we will normalize our count by dividing it by the total population of the city."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": [
     3,
     7,
     23
    ],
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "### Helper functions\n",
    "\n",
    "# given shapely bounds return bbox compatiable with overpass turbo openstreetmap API\n",
    "def bbox(bounds):\n",
    "    return (bounds[1],bounds[0],bounds[3],bounds[2])\n",
    "\n",
    "# given an openstreetmap node retrun a GeoJSON feature\n",
    "def nodeToFeature(node):\n",
    "    properties = node.tags\n",
    "    properties['wkt'] = Point(node.lon, node.lat).wkt\n",
    "    return {\n",
    "        \"type\": \"Feature\",\n",
    "        \"properties\": properties,\n",
    "        \"geometry\": {\n",
    "            \"type\": \"Point\",\n",
    "            \"coordinates\": [\n",
    "                float(node.lon),\n",
    "                float(node.lat)\n",
    "            ]\n",
    "        }\n",
    "    }\n",
    "\n",
    "# given an array of nodes return an array of GeoJSON features\n",
    "def nodesToFeatures(nodes):\n",
    "    \"\"\"\n",
    "    :param nodes\n",
    "    :type nodes from overpy.Result (result.nodes)\n",
    "    :return:\n",
    "    \"\"\"\n",
    "    features = []\n",
    "    for node in nodes:\n",
    "        features.append(nodeToFeature(node))\n",
    "    return features\n",
    "\n",
    "def waysToFeatures(ways):\n",
    "    print ways\n",
    "    features = []\n",
    "    return features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# method to handle OverpassTooManyRequests exception from OpenStreetMap/overpass turbo API\n",
    "def retry_if_overpass_too_many_requests(exception):\n",
    "    return isinstance(exception, overpy.exception.OverpassTooManyRequests)\n",
    "\n",
    "# decorator to retry with exponential back off\n",
    "@retry(wait_exponential_multiplier=2000, \n",
    "       wait_exponential_max=60000,\n",
    "       retry_on_exception=retry_if_overpass_too_many_requests)\n",
    "def call_overpass_api(q):\n",
    "    return OVERPASS_API.query(q)\n",
    "\n",
    "def run_overpass_api(bounding_geo_df):\n",
    "    local_pois = []\n",
    "    for index, row in bounding_geo_df.iterrows():\n",
    "        # For documentation see:\n",
    "        # http://wiki.openstreetmap.org/wiki/Tag:{key}={value}\n",
    "        # e.g: http://wiki.openstreetmap.org/wiki/Tag:amenity=theatre\n",
    "        payload = \"\"\"\n",
    "            [out:json][timeout:60];\n",
    "            (\n",
    "              node[\"tourism\"=\"gallery\"]%(box)s;\n",
    "              node[\"tourism\"=\"artwork\"]%(box)s;\n",
    "              node[\"tourism\"=\"museum\"]%(box)s;\n",
    "              node[\"amenity\"=\"arts_centre\"]%(box)s;\n",
    "              node[\"amenity\"=\"theatre\"]%(box)s; \n",
    "            );\n",
    "            out body;\"\"\" % {'box': str(bbox(row.geometry.bounds))}\n",
    "        result = call_overpass_api(payload)\n",
    "        local_pois.extend(nodesToFeatures(result.nodes))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": [],
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "OVERPASS_API         = overpy.Overpass()\n",
    "BASE_DIR             = os.path.join(os.path.abspath('.'), 'work-flow')\n",
    "URBAN_BOUNDRIES_FILE = '06_Europe_Cities_Boundries_with_Labels_Population.geo.json'\n",
    "\n",
    "# Paths to base datasets that we are using:\n",
    "URBAN_BOUNDRIES_PATH = os.path.join(BASE_DIR,URBAN_BOUNDRIES_FILE)\n",
    "POIS_PATH            = os.path.join(BASE_DIR, \"pois.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# GeoPandas\n",
    "\n",
    "GeoPandas is an open source project to make working with geospatial data in python easier. GeoPandas extends the datatypes used by pandas to allow spatial operations on geometric types. Geometric operations are performed by shapely. Geopandas further depends on fiona for file access and descartes and matplotlib for plotting.\n",
    "\n",
    "The goal of GeoPandas is to make working with geospatial data in python easier. It combines the capabilities of pandas and shapely, providing geospatial operations in pandas and a high-level interface to multiple geometries to shapely. GeoPandas enables you to easily do operations in python that would otherwise require a spatial database such as PostGIS."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## OGC SimpleFeatures\n",
    "\n",
    "A standard that specifies a common storage and access model of mostly two-dimensional geographical data (point, line, polygon, multi-point, multi-line, etc.) The formats were originally defined by the Open Geospatial Consortium (OGC) and described in their Simple Feature Access and Coordinate Transformation Service specifications. \n",
    "\n",
    "The standard defines a model for two-dimensional simple features, with linear interpolation between vertices. The data model defined is a hierarchy of classes. This part also defines representation using Well-Known Text (and Binary).\n",
    "\n",
    "### WKT/WKB: Well Known Text and Well Known Binary\n",
    "\n",
    "Well-known text (WKT) is a text markup language for representing vector geometry objects on a map, spatial reference systems of spatial objects and transformations between spatial reference systems. A binary equivalent, known as well-known binary (WKB), is used to transfer and store the same information on databases, such as PostGIS, Microsoft SQL Server and DB2. The formats were originally defined by the Open Geospatial Consortium (OGC) and described in their Simple Feature Access and Coordinate Transformation Service specifications.\n",
    "\n",
    "\n",
    "### Implications for PySpark data de/serialization & un/marshalling\n",
    "\n",
    "When using pyspark we want have to send data back and forth between master node and the workers which run jobs on the JVM. In order to simplify this rather than sending Python or more precisely Shapely objects we will use WKT. It works with one the libraries I am using today to perform Spatial Joins.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Load the boundries data\n",
    "geo_df = GeoDataFrame.from_file(URBAN_BOUNDRIES_PATH)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Fetch the the POIs but not now\n",
    "overwritePois = False # <= DON'T Set to True during workshop!        \n",
    "if overwritePois or not os.path.isfile(POIS_PATH):\n",
    "    # Write POIs file\n",
    "    pois = run_overpass_api(geo_df)\n",
    "    with io.open(POIS_PATH, 'w+', encoding='utf-8') as f:\n",
    "        f.write(unicode(json.dumps(pois, ensure_ascii= False)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "solution2": "hidden",
    "solution2_first": true
   },
   "source": [
    "For the exercise below consult GeoPandas documentation http://geopandas.org/user.html\n",
    "\n",
    "## Exercise: \n",
    "\n",
    "* find out what the schema for the data is?\n",
    "* plot the geometries?\n",
    "* print out the wkt version of the geometries\n",
    "* find out the Spatial Reference System (CRS/SRS) for the GeoDataframe?\n",
    "* change the reference system to a cartesian system that is suitable for calculating areas\n",
    "* calculate the total area of the cities we are using HINT: Europe using this CRS: http://spatialreference.org/ref/epsg/3035/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "# * find out what the schema for the data is?\n",
    "geo_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "# * plot the geometries?\n",
    "plt.figure(figsize=(18, 20))\n",
    "geo_df.plot()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "# * print out the wkt version of the geometries\n",
    "wkts = map(lambda g: g.to_wkt() , geo_df.geometry)\n",
    "wkts[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "# * find out the Spatial Reference System (CRS/SRS) for the GeoDataframe?\n",
    "# A note about CRS's and Geospatial Data.\n",
    "# http://spatialreference.org/ref/epsg/etrs89-etrs-laea/\n",
    "geo_df.crs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "# * change the reference system to a cartesian system that is suitable for calculating areas\n",
    "# * calculate the total area of the cities we are using \n",
    "#   HINT: Europe using this CRS: http://spatialreference.org/ref/epsg/3035/\n",
    "geo_df_cartesian = geo_df.to_crs(epsg=3035)\n",
    "print geo_df_cartesian.crs\n",
    "print geo_df.crs\n",
    "geo_df_cartesian.area.sum()/1000000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Add a WKT column for use later\n",
    "geo_df['wkt'] = pandas.Series(\n",
    "    map(lambda geom: str(geom.to_wkt()), geo_df['geometry']), \n",
    "    index=geo_df.index, dtype='string')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a pyspark dataframe from GeoPanadas\n",
    "\n",
    "Spark Dataframes are an abstraction over RDDs. Lets look at what happens when we call pyspark transform and action. \n",
    "\n",
    "**SLIDES Switch: Spark Internals**\n",
    "\n",
    "![](images/pyspark.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.SQLContext.createDataFrame\n",
    "# Creates a DataFrame from an RDD of tuple/list, list or pandas.DataFrame.\n",
    "\n",
    "# When schema is a list of column names, the type of each column will be inferred from data.\n",
    "\n",
    "# When schema is None, it will try to infer the schema (column names and types) from data, which should\n",
    "# be an RDD of Row, or namedtuple, or dict.\n",
    "\n",
    "# If schema inference is needed, samplingRatio is used to determined the ratio of rows used for schema inference.\n",
    "# The first row will be used if samplingRatio is None.\n",
    "\n",
    "# Parameters:\n",
    "# data – an RDD of Row/tuple/list/dict, list, or pandas.DataFrame.\n",
    "# schema – a StructType or list of column names. default None.\n",
    "# samplingRatio – the sample ratio of rows used for inferring\n",
    "# Returns:\n",
    "# DataFrame\n",
    "\n",
    "boundries_from_pd = sqlContext.createDataFrame(geo_df)\n",
    "\n",
    "# Lets find Berlin\n",
    "berlin_rdd = boundries_from_pd.filter(boundries_from_pd.NAMEASCII == 'Berlin')\n",
    "wkt.loads(berlin_rdd.take(1)[0].wkt)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "solution2": "hidden",
    "solution2_first": true
   },
   "source": [
    "### Exercise \n",
    "\n",
    "* Look at the schema of the boundries_from_pd\n",
    "* Graph the name of the city and its population sorted\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "code_folding": [],
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "boundries_from_pd.printSchema()\n",
    "\n",
    "df = boundries_from_pd.select(boundries_from_pd.NAMEASCII, \n",
    "                              boundries_from_pd.POPEU2013.cast(IntegerType())\n",
    "                             )\n",
    "\n",
    "df = df.sort(df.POPEU2013.desc())\n",
    "df.show()\n",
    "\n",
    "# SQL Version\n",
    "boundries_from_pd.registerTempTable('boundries')\n",
    "sqlContext.sql(\"SELECT NAME, POPEU2013 FROM boundries\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "df = df.toPandas()\n",
    "\n",
    "df.plot.bar(x='NAMEASCII')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "source": [
    "# Lets look at the POI data next\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pois_df = sqlContext.read.json(POIS_PATH)\n",
    "\n",
    "print pois_df.count()\n",
    "rec = pois_df.take(1)[0]\n",
    "pois_df = pois_df.toPandas()\n",
    "pois_df.count()\n",
    "\n",
    "\n",
    "def toWktColumn(coords):\n",
    "    return (Point(coords).wkt)\n",
    "\n",
    "pois_df['wkt'] = pandas.Series(\n",
    "    map(lambda geom: toWktColumn(geom.coordinates), pois_df['geometry']), \n",
    "    index=pois_df.index, dtype='string')\n",
    "\n",
    "pois_df = sqlContext.createDataFrame(pois_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "solution2": "hidden",
    "solution2_first": true
   },
   "source": [
    "# Count the number of museums in Zurich\n",
    "\n",
    "## Exercise\n",
    "\n",
    "* Count the number of museums in Zurich\n",
    "HINT: You will need Shapely Spatial Operations\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "pois_df.registerTempTable(\"pois\")\n",
    "museums = sqlContext.sql(\n",
    "    \"SELECT geometry, \\\n",
    "    properties.name, \\\n",
    "    properties.tourism \\\n",
    "    FROM pois WHERE properties.tourism = 'museum'\")\n",
    "print museums.count()\n",
    "\n",
    "# select museums within Zurich\n",
    "zurich = sqlContext.sql(\"SELECT wkt, POPEU2013 \\\n",
    "    FROM boundries WHERE NAME = 'Zurich'\")\n",
    "\n",
    "jsonstr = zurich.toJSON().take(1)[0]\n",
    "jsonobj = json.loads(jsonstr)\n",
    "zurich_boundry = wkt.loads(jsonobj['wkt'])\n",
    "zurich_boundry"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "solution2": "hidden"
   },
   "outputs": [],
   "source": [
    "museums_in_zurich = museums.rdd.filter(lambda r: Point(r['geometry']['coordinates']).within(zurich_boundry))\n",
    "museums_in_zurich.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SQL is nice but...\n",
    "\n",
    "Most of the time in Spark SQL you can use Strings (in SQL) to reference columns but there are two cases where  you’ll want to use the Column objects rather than Strings :\n",
    "\n",
    "* In Spark SQL DataFrame columns are allowed to have the same name, they’ll be given unique names inside of Spark SQL, but this means that you can’t reference them with the column name only as this becomes ambiguous.\n",
    "\n",
    "* When you need to manipulate columns using expressions like Adding two columns to each other, Twice the value of this column or even Is the column value larger than 0 ?, you won’t be able to use simple strings and will need the Column reference.\n",
    "\n",
    "* Finally if you need renaming, cast or any other complex feature, you’ll need the Column reference too.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Joins, Spatial Predicate, Broadcast Variables, UDFs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Exercise\n",
    "# for each city count the number of museums\n",
    "# and return a DF with:\n",
    "# city_name, museum_count\n",
    "\n",
    "# SQL VErsion\n",
    "# cities_df = sqlContext.sql(\n",
    "#     \"\"\"\n",
    "#     SELECT properties.NAMEASCII AS city_name, \n",
    "#         geometry AS city_geom FROM boundries\n",
    "#     \"\"\")\n",
    "\n",
    "cities_df = boundries_from_pd.select(\n",
    "    (boundries_from_pd.NAMEASCII).alias('city_name'),\n",
    "    (boundries_from_pd.wkt).alias('city_geom'))\n",
    "\n",
    "cities_df.cache()\n",
    "\n",
    "# Create a broadcast variable\n",
    "# Broadcast http://spark.apache.org/docs/latest/programming-guide.html#broadcast-variables\n",
    "# Broadcast variables allow the programmer to keep a read-only variable cached on each machine \n",
    "# rather than shipping a copy of it with tasks. They can be used, for example, to give every node \n",
    "# a copy of a large input dataset in an efficient manner. Spark also attempts to distribute broadcast\n",
    "# variables using efficient broadcast algorithms to reduce communication cost.\n",
    "\n",
    "# Spark actions are executed through a set of stages, separated by distributed “shuffle” \n",
    "# operations. Spark automatically broadcasts the common data needed by tasks within each stage. \n",
    "# The data broadcasted this way is cached in serialized form and deserialized before running each task. \n",
    "# This means that explicitly creating broadcast variables is only useful when tasks across multiple \n",
    "# stages need the same data or when caching the data in deserialized form is important.\n",
    "\n",
    "# Broadcast variables are created from a variable v by calling SparkContext.broadcast(v). The broadcast \n",
    "# variable is a wrapper around v, and its value can be accessed by calling the value method.\n",
    "\n",
    "_cities_df = cities_df.toJSON().collect()[0:3]\n",
    "broadcastCitiesJSON = sc.broadcast(_cities_df)\n",
    "\n",
    "# wkt.loads(json.loads(broadcastCitiesJSON.value[0])['city_geom'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def get_city_name(poi_geom):\n",
    "    # get an array of dict [(city_name, city_geom)]\n",
    "    cities = map(lambda c: {\n",
    "                    'city_name': json.loads(c)['city_name'],\n",
    "                    'city_wkt': wkt.loads(json.loads(c)['city_geom'])\n",
    "                }, broadcastCitiesJSON.value)\n",
    "\n",
    "    shply_poi = shape(poi_geom.asDict())\n",
    "    city = filter(lambda city: shply_poi.within(city['city_wkt']), cities)\n",
    "    name = None\n",
    "    if city:\n",
    "        name = city[0]['city_name']\n",
    "    return name\n",
    "\n",
    "# http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.SQLContext.registerFunction\n",
    "# Registers a python function (including lambda function) as a UDF so it can be used in SQL statements.\n",
    "# In addition to a name and the function itself, the return type can be optionally specified. When the \n",
    "# return type is not given it default to a string and conversion will automatically be done. For any other \n",
    "# return type, the produced object must match the specified type.\n",
    "\n",
    "sqlContext.udf.register(\"get_city_name\", get_city_name, StringType())\n",
    "\n",
    "# http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.functions.udf\n",
    "# Creates a Column expression representing a user defined function (UDF).    \n",
    "get_city_name_udf = func.udf(get_city_name, StringType())\n",
    "\n",
    "# SQL VERSION\n",
    "# museums_df = sqlContext.sql(\n",
    "#     \"SELECT geometry as museum_geom, \\\n",
    "#     properties.name as museum_name, \\\n",
    "#     get_city_name(geometry) as city_name \\\n",
    "#     FROM pois WHERE properties.tourism = 'museum'\")\n",
    "\n",
    "\n",
    "museums_df = pois_df.select(\n",
    "    (pois_df.geometry).alias('museum_geom'),\n",
    "    (pois_df.properties.name).alias('museum_name'),\n",
    "    (get_city_name_udf(pois_df.geometry).alias('city_name'))\n",
    ")\n",
    "\n",
    "museums_df.registerTempTable(\"museums\")\n",
    "                       \n",
    "print museums_df.count()\n",
    "# print cities_df.count()\n",
    "# print museums_df.where(museums_df.city_name.isNotNull()).count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "museums_df.cache() # Try without and with\n",
    "print museums_df.where(museums_df.city_name.isNotNull()).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " ## Grouping "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "grouped_by_city = museums_df.dropna().groupBy('city_name')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "grouped_by_city.count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Adding population column\n",
    "Accoring to our algorithm we will have to divide count by the the population to scale it per capita. Lets try to run our algorithm on a subset of the data to get practice for the real deal!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# lets recreate the cities DF this time including the population\n",
    "# http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.Column.cast\n",
    "# Convert the column into type dataType\n",
    "cities_df = boundries_from_pd.select(\n",
    "    (boundries_from_pd.NAMEASCII).alias('city_name'),\n",
    "    (boundries_from_pd.POPEU2013.cast(IntegerType())).alias('population'),\n",
    "    (boundries_from_pd.wkt).alias('city_geom'))\n",
    "\n",
    "cities_df.cache()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "museums_df = museums_df.dropna()\n",
    "\n",
    "# http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame.join\n",
    "# Joins with another DataFrame, using the given join expression.\n",
    "\n",
    "# The following performs a full outer join between df1 and df2.\n",
    "\n",
    "# Parameters:\n",
    "# other – Right side of the join\n",
    "# on – a string for join column name, a list of column names, , a join expression (Column) or a list of Columns.\n",
    "#      If on is a string or a list of string indicating the name of the join column(s), the column(s) must exist \n",
    "#      on both sides, and this performs an equi-join.\n",
    "# how – str, default ‘inner’. One of inner, outer, left_outer, right_outer, leftsemi.\n",
    "\n",
    "df = museums_df.join(cities_df, museums_df.city_name == cities_df.city_name).select(\n",
    "    museums_df.museum_geom,\n",
    "    museums_df.museum_name,    \n",
    "    museums_df.city_name,\n",
    "    cities_df.population,\n",
    "    cities_df.city_geom\n",
    ")\n",
    "\n",
    "df.printSchema( )\n",
    "# Love Spark!\n",
    "\n",
    "grouped_by_city = df.groupBy('city_name', 'population')\n",
    "grouped_by_city = grouped_by_city.count()\n",
    "            "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Calculating the score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame.withColumnRenamed\n",
    "grouped_by_city = grouped_by_city.withColumnRenamed('count', 'museum_count')\n",
    "\n",
    "cultural_weight_lookup = { 'museum': 1, 'gallery': 2, 'artwork': 3 }\n",
    "\n",
    "# http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame.withColumn\n",
    "grouped_by_city = grouped_by_city.withColumn('cultural_score', (\n",
    "                            grouped_by_city.museum_count/grouped_by_city.population\n",
    "                           )*10000*cultural_weight_lookup['museum'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "grouped_by_city.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# understanding de/serialization\n",
    "\n",
    "from pyspark.rdd import RDD\n",
    "dl = [\n",
    "    (u'2', {u'director': u'David Lean'}), \n",
    "    (u'7', {u'director': u'Andrew Dominik'})\n",
    "]\n",
    "\n",
    "dl_rdd = sc.parallelize(dl)\n",
    "tmp = dl_rdd._to_java_object_rdd()\n",
    "tmp2 = sc._jvm.SerDe.javaToPython(tmp)\n",
    "t = RDD(tmp2, sc)\n",
    "t.count()\n",
    "\n",
    "tmp = t._to_java_object_rdd()\n",
    "tmp2 = sc._jvm.SerDe.javaToPython(tmp)\n",
    "t = RDD(tmp2, sc)\n",
    "t.count() "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spatial Spark\n",
    "\n",
    "SpatialSpark aims to provide efficient spatial operations using Apache Spark. It can be used as a Spark library for spatial extension as well as a standalone application to process large scale spatial join operations.\n",
    "\n",
    "SpatialSpark has been compiled and tested on Spark 1.6.1. For geometry operations and data structures for indexes, well known JTS library is used.\n",
    "\n",
    "## Spatial Partition\n",
    "\n",
    "Generate a spatial partition from input dataset, currently Fixed-Grid Partition (FGP), Binary-Split Partition (BSP) and Sort-Tile Partition (STP) are supported.\n",
    "\n",
    "## Spatial Range Query\n",
    "\n",
    "Spatial range query includes both indexed and non-indexed query. For non-indexed query, a full scan is performed on the dataset and returns filtered results. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from ast import literal_eval as make_tuple\n",
    "print \"Java Spark context version:\", sc._jsc.version()\n",
    "spatialspark = sc._jvm.spatialspark\n",
    "\n",
    "rectangleA = Polygon([(0, 0), (0, 10), (10, 10), (10, 0)])\n",
    "rectangleB = Polygon([(-4, -4), (-4, 4), (4, 4), (4, -4)])\n",
    "rectangleC = Polygon([(7, 7), (7, 8), (8, 8), (8, 7)])\n",
    "pointD = Point((-1, -1))\n",
    "\n",
    "def geomABWithId():\n",
    "  return sc.parallelize([\n",
    "    (0L, rectangleA.wkt),\n",
    "    (1L, rectangleB.wkt)\n",
    "  ])\n",
    "\n",
    "def geomCWithId():\n",
    "  return sc.parallelize([\n",
    "    (0L, rectangleC.wkt)\n",
    "  ])\n",
    "\n",
    "def geomABCWithId():\n",
    "  return sc.parallelize([\n",
    "  (0L, rectangleA.wkt),\n",
    "  (1L, rectangleB.wkt),\n",
    "  (2L, rectangleC.wkt)])\n",
    "\n",
    "def geomDWithId():\n",
    "  return sc.parallelize([\n",
    "    (0L, pointD.wkt)\n",
    "  ])\n",
    "\n",
    "\n",
    "dfAB = sqlContext.createDataFrame(geomABWithId(), ['id', 'wkt'])\n",
    "dfABC = sqlContext.createDataFrame(geomABCWithId(), ['id', 'wkt'])\n",
    "dfC = sqlContext.createDataFrame(geomCWithId(), ['id', 'wkt'])\n",
    "dfD = sqlContext.createDataFrame(geomDWithId(), ['id', 'wkt'])\n",
    "# Supported Operators: Within, WithinD, Contains, Intersects, Overlaps, NearestD\n",
    "SpatialOperator      = spatialspark.operator.SpatialOperator \n",
    "BroadcastSpatialJoin = spatialspark.join.BroadcastSpatialJoin\n",
    "\n",
    "joinRDD = BroadcastSpatialJoin.apply(sc._jsc, dfABC._jdf, dfAB._jdf, SpatialOperator.Within(), 0.0)\n",
    "\n",
    "joinRDD.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "results = joinRDD.collect()\n",
    "map(lambda result: make_tuple(result.toString()), results)\n",
    "\n",
    "# [(0, 0), (1, 1), (2, 0)] read as:\n",
    "# ID 0 is within 0\n",
    "# ID 1 is within 1\n",
    "# ID 2 is within 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Exercise: Find out if geometry with ID 2 overlaps geometry with ID 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Perform Analysis on All Features and All Cities\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import monotonicallyIncreasingId\n",
    "\n",
    "# create dataframe with (id, geometry) for POIs\n",
    "# 1. Add and ID Column to POIs\n",
    "\n",
    "pois_df = pois_df.withColumn(\"id\", monotonicallyIncreasingId())\n",
    "pois_tuple_id_wkt = pois_df.select(pois_df.id, pois_df.wkt)\n",
    "pois_tuple_id_wkt.show()\n",
    "pois_tuple_id_wkt.printSchema()\n",
    "print pois_tuple_id_wkt.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "boundries_from_pd = boundries_from_pd.withColumn(\"id\", monotonicallyIncreasingId())\n",
    "boundries_tuple_id_wkt = boundries_from_pd.select(boundries_from_pd.id, boundries_from_pd.wkt)\n",
    "boundries_tuple_id_wkt.printSchema()\n",
    "print boundries_from_pd.count()\n",
    "boundries_tuple_id_wkt.show()\n",
    "wkt.loads(boundries_tuple_id_wkt.take(1)[0].wkt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "joinPoiBdryRDD = BroadcastSpatialJoin.apply(sc._jsc, \n",
    "                                            pois_tuple_id_wkt._jdf, \n",
    "                                            boundries_tuple_id_wkt._jdf, \n",
    "                                            SpatialOperator.Within(), \n",
    "                                            0.0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print joinPoiBdryRDD.count()\n",
    "\n",
    "joinResults = map(lambda result: make_tuple(result.toString()), joinPoiBdryRDD.collect())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print pois_tuple_id_wkt.count()\n",
    "print boundries_tuple_id_wkt.count()\n",
    "print joinResults[695]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# make the results a DF\n",
    "rddResult = sc.parallelize(joinResults)\n",
    "df = sqlContext.createDataFrame(rddResult, [\"poi_id\", \"boundry_id\"])\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Do a join with poi df and bdry df\n",
    "df_with_pois = df.join(pois_df, df.poi_id == pois_df.id).select(\n",
    "    df.poi_id,\n",
    "    df.boundry_id,\n",
    "    pois_df.properties.alias(\"poi_properties\"),    \n",
    "    pois_df.wkt.alias(\"poi_wkt\")    \n",
    "\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_with_pois.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_with_pois_bdrys = df_with_pois.join(boundries_from_pd, \n",
    "                                       df_with_pois.boundry_id == boundries_from_pd.id).select(\n",
    "    df_with_pois.poi_id,\n",
    "    df_with_pois.boundry_id,\n",
    "    df_with_pois.poi_properties,\n",
    "    df_with_pois.poi_wkt,\n",
    "    boundries_from_pd.wkt.alias(\"boundry_wkt\"),\n",
    "    boundries_from_pd.NAMEASCII.alias(\"city_name\"),\n",
    "    boundries_from_pd.POPEU2013.alias(\"population\")    \n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_with_pois_bdrys.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_with_pois_bdrys.cache()\n",
    "df_with_pois_bdrys.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "df_with_pois_bdrys.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "rec = df_with_pois_bdrys.take(1)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print rec.poi_properties.amenity or rec.poi_properties.tourism"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "df_with_pois_bdrys = df_with_pois_bdrys.withColumn(\n",
    "        'tourism', df_with_pois_bdrys.poi_properties.tourism\n",
    "    ).withColumn(\n",
    "        'amenity', df_with_pois_bdrys.poi_properties.amenity)\n",
    "\n",
    "df_with_pois_bdrys.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create an expression for a column that calculates the score per record\n",
    "\n",
    "# df_with_pois_bdrys.groupby(\"poi_properties.touris\")\n",
    "\n",
    "df = df_with_pois_bdrys.select('*', func.coalesce(\n",
    "        df_with_pois_bdrys.poi_properties.tourism, df_with_pois_bdrys.poi_properties.amenity\n",
    "    ).alias(\"location_type\")).groupby('boundry_id', 'city_name', 'location_type', 'population').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def get_cultural_score(location_type, count, population):\n",
    "    cultural_weight_lookup = { u'museum': 1.0, u'arts_centre': 2.0, u'theatre': 3.0, u'gallery': 4.0, u'artwork': 5.0 }\n",
    "    wgt = cultural_weight_lookup.get(location_type, 0.0)\n",
    "    return float((wgt* float(count) * 100000)/float(population))\n",
    "\n",
    "sqlContext.registerFunction(\"get_cultural_score\", get_cultural_score, FloatType())\n",
    "\n",
    "# score_udf = func.udf(get_cultural_score, FloatType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# score_df = df.select(df.boundry_id, \n",
    "#           df.city_name, \n",
    "#           df.population,\n",
    "#           df.location_type,\n",
    "#           df.count,\n",
    "#           score_udf(df.location_type, df.count, df.population).alias('cultural_score')\n",
    "#          )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.registerTempTable(\"cultural_score\")\n",
    "\n",
    "score_df = sqlContext.sql(\n",
    "    \"SELECT boundry_id, \\\n",
    "        city_name, \\\n",
    "        location_type, \\\n",
    "        population, \\\n",
    "        count, get_cultural_score(location_type, count, population) as score \\\n",
    "    FROM cultural_score\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "score_df.sort(score_df.city_name.asc()).show()\n",
    "# score_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df = score_df.groupBy(\"boundry_id\", \"city_name\", \"population\").agg(func.sum(score_df.score)).sort(\"city_name\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df= df.withColumnRenamed(\"sum(score)\",\"final_score\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pd = df.toPandas()\n",
    "pd.sort_values(\"final_score\", ascending=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "pd_df = df_with_pois_bdrys.toPandas()\n",
    "\n",
    "pd_df.head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pd_df = pd_df[['boundry_id', 'boundry_wkt', 'city_name', 'population']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "pd_df = pd_df.drop_duplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "geometry = [wkt.loads(boundry_wkt) for boundry_wkt in pd_df.boundry_wkt]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "geodf = GeoDataFrame(pd_df, geometry=geometry)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "geodf.head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "scores = pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "scores.head()\n",
    "scores_merged_df=pandas.DataFrame.merge(geodf, scores, on='boundry_id')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "scores_merged_df.head(1)\n",
    "geo_scores_merged = GeoDataFrame(scores_merged_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import folium\n",
    "map_osm = folium.Map(location=[47.19094, 11.98566], \n",
    "    tiles='Mapbox Bright',\n",
    "    zoom_start=6)\n",
    "\n",
    "map_osm.choropleth(geo_str=geo_scores_merged.to_json(),\n",
    "              data=geo_scores_merged,\n",
    "              columns=['city_name_x', 'final_score'],\n",
    "              fill_color='RdBu',\n",
    "              key_on='properties.city_name_x')\n",
    "map_osm"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "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.11"
  },
  "toc": {
   "toc_cell": true,
   "toc_number_sections": true,
   "toc_section_display": "block",
   "toc_threshold": 6,
   "toc_window_display": true
  },
  "toc_position": {
   "height": "594px",
   "left": "1531.83px",
   "right": "20px",
   "top": "158px",
   "width": "327px"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}