create_all_static_data_dfs.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "source": [
    "from pyspark.sql.types import *\n",
    "import pyspark.sql.functions as F\n",
    "from elasticsearch import Elasticsearch, helpers\n",
    "from pyspark.sql.functions import col, split, randn\n",
    "from pyspark.ml import Pipeline\n",
    "import pandas as pd \n",
    "from pyspark.sql.window import Window\n",
    "from collections import defaultdict"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "Imports ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "bb7eff79-4807-4d28-b9a6-a0dc8f2eaffc"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# link for working with dict of dicts (primary and secondary keys- http://python.omics.wiki/data-structures/dictionary/multiple-keys "
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "                                    create lines & stops table: T[src][dest] = [lineNum & direction] ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "ce9695d8-af88-46db-ac0b-6bc79f069c58"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "schema = StructType([\n",
    "                    StructField(\"ID\",LongType(),True),\n",
    "                    StructField(\"ShapeId\",StringType(),True),\n",
    "                    StructField(\"Operator\",StringType(),True),\n",
    "                    StructField(\"StopSequence\",LongType(),True),\n",
    "                    StructField(\"RouteName\",StringType(),True),\n",
    "                    StructField(\"Direction\",StringType(),True),\n",
    "                    StructField(\"PlateCode\",LongType(),True),\n",
    "                    StructField(\"RouteData\",StringType() ,True)])   \n",
    "\n",
    "\n",
    "df = spark.read.csv('/data/lines_and_stops.csv', header=True, schema=schema)\n",
    "df = df[['StopSequence', 'RouteName', \"Direction\", \"PlateCode\", \"RouteData\"]]\n",
    "df = df.withColumnRenamed('RouteName', 'lineNum').withColumnRenamed('PlateCode', 'busStop').withColumnRenamed('RouteData', 'allLines')\n",
    "df = df.dropDuplicates().sort('lineNum', 'StopSequence')\n",
    "df = df.na.drop(how=\"any\")\n",
    "df = df[df.busStop != 0]\n",
    "# display(df)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "Load lines_and_stops CSV ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "908b6158-29f6-4747-a8e8-0067ccc2e8f8"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "df",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "StopSequence",
            "nullable": true,
            "type": "long"
           },
           {
            "metadata": {},
            "name": "lineNum",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Direction",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "busStop",
            "nullable": true,
            "type": "long"
           },
           {
            "metadata": {},
            "name": "allLines",
            "nullable": true,
            "type": "string"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "stops_per_line = df.groupBy(\"lineNum\", \"Direction\").agg(F.collect_list(\"busStop\")).sort(F.col(\"lineNum\")).withColumnRenamed('collect_list(busStop)', 'allStops')\n",
    "# display(stops_per_line)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "group by lineNum & direction to get a sorted stops list ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "1750a356-68d9-46f5-a779-fab9103b487b"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "stops_per_line",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "lineNum",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Direction",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "allStops",
            "nullable": true,
            "type": {
             "containsNull": true,
             "elementType": "long",
             "type": "array"
            }
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# display(stops_per_line[stops_per_line.lineNum==16])"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "Sanety check for one line ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "c57197ef-2bd1-403b-9685-a3d1a747d527"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "src_dest_lines = defaultdict(dict)\n",
    "for row in stops_per_line.rdd.collect():\n",
    "  line = row.lineNum \n",
    "  direction = row.Direction \n",
    "  allStops = row.allStops\n",
    "  for src, dest in zip(allStops, allStops[1:]):\n",
    "    if not (( src in src_dest_lines.keys() ) & ( dest in src_dest_lines[src].keys() )): \n",
    "      src_dest_lines[src][dest] = set()\n",
    "    src_dest_lines[src][dest].add(tuple((line, direction)))"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "Create line & stop table ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "134a8245-4c57-4a5e-ac6b-cd7c1d7f5c43"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# for key, value in src_dest_lines.items(): \n",
    "#   print(key, value) "
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "Sanety check ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "6d349b40-2b7c-4492-84e9-4cba59af0be3"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "                                                             create att_table: {dest:[(a,r,c), (a2, r2, c2),…,] , dest2: [(a,r,c), (),..], …}",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "60254cff-9d1f-49f0-aace-a16c04008abe"
    }
   },
   "outputs": [],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "schema = StructType([\n",
    "                    StructField(\"loc_name\",StringType(),True),\n",
    "                    StructField(\"rating_name\",StringType(),True),\n",
    "                    StructField(\"category\",StringType(),True),\n",
    "                    StructField(\"Longitude\",DoubleType(),True),\n",
    "                    StructField(\"Latitude\",DoubleType(),True),\n",
    "                    StructField(\"agg_score\",DoubleType(),True),\n",
    "                    StructField(\"rating\",DoubleType(),True)])\n",
    "\n",
    "attractions_df = spark.read.csv('/data/rating_loc_attractions.csv', header=True)#, schema=schema)\n",
    "\n",
    "attractions_df = attractions_df[['loc_name', \"Longitude\", 'Latitude','rating','category']]\n",
    "attractions_with_rankings=attractions_df.withColumnRenamed('loc_name','name')\n",
    "# display(attractions_with_rankings) "
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "read attractions data",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "ff306f27-7b7a-4f4c-94e4-18d876efb2d5"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "attractions_df",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "loc_name",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Longitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Latitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "rating",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "category",
            "nullable": true,
            "type": "string"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        },
        {
         "name": "attractions_with_rankings",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "name",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Longitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Latitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "rating",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "category",
            "nullable": true,
            "type": "string"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "stops_df = spark.read.csv('/data/stops.txt',header=True)\n",
    "stops_df=stops_df.withColumn(\"busStop\", split(F.col(\"stop_name\"), \",\")[1])\n",
    "stops_df=stops_df.withColumn(\"busStop\", split(F.col(\"busStop\"), \" \"))\n",
    "stops_df = stops_df.filter(F.col(\"busStop\")[2] != \"No.\")\n",
    "stops_df=stops_df.withColumn(\"busStop\",F.col(\"busStop\")[2])\n",
    "stops_df=stops_df[[\"stop_lat\",\"stop_lon\",\"busStop\"]]\n",
    "stops_df = stops_df.withColumnRenamed('stop_lat', 'latitude').withColumnRenamed('stop_lon', 'longitude')\n",
    "# display(stops_df)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "read stops data",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "4a20ddcc-ac33-473f-9d0b-6e54af441de9"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "stops_df",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "latitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "longitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "busStop",
            "nullable": true,
            "type": "string"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "stops_df.createOrReplaceTempView('bus_stops_names')\nattractions_with_rankings.createOrReplaceTempView('attractions_rank_df')\nattractions_with_busStops=sqlContext.sql(\"select A.name, A.Longitude, A.Latitude,A.rating, A.category,B.busStop, abs(A.Longitude-B.longitude) as long_dist, abs(A.Latitude-B.latitude) as lat_dist from attractions_rank_df as A left join bus_stops_names as B on abs(A.Longitude-B.longitude)<=0.00700 and abs(A.Latitude-B.latitude)<=0.00700\")\n\n#calc distance and keep 3 nearest stops\nattractions_with_busStops=attractions_with_busStops.withColumn(\"distance\",F.sqrt(F.pow(F.col(\"long_dist\"),2)+F.pow(F.col(\"lat_dist\"),2)))\n\nwindow = Window.partitionBy(attractions_with_busStops['Name']).orderBy(attractions_with_busStops[\"distance\"])\nclosest_stops_df=attractions_with_busStops.select('*', F.rank().over(window).alias('rank')).filter(F.col('rank') <= 3)\nclosest_stops_df=closest_stops_df[\"name\",\"rating\",\"category\",\"busStop\"]"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "get 3 closest bus stop for each attraction",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "66a6d99c-592f-4211-a822-ee02839c4a5b"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "attractions_with_busStops",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "name",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Longitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "Latitude",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "rating",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "category",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "busStop",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "long_dist",
            "nullable": true,
            "type": "double"
           },
           {
            "metadata": {},
            "name": "lat_dist",
            "nullable": true,
            "type": "double"
           },
           {
            "metadata": {},
            "name": "distance",
            "nullable": true,
            "type": "double"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        },
        {
         "name": "closest_stops_df",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "name",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "rating",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "category",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "busStop",
            "nullable": true,
            "type": "string"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "closest_stops_df=closest_stops_df.dropna()"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "9382766c-321c-481a-ab46-796495cfc7a1"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "closest_stops_df",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "name",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "rating",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "category",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "busStop",
            "nullable": true,
            "type": "string"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# display(closest_stops_df)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "130a3294-853f-4c61-9564-4d37f8d00e41"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "att_dict = defaultdict(list)\nfor row in closest_stops_df.rdd.collect():\n  dest = row.busStop\n  attraction = row.name \n  rank = row.rating\n  category = row.category\n  att_dict[dest].append((attraction,rank,category))\n  \n# Initializing N  \nN = 2\nfor key, value in att_dict.items():\n  res = sorted(value, key = lambda x: x[1], reverse = True)[:N] \n  att_dict[key]=res"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "create att_table",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "69f7ca46-b8c8-40dd-acee-5cbe5a0e1dae"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# for key, value in att_dict.items(): \n#   print(key, value) "
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "f2a44513-51ed-483c-bb11-e5d26028f393"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# for key, value in src_dest_lines.items(): \n#   print(type(key), type(value.keys())) "
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "841c1008-d1ec-4857-802f-7034883cb1ea"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    ""
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "                                              create table_ard=  Si : { (a1, r1, d1):[1, 11, 34, 5], (a2, r2, d2):[],....} ",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "4235c9b1-22a3-4d97-93d7-6146e60501ea"
    }
   },
   "outputs": [],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "table_ard = defaultdict(dict) \nfor S in src_dest_lines.keys():\n  temp = {}\n  for D in src_dest_lines[S].keys():\n    if att_dict.get(str(D)):\n      for attraction in att_dict[str(D)]:\n        a,r,c=attraction\n        key = (a,r,D,c)\n        temp[key] = list(src_dest_lines[S][D]) # this value is list of lines \n  res = {key: temp[key] for key in sorted(temp.keys(), key=lambda ele: ele[1])}\n  if res:\n    table_ard[S] =  res\n"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "e6803ab0-0005-4c65-933e-7b988d08d3f9"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "# for key, value in table_ard.items(): \n#   print(key, value) "
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "008f5dd1-6cd3-4b84-aa5a-67a29182becf"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    ""
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "                                                                   create Si: {line:(a,dest) ... }",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "7c3577db-41e4-4e2e-a787-e60c137db653"
    }
   },
   "outputs": [],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "Ready =  defaultdict(dict) \nfor S in table_ard: \n  for key, value in table_ard[S].items(): \t# key=(a1, r1, d1), value=[lines ] #check if the keys are still sorted by rank. If not- think again \n    for line in value:\n      if line not in Ready.keys():\n        Ready[S][line[0]] = (key[0], key[3],key[2]) \n"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "bfd95612-039a-4de1-98cd-3524162a31a6"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "Ready_values=list(Ready.values())\nReady_keys=list(Ready.keys())\n\nsourceStop=[source  for source, inner_dict in zip(Ready_keys,Ready_values) for line in list(inner_dict.keys())]\nline_nums=[line  for inner_dict in Ready_values for line in list(inner_dict.keys())]\n\nattractions=[data[0]  for inner_dict in Ready_values for data in list(inner_dict.values())]\ncatagorys=[data[1]  for inner_dict in Ready_values for data in list(inner_dict.values())]\ndests=[data[2]  for inner_dict in Ready_values for data in list(inner_dict.values())]\n"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "70474f18-5b92-4f05-aa65-b0a587223d37"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "data = {'sourceStop':  sourceStop,\n        'line_num': line_nums,\n        'attraction': attractions,\n       \"catagory\":catagorys,\n       \"destStop\":dests}\natrractions_final_df = pd.DataFrame (data, columns = ['sourceStop','line_num','attraction',\"catagory\",\"destStop\"])\natrractions_final_df=spark.createDataFrame(atrractions_final_df)\n# display(atrractions_final_df)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "20d06aea-75e6-462d-957b-48c888927ae5"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "datasetInfos": [
        {
         "name": "atrractions_final_df",
         "typeStr": "pyspark.sql.dataframe.DataFrame",
         "schema": {
          "fields": [
           {
            "metadata": {},
            "name": "sourceStop",
            "nullable": true,
            "type": "long"
           },
           {
            "metadata": {},
            "name": "line_num",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "attraction",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "catagory",
            "nullable": true,
            "type": "string"
           },
           {
            "metadata": {},
            "name": "destStop",
            "nullable": true,
            "type": "long"
           }
          ],
          "type": "struct"
         },
         "tableIdentifier": null
        }
       ],
       "data": "<div class=\"ansiout\"></div>",
       "removedWidgets": [],
       "addedWidgets": {},
       "type": "html",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>\n<div class=\"ansiout\"></div>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "atrractions_final_df.write.csv('atrractions_with_busStop.csv',header=True)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "98f3b5e8-6e3f-4a90-9a10-52e97ded1569"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "type": "ipynbError",
       "data": "",
       "errorSummary": "",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>"
      ]
     }
    }
   ],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    ""
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "                                                                      sanety check",
     "showTitle": true,
     "inputWidgets": {},
     "nuid": "49b5d745-b273-4e88-aab0-a873312a24ef"
    }
   },
   "outputs": [],
   "execution_count": 0
  },
  {
   "cell_type": "code",
   "source": [
    "df3=spark.read.csv('atrractions_with_busStop.csv',header=True)\n# display(df3)"
   ],
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "title": "",
     "showTitle": false,
     "inputWidgets": {},
     "nuid": "4e9d23e0-0d2e-430c-9e38-15cc2fc5f9a2"
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "type": "ipynbError",
       "data": "",
       "errorSummary": "",
       "arguments": {}
      }
     },
     "data": {
      "text/html": [
       "<style scoped>\n  .ansiout {\n    display: block;\n    unicode-bidi: embed;\n    white-space: pre-wrap;\n    word-wrap: break-word;\n    word-break: break-all;\n    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n    font-size: 13px;\n    color: #555;\n    margin-left: 4px;\n    line-height: 19px;\n  }\n</style>"
      ]
     }
    }
   ],
   "execution_count": 0
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "notebookName": "sorce_dest_lines",
   "dashboards": [],
   "language": "python",
   "widgets": {},
   "notebookOrigID": 2483473424244050
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}