notebooks/Users/denny@databricks.com/flights/On-Time Flight Performance (Spark 2.0).ipynb

{"cells":[{"cell_type":"markdown","source":["*If you see ![](http://training.databricks.com/databricks_guide/ImportNotebookIcon3.png) at the top-left or top-right, click on the link to import this notebook in order to run it.*"],"metadata":{}},{"cell_type":"markdown","source":["# On-Time Flight Performance with GraphFrames for Apache Spark\nThis notebook provides an analysis of On-Time Flight Performance and Departure Delays data using GraphFrames for Apache Spark.  This notebook has been updated to use Apache Spark 2.0 and GraphFrames 0.3.\n* Original blog post: [On-Time Flight Performance with GraphFrames with Apache Spark Blog Post](https://databricks.com/blog/2016/03/16/on-time-flight-performance-with-graphframes-for-apache-spark.html)\n* Original Notebook: [On-Time Flight Performance with GraphFrames with Apache Spark Notebook](http://cdn2.hubspot.net/hubfs/438089/notebooks/Samples/Miscellaneous/On-Time_Flight_Performance.html)\n\n\nSource Data: \n* [OpenFlights: Airport, airline and route data](http://openflights.org/data.html)\n* [United States Department of Transportation: Bureau of Transportation Statistics (TranStats)](http://www.transtats.bts.gov/DL_SelectFields.asp?Table_ID=236&DB_Short_Name=On-Time)\n * Note, the data used here was extracted from the US DOT:BTS between 1/1/2014 and 3/31/2014*\n\nReferences:\n* [GraphFrames User Guide](http://graphframes.github.io/user-guide.html)\n* [GraphFrames: DataFrame-based Graphs (GitHub)](https://github.com/graphframes/graphframes)\n* [D3 Airports Example](http://mbostock.github.io/d3/talk/20111116/airports.html)\n* [MLlib and Machine Learning: Binary Classification](https://docs.databricks.com/spark/latest/mllib/binary-classification-mllib-pipelines.html)"],"metadata":{}},{"cell_type":"markdown","source":["### Preparation\nExtract the Airports and Departure Delays information from S3 / DBFS"],"metadata":{}},{"cell_type":"code","source":["%fs ls /databricks-datasets/flights/"],"metadata":{},"outputs":[],"execution_count":4},{"cell_type":"code","source":["# Set File Paths\ntripdelaysFilePath = \"/databricks-datasets/flights/departuredelays.csv\"\nairportsnaFilePath = \"/databricks-datasets/flights/airport-codes-na.txt\"\n\n# Obtain airports dataset\nairportsna = spark.read.csv(airportsnaFilePath, header='true', inferSchema='true', sep='\\t')\nairportsna.createOrReplaceTempView(\"airports_na\")\n\n# Obtain departure Delays data\ndepartureDelays = spark.read.csv(tripdelaysFilePath, header='true')\ndepartureDelays.createOrReplaceTempView(\"departureDelays\")\ndepartureDelays.cache()\n\n# Available IATA codes from the departuredelays sample dataset\ntripIATA = spark.sql(\"select distinct iata from (select distinct origin as iata from departureDelays union all select distinct destination as iata from departureDelays) a\")\ntripIATA.createOrReplaceTempView(\"tripIATA\")\n\n# Only include airports with atleast one trip from the departureDelays dataset\nairports = spark.sql(\"select f.IATA, f.City, f.State, f.Country from airports_na f join tripIATA t on t.IATA = f.IATA\")\nairports.createOrReplaceTempView(\"airports\")\nairports.cache()"],"metadata":{},"outputs":[],"execution_count":5},{"cell_type":"code","source":["departureDelays.count()"],"metadata":{},"outputs":[],"execution_count":6},{"cell_type":"code","source":["# Build `departureDelays_geo` DataFrame\n#  Obtain key attributes such as Date of flight, delays, distance, and airport information (Origin, Destination)  \ndepartureDelays_geo = spark.sql(\"select cast(f.date as int) as tripid, cast(concat(concat(concat(concat(concat(concat('2014-', concat(concat(substr(cast(f.date as string), 1, 2), '-')), substr(cast(f.date as string), 3, 2)), ' '), substr(cast(f.date as string), 5, 2)), ':'), substr(cast(f.date as string), 7, 2)), ':00') as timestamp) as `localdate`, cast(f.delay as int), cast(f.distance as int), f.origin as src, f.destination as dst, o.city as city_src, d.city as city_dst, o.state as state_src, d.state as state_dst from departuredelays f join airports o on o.iata = f.origin join airports d on d.iata = f.destination\") \n\n# Create Temporary View and cache\ndepartureDelays_geo.createOrReplaceTempView(\"departureDelays_geo\")\ndepartureDelays_geo.cache()\n\n# Count\ndepartureDelays_geo.count()"],"metadata":{},"outputs":[],"execution_count":7},{"cell_type":"code","source":["# Review the top 10 rows of the `departureDelays_geo` DataFrame\ndepartureDelays_geo.show(10)"],"metadata":{},"outputs":[],"execution_count":8},{"cell_type":"code","source":["# Using `display` to view the data\ndisplay(departureDelays_geo)"],"metadata":{},"outputs":[],"execution_count":9},{"cell_type":"markdown","source":["## Building the Graph\nNow that we've imported our data, we're going to need to build our graph. To do so we're going to do two things: we are going to build the structure of the vertices (or nodes) and we're going to build the structure of the edges. What's awesome about GraphFrames is that this process is incredibly simple. \n* Rename IATA airport code to **id** in the Vertices Table\n* Start and End airports to **src** and **dst** for the Edges Table (flights)\n\nThese are required naming conventions for vertices and edges in GraphFrames as of the time of this writing (Feb. 2016)."],"metadata":{}},{"cell_type":"markdown","source":["**WARNING:** If the graphframes package, required in the cell below, is not installed, follow the instructions [here](http://cdn2.hubspot.net/hubfs/438089/notebooks/help/Setup_graphframes_package.html)."],"metadata":{}},{"cell_type":"code","source":["# Note, ensure you have already installed the GraphFrames spack-package\nfrom pyspark.sql.functions import *\nfrom graphframes import *\n\n# Create Vertices (airports) and Edges (flights)\ntripVertices = airports.withColumnRenamed(\"IATA\", \"id\").distinct()\ntripEdges = departureDelays_geo.select(\"tripid\", \"delay\", \"src\", \"dst\", \"city_dst\", \"state_dst\")\n\n# Cache Vertices and Edges\ntripEdges.cache()\ntripVertices.cache()"],"metadata":{},"outputs":[],"execution_count":12},{"cell_type":"code","source":["# Vertices\n#   The vertices of our graph are the airports\ndisplay(tripVertices)"],"metadata":{},"outputs":[],"execution_count":13},{"cell_type":"code","source":["# Edges\n#  The edges of our graph are the flights between airports\ndisplay(tripEdges)"],"metadata":{},"outputs":[],"execution_count":14},{"cell_type":"code","source":["# Build `tripGraph` GraphFrame\n#  This GraphFrame builds up on the vertices and edges based on our trips (flights)\ntripGraph = GraphFrame(tripVertices, tripEdges)\n\n# Build `tripGraphPrime` GraphFrame\n#   This graphframe contains a smaller subset of data to make it easier to display motifs and subgraphs (below)\ntripEdgesPrime = departureDelays_geo.select(\"tripid\", \"delay\", \"src\", \"dst\")\ntripGraphPrime = GraphFrame(tripVertices, tripEdgesPrime)"],"metadata":{},"outputs":[],"execution_count":15},{"cell_type":"markdown","source":["## Simple Queries\nLet's start with a set of simple graph queries to understand flight performance and departure delays"],"metadata":{}},{"cell_type":"markdown","source":["#### Determine the number of airports and trips"],"metadata":{}},{"cell_type":"code","source":["print \"Airports: %d\" % tripGraph.vertices.count()\nprint \"Trips: %d\" % tripGraph.edges.count()\n"],"metadata":{},"outputs":[],"execution_count":18},{"cell_type":"markdown","source":["#### Determining the longest delay in this dataset"],"metadata":{}},{"cell_type":"code","source":["tripGraph.edges.groupBy().max(\"delay\").show()"],"metadata":{},"outputs":[],"execution_count":20},{"cell_type":"code","source":["# Finding the longest Delay\nlongestDelay = tripGraph.edges.groupBy().max(\"delay\")\ndisplay(longestDelay)"],"metadata":{},"outputs":[],"execution_count":21},{"cell_type":"markdown","source":["#### Determining the number of delayed vs. on-time / early flights"],"metadata":{}},{"cell_type":"code","source":["# Determining number of on-time / early flights vs. delayed flights\nprint \"On-time / Early Flights: %d\" % tripGraph.edges.filter(\"delay <= 0\").count()\nprint \"Delayed Flights: %d\" % tripGraph.edges.filter(\"delay > 0\").count()"],"metadata":{},"outputs":[],"execution_count":23},{"cell_type":"markdown","source":["#### What flights departing SEA are most likely to have significant delays\nNote, delay can be <= 0 meaning the flight left on time or early"],"metadata":{}},{"cell_type":"code","source":["tripGraph.edges\\\n  .filter(\"src = 'SEA' and delay > 0\")\\\n  .groupBy(\"src\", \"dst\")\\\n  .avg(\"delay\")\\\n  .sort(desc(\"avg(delay)\"))\\\n  .show(5)\n  "],"metadata":{},"outputs":[],"execution_count":25},{"cell_type":"code","source":["display(tripGraph.edges.filter(\"src = 'SEA' and delay > 0\").groupBy(\"src\", \"dst\").avg(\"delay\").sort(desc(\"avg(delay)\")))"],"metadata":{},"outputs":[],"execution_count":26},{"cell_type":"markdown","source":["#### What destinations tend to have delays"],"metadata":{}},{"cell_type":"code","source":["# After displaying tripDelays, use Plot Options to set `state_dst` as a Key.\ntripDelays = tripGraph.edges.filter(\"delay > 0\")\ndisplay(tripDelays)"],"metadata":{},"outputs":[],"execution_count":28},{"cell_type":"markdown","source":["#### What destinations tend to have significant delays departing from SEA"],"metadata":{}},{"cell_type":"code","source":["# States with the longest cumulative delays (with individual delays > 100 minutes) (origin: Seattle)\ndisplay(tripGraph.edges.filter(\"src = 'SEA' and delay > 100\"))"],"metadata":{},"outputs":[],"execution_count":30},{"cell_type":"markdown","source":["## Vertex Degrees\n* `inDegrees`: Incoming connections to the airport\n* `outDegrees`: Outgoing connections from the airport \n* `degrees`: Total connections to and from the airport\n\nReviewing the various properties of the property graph to understand the incoming and outgoing connections between airports."],"metadata":{}},{"cell_type":"code","source":["# Degrees\n#  The number of degrees - the number of incoming and outgoing connections - for various airports within this sample dataset\ndisplay(tripGraph.degrees.sort(desc(\"degree\")).limit(20))"],"metadata":{},"outputs":[],"execution_count":32},{"cell_type":"code","source":["# inDegrees\n#  The number of degrees - the number of incoming connections - for various airports within this sample dataset\ndisplay(tripGraph.inDegrees.sort(desc(\"inDegree\")).limit(20))"],"metadata":{},"outputs":[],"execution_count":33},{"cell_type":"code","source":["# outDegrees\n#  The number of degrees - the number of outgoing connections - for various airports within this sample dataset\ndisplay(tripGraph.outDegrees.sort(desc(\"outDegree\")).limit(20))"],"metadata":{},"outputs":[],"execution_count":34},{"cell_type":"markdown","source":["## City / Flight Relationships through Motif Finding\nTo more easily understand the complex relationship of city airports and their flights with each other, we can use motifs to find patterns of airports (i.e. vertices) connected by flights (i.e. edges). The result is a DataFrame in which the column names are given by the motif keys."],"metadata":{}},{"cell_type":"markdown","source":["#### What delays might we blame on SFO"],"metadata":{}},{"cell_type":"code","source":["# Using tripGraphPrime to more easily display \n#   - The associated edge (ab, bc) relationships \n#   - With the different the city / airports (a, b, c) where SFO is the connecting city (b)\n#   - Ensuring that flight ab (i.e. the flight to SFO) occured before flight bc (i.e. flight leaving SFO)\n#   - Note, TripID was generated based on time in the format of MMDDHHMM converted to int\n#       - Therefore bc.tripid < ab.tripid + 10000 means the second flight (bc) occured within approx a day of the first flight (ab)\n# Note: In reality, we would need to be more careful to link trips ab and bc.\nmotifs = tripGraphPrime.find(\"(a)-[ab]->(b); (b)-[bc]->(c)\")\\\n  .filter(\"(b.id = 'SFO') and (ab.delay > 500 or bc.delay > 500) and bc.tripid > ab.tripid and bc.tripid < ab.tripid + 10000\")\ndisplay(motifs)"],"metadata":{},"outputs":[],"execution_count":37},{"cell_type":"markdown","source":["## Determining Airport Ranking using PageRank\nThere are a large number of flights and connections through these various airports included in this Departure Delay Dataset.  Using the `pageRank` algorithm, Spark iteratively traverses the graph and determines a rough estimate of how important the airport is."],"metadata":{}},{"cell_type":"code","source":["# Determining Airport ranking of importance using `pageRank`\nranks = tripGraph.pageRank(resetProbability=0.15, maxIter=5)\ndisplay(ranks.vertices.orderBy(ranks.vertices.pagerank.desc()).limit(10))"],"metadata":{},"outputs":[],"execution_count":39},{"cell_type":"markdown","source":["## Most popular flights (single city hops)\nUsing the `tripGraph`, we can quickly determine what are the most popular single city hop flights"],"metadata":{}},{"cell_type":"code","source":["# Determine the most popular flights (single city hops)\nimport pyspark.sql.functions as func\ntopTrips = tripGraph \\\n  .edges \\\n  .groupBy(\"src\", \"dst\") \\\n  .agg(func.count(\"delay\").alias(\"trips\")) "],"metadata":{},"outputs":[],"execution_count":41},{"cell_type":"code","source":["# Show the top 20 most popular flights (single city hops)\ndisplay(topTrips.orderBy(topTrips.trips.desc()).limit(10))"],"metadata":{},"outputs":[],"execution_count":42},{"cell_type":"markdown","source":["## Top Transfer Cities\nMany airports are used as transfer points instead of the final Destination.  An easy way to calculate this is by calculating the ratio of inDegree (the number of flights to the airport) / outDegree (the number of flights leaving the airport).  Values close to 1 may indicate many transfers, whereas values < 1 indicate many outgoing flights and > 1 indicate many incoming flights.  Note, this is a simple calculation that does not take into account of timing or scheduling of flights, just the overall aggregate number within the dataset."],"metadata":{}},{"cell_type":"code","source":["# Calculate the inDeg (flights into the airport) and outDeg (flights leaving the airport)\ninDeg = tripGraph.inDegrees\noutDeg = tripGraph.outDegrees\n\n# Calculate the degreeRatio (inDeg/outDeg)\ndegreeRatio = inDeg.join(outDeg, inDeg.id == outDeg.id) \\\n  .drop(outDeg.id) \\\n  .selectExpr(\"id\", \"double(inDegree)/double(outDegree) as degreeRatio\") \\\n  .cache()\n\n# Join back to the `airports` DataFrame (instead of registering temp table as above)\nnonTransferAirports = degreeRatio.join(airports, degreeRatio.id == airports.IATA) \\\n  .selectExpr(\"id\", \"city\", \"degreeRatio\") \\\n  .filter(\"degreeRatio < .9 or degreeRatio > 1.1\")\n\n# List out the city airports which have abnormal degree ratios.\ndisplay(nonTransferAirports)"],"metadata":{},"outputs":[],"execution_count":44},{"cell_type":"code","source":["# Join back to the `airports` DataFrame (instead of registering temp table as above)\ntransferAirports = degreeRatio.join(airports, degreeRatio.id == airports.IATA) \\\n  .selectExpr(\"id\", \"city\", \"degreeRatio\") \\\n  .filter(\"degreeRatio between 0.9 and 1.1\")\n  \n# List out the top 10 transfer city airports\ndisplay(transferAirports.orderBy(\"degreeRatio\").limit(10))"],"metadata":{},"outputs":[],"execution_count":45},{"cell_type":"markdown","source":["## Breadth First Search \nBreadth-first search (BFS) is designed to traverse the graph to quickly find the desired vertices (i.e. airports) and edges (i.e flights).  Let's try to find the shortest number of connections between cities based on the dataset.  Note, these examples do not take into account of time or distance, just hops between cities."],"metadata":{}},{"cell_type":"code","source":["# Example 1: Direct Seattle to San Francisco \nfilteredPaths = tripGraph.bfs(\n  fromExpr = \"id = 'SEA'\",\n  toExpr = \"id = 'SFO'\",\n  maxPathLength = 1)\ndisplay(filteredPaths)"],"metadata":{},"outputs":[],"execution_count":47},{"cell_type":"markdown","source":["As you can see, there are a number of direct flights between Seattle and San Francisco."],"metadata":{}},{"cell_type":"code","source":["# Example 2: Direct San Francisco and Buffalo\nfilteredPaths = tripGraph.bfs(\n  fromExpr = \"id = 'SFO'\",\n  toExpr = \"id = 'BUF'\",\n  maxPathLength = 2)\ndisplay(filteredPaths)"],"metadata":{},"outputs":[],"execution_count":49},{"cell_type":"code","source":["# Display most popular layover cities by descending count\ndisplay(filteredPaths.groupBy(\"v1.id\", \"v1.City\").count().orderBy(desc(\"count\")).limit(10))"],"metadata":{},"outputs":[],"execution_count":50},{"cell_type":"code","source":["# Example 2: Direct San Francisco and Buffalo\nfilteredPaths = tripGraph.bfs(\n  fromExpr = \"id = 'SFO'\",\n  toExpr = \"id = 'BUF'\",\n  maxPathLength = 1)\ndisplay(filteredPaths)"],"metadata":{},"outputs":[],"execution_count":51},{"cell_type":"markdown","source":["But there are no direct flights between San Francisco and Buffalo."],"metadata":{}},{"cell_type":"code","source":["# Example 2a: Flying from San Francisco to Buffalo\nfilteredPaths = tripGraph.bfs(\n  fromExpr = \"id = 'SFO'\",\n  toExpr = \"id = 'BUF'\",\n  maxPathLength = 2)\ndisplay(filteredPaths)"],"metadata":{},"outputs":[],"execution_count":53},{"cell_type":"markdown","source":["But there are flights from San Francisco to Buffalo with Minneapolis as the transfer point.  But what are the most popular layovers between `SFO` and `BUF`?"],"metadata":{}},{"cell_type":"code","source":["# Display most popular layover cities by descending count\ndisplay(filteredPaths.groupBy(\"v1.id\", \"v1.City\").count().orderBy(desc(\"count\")).limit(10))"],"metadata":{},"outputs":[],"execution_count":55},{"cell_type":"markdown","source":["## Loading the D3 Visualization\nUsing the airports D3 visualization to visualize airports and flight paths"],"metadata":{}},{"cell_type":"code","source":["%scala\npackage d3a\n// We use a package object so that we can define top level classes like Edge that need to be used in other cells\n\nimport org.apache.spark.sql._\nimport com.databricks.backend.daemon.driver.EnhancedRDDFunctions.displayHTML\n\ncase class Edge(src: String, dest: String, count: Long)\n\ncase class Node(name: String)\ncase class Link(source: Int, target: Int, value: Long)\ncase class Graph(nodes: Seq[Node], links: Seq[Link])\n\nobject graphs {\nval sqlContext = SQLContext.getOrCreate(org.apache.spark.SparkContext.getOrCreate())\nimport sqlContext.implicits._\n\ndef force(clicks: Dataset[Edge], height: Int = 100, width: Int = 960): Unit = {\n  val data = clicks.collect()\n  val nodes = (data.map(_.src) ++ data.map(_.dest)).map(_.replaceAll(\"_\", \" \")).toSet.toSeq.map(Node)\n  val links = data.map { t =>\n    Link(nodes.indexWhere(_.name == t.src.replaceAll(\"_\", \" \")), nodes.indexWhere(_.name == t.dest.replaceAll(\"_\", \" \")), t.count / 20 + 1)\n  }\n  showGraph(height, width, Seq(Graph(nodes, links)).toDF().toJSON.first())\n}\n\n/**\n * Displays a force directed graph using d3\n * input: {\"nodes\": [{\"name\": \"...\"}], \"links\": [{\"source\": 1, \"target\": 2, \"value\": 0}]}\n */\ndef showGraph(height: Int, width: Int, graph: String): Unit = {\n\ndisplayHTML(s\"\"\"<!DOCTYPE html>\n<html>\n  <head>\n    <link type=\"text/css\" rel=\"stylesheet\" href=\"https://mbostock.github.io/d3/talk/20111116/style.css\"/>\n    <style type=\"text/css\">\n      #states path {\n        fill: #ccc;\n        stroke: #fff;\n      }\n\n      path.arc {\n        pointer-events: none;\n        fill: none;\n        stroke: #000;\n        display: none;\n      }\n\n      path.cell {\n        fill: none;\n        pointer-events: all;\n      }\n\n      circle {\n        fill: steelblue;\n        fill-opacity: .8;\n        stroke: #fff;\n      }\n\n      #cells.voronoi path.cell {\n        stroke: brown;\n      }\n\n      #cells g:hover path.arc {\n        display: inherit;\n      }\n    </style>\n  </head>\n  <body>\n    <script src=\"https://mbostock.github.io/d3/talk/20111116/d3/d3.js\"></script>\n    <script src=\"https://mbostock.github.io/d3/talk/20111116/d3/d3.csv.js\"></script>\n    <script src=\"https://mbostock.github.io/d3/talk/20111116/d3/d3.geo.js\"></script>\n    <script src=\"https://mbostock.github.io/d3/talk/20111116/d3/d3.geom.js\"></script>\n    <script>\n      var graph = $graph;\n      var w = $width;\n      var h = $height;\n\n      var linksByOrigin = {};\n      var countByAirport = {};\n      var locationByAirport = {};\n      var positions = [];\n\n      var projection = d3.geo.azimuthal()\n          .mode(\"equidistant\")\n          .origin([-98, 38])\n          .scale(1400)\n          .translate([640, 360]);\n\n      var path = d3.geo.path()\n          .projection(projection);\n\n      var svg = d3.select(\"body\")\n          .insert(\"svg:svg\", \"h2\")\n          .attr(\"width\", w)\n          .attr(\"height\", h);\n\n      var states = svg.append(\"svg:g\")\n          .attr(\"id\", \"states\");\n\n      var circles = svg.append(\"svg:g\")\n          .attr(\"id\", \"circles\");\n\n      var cells = svg.append(\"svg:g\")\n          .attr(\"id\", \"cells\");\n\n      var arc = d3.geo.greatArc()\n          .source(function(d) { return locationByAirport[d.source]; })\n          .target(function(d) { return locationByAirport[d.target]; });\n\n      d3.select(\"input[type=checkbox]\").on(\"change\", function() {\n        cells.classed(\"voronoi\", this.checked);\n      });\n\n      // Draw US map.\n      d3.json(\"https://mbostock.github.io/d3/talk/20111116/us-states.json\", function(collection) {\n        states.selectAll(\"path\")\n          .data(collection.features)\n          .enter().append(\"svg:path\")\n          .attr(\"d\", path);\n      });\n\n      // Parse links\n      graph.links.forEach(function(link) {\n        var origin = graph.nodes[link.source].name;\n        var destination = graph.nodes[link.target].name;\n\n        var links = linksByOrigin[origin] || (linksByOrigin[origin] = []);\n        links.push({ source: origin, target: destination });\n\n        countByAirport[origin] = (countByAirport[origin] || 0) + 1;\n        countByAirport[destination] = (countByAirport[destination] || 0) + 1;\n      });\n\n      d3.csv(\"https://mbostock.github.io/d3/talk/20111116/airports.csv\", function(data) {\n\n        // Build list of airports.\n        var airports = graph.nodes.map(function(node) {\n          return data.find(function(airport) {\n            if (airport.iata === node.name) {\n              var location = [+airport.longitude, +airport.latitude];\n              locationByAirport[airport.iata] = location;\n              positions.push(projection(location));\n\n              return true;\n            } else {\n              return false;\n            }\n          });\n        });\n\n        // Compute the Voronoi diagram of airports' projected positions.\n        var polygons = d3.geom.voronoi(positions);\n\n        var g = cells.selectAll(\"g\")\n            .data(airports)\n          .enter().append(\"svg:g\");\n\n        g.append(\"svg:path\")\n            .attr(\"class\", \"cell\")\n            .attr(\"d\", function(d, i) { return \"M\" + polygons[i].join(\"L\") + \"Z\"; })\n            .on(\"mouseover\", function(d, i) { d3.select(\"h2 span\").text(d.name); });\n\n        g.selectAll(\"path.arc\")\n            .data(function(d) { return linksByOrigin[d.iata] || []; })\n          .enter().append(\"svg:path\")\n            .attr(\"class\", \"arc\")\n            .attr(\"d\", function(d) { return path(arc(d)); });\n\n        circles.selectAll(\"circle\")\n            .data(airports)\n            .enter().append(\"svg:circle\")\n            .attr(\"cx\", function(d, i) { return positions[i][0]; })\n            .attr(\"cy\", function(d, i) { return positions[i][1]; })\n            .attr(\"r\", function(d, i) { return Math.sqrt(countByAirport[d.iata]); })\n            .sort(function(a, b) { return countByAirport[b.iata] - countByAirport[a.iata]; });\n      });\n    </script>\n  </body>\n</html>\"\"\")\n  }\n\n  def help() = {\ndisplayHTML(\"\"\"\n<p>\nProduces a force-directed graph given a collection of edges of the following form:</br>\n<tt><font color=\"#a71d5d\">case class</font> <font color=\"#795da3\">Edge</font>(<font color=\"#ed6a43\">src</font>: <font color=\"#a71d5d\">String</font>, <font color=\"#ed6a43\">dest</font>: <font color=\"#a71d5d\">String</font>, <font color=\"#ed6a43\">count</font>: <font color=\"#a71d5d\">Long</font>)</tt>\n</p>\n<p>Usage:<br/>\n<tt>%scala</tt></br>\n<tt><font color=\"#a71d5d\">import</font> <font color=\"#ed6a43\">d3._</font></tt><br/>\n<tt><font color=\"#795da3\">graphs.force</font>(</br>\n&nbsp;&nbsp;<font color=\"#ed6a43\">height</font> = <font color=\"#795da3\">500</font>,<br/>\n&nbsp;&nbsp;<font color=\"#ed6a43\">width</font> = <font color=\"#795da3\">500</font>,<br/>\n&nbsp;&nbsp;<font color=\"#ed6a43\">clicks</font>: <font color=\"#795da3\">Dataset</font>[<font color=\"#795da3\">Edge</font>])</tt>\n</p>\"\"\")\n  }\n}"],"metadata":{},"outputs":[],"execution_count":57},{"cell_type":"code","source":["%scala d3a.graphs.help()"],"metadata":{},"outputs":[],"execution_count":58},{"cell_type":"markdown","source":["#### Visualize On-time and Early Arrivals"],"metadata":{}},{"cell_type":"code","source":["%scala\n// On-time and Early Arrivals\nimport d3a._\ngraphs.force(\n  height = 800,\n  width = 1200,\n  clicks = sql(\"\"\"select src, dst as dest, count(1) as count from departureDelays_geo where delay <= 0 group by src, dst\"\"\").as[Edge])"],"metadata":{},"outputs":[],"execution_count":60},{"cell_type":"markdown","source":["#### Visualize Delayed Trips Departing from the West Coast\n\nNotice that most of the delayed trips are with Western US cities"],"metadata":{}},{"cell_type":"code","source":["%scala\n// Delayed Trips from CA, OR, and/or WA\nimport d3a._\ngraphs.force(\n  height = 800,\n  width = 1200,\n  clicks = sql(\"\"\"select src, dst as dest, count(1) as count from departureDelays_geo where state_src in ('CA', 'OR', 'WA') and delay > 0 group by src, dst\"\"\").as[Edge])"],"metadata":{},"outputs":[],"execution_count":62},{"cell_type":"markdown","source":["#### Visualize All Flights (from this dataset)"],"metadata":{}},{"cell_type":"code","source":["%scala\n// Trips (from DepartureDelays Dataset)\nimport d3a._\ngraphs.force(\n  height = 800,\n  width = 1200,\n  clicks = sql(\"\"\"select src, dst as dest, count(1) as count from departureDelays_geo group by src, dst\"\"\").as[Edge])"],"metadata":{},"outputs":[],"execution_count":64},{"cell_type":"markdown","source":["## Predicting Flight Delays\nExtending upon analysis we have done up to this point, can we also predict if a flight will be delayed, on-time, or early based on the available data.\n\n### Prepare the Dataset\nThe first thing we will do is to cleanse the data and apply some labels to our information (e.g. early, on-time, delayed).  As well, we will want to remove any rows with NULL values."],"metadata":{}},{"cell_type":"code","source":["# This contains a generated mapping between tripid and airline\n#   You can get the file at https://github.com/dennyglee/databricks/blob/master/misc/trip_airline_map.csv\n#   For this example, the trip_airline_map.csv file has been pushed to in my mounted bucket.\ntripAirlineMap = spark.read.csv(\"/mnt/tardis6/departuredelays/trip_airline_map.csv\", sep=\",\", header=True)\ntripAirlineMap.createOrReplaceTempView(\"tripAirlineMap\")"],"metadata":{},"outputs":[],"execution_count":66},{"cell_type":"code","source":["# Prep dataset\n# Limiting to only Las Vegas (LAS) and Seattle (SEA) predictions so it can run on Databricks Community Edition\nflightML = spark.sql(\"select cast(distance as double) as distance, src as origin, state_src as origin_state, dst as destination, state_dst as destination_state, concat(concat(concat(cast(tripid as string), src), dst), cast((delay + 2000) as string)) as trip_identifier, case when delay <= 0 then 'on-time' else 'delayed' end as flight_status from departureDelays_geo where src in ('LAS', 'SEA')\")\nflightML = flightML.dropna().dropDuplicates()\nflightML.createOrReplaceTempView(\"flightML\")"],"metadata":{},"outputs":[],"execution_count":67},{"cell_type":"code","source":["# Join flights and airline information\ndataset = spark.sql(\"select f.distance, f.origin, f.origin_state, f.destination, f.destination_state, f.trip_identifier, f.flight_status, m.airline from flightML f join tripAirlineMap m on m.trip_identifier = f.trip_identifier\")\ndataset = dataset.dropDuplicates()\n#dataset = flightML\ncols = dataset.columns"],"metadata":{},"outputs":[],"execution_count":68},{"cell_type":"code","source":["dataset.printSchema()"],"metadata":{},"outputs":[],"execution_count":69},{"cell_type":"code","source":["dataset.count()"],"metadata":{},"outputs":[],"execution_count":70},{"cell_type":"markdown","source":["### Building ML Pipeline\nBefore we can run our various models against this data, we will first need to vectorize our data via One-Hot Encorder (for category data), String Indexer (create an index based on our labelled values), and Vector Assembler."],"metadata":{}},{"cell_type":"code","source":["# One-Hot Encoding\nfrom pyspark.ml import Pipeline\nfrom pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler\n\ncategoricalColumns = [\"origin\", \"origin_state\", \"destination\", \"destination_state\", \"trip_identifier\", \"airline\"]\n#categoricalColumns = [\"origin\", \"origin_state\", \"destination\", \"destination_state\", \"trip_identifier\"]\nstages = [] # stages in our Pipeline\nfor categoricalCol in categoricalColumns:\n  # Category Indexing with StringIndexer\n  stringIndexer = StringIndexer(inputCol=categoricalCol, outputCol=categoricalCol+\"Index\")\n  \n  # Use OneHotEncoder to convert categorical variables into binary SparseVectors\n  encoder = OneHotEncoder(inputCol=categoricalCol+\"Index\", outputCol=categoricalCol+\"classVec\")\n  \n  # Add stages.  These are not run here, but will run all at once later on.\n  stages += [stringIndexer, encoder]\n\n# Convert label into label indices using the StringIndexer\nlabel_stringIdx = StringIndexer(inputCol = \"flight_status\", outputCol = \"label\")\nstages += [label_stringIdx]\n\n# Transform all features into a vector using VectorAssembler\nnumericCols = [\"distance\"]\nassemblerInputs = map(lambda c: c + \"classVec\", categoricalColumns) + numericCols\nassembler = VectorAssembler(inputCols=assemblerInputs, outputCol=\"features\")\nstages += [assembler]"],"metadata":{},"outputs":[],"execution_count":72},{"cell_type":"code","source":["# Create a Pipeline.\npipeline = Pipeline(stages=stages)\n# Run the feature transformations.\n#  - fit() computes feature statistics as needed.\n#  - transform() actually transforms the features.\npipelineModel = pipeline.fit(dataset)\ndataset = pipelineModel.transform(dataset)\n\n# Keep relevant columns\nselectedcols = [\"label\", \"features\"] + cols\ndataset = dataset.select(selectedcols)\ndisplay(dataset)"],"metadata":{},"outputs":[],"execution_count":73},{"cell_type":"markdown","source":["### Randomly split data into training and test datasets\n* Set the seed for reproducibility"],"metadata":{}},{"cell_type":"code","source":["(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed = 100)\nprint trainingData.count()\nprint testData.count()"],"metadata":{},"outputs":[],"execution_count":75},{"cell_type":"markdown","source":["## Logistic Regression\nLet's try using logistic regression to see if we can accurately predict if a flight will be delayed.\n* First, we will train the data using Logistic Regression\n* Next we will run that model against the testData"],"metadata":{}},{"cell_type":"code","source":["from pyspark.ml.classification import LogisticRegression\n\n# Create initial LogisticRegression model\nlr = LogisticRegression(labelCol=\"label\", featuresCol=\"features\", maxIter=10)\n\n# Train model with Training Data\nlrModel = lr.fit(trainingData)"],"metadata":{},"outputs":[],"execution_count":77},{"cell_type":"code","source":["# Make predictions on test data using the transform() method.\n# LogisticRegression.transform() will only use the 'features' column.\npredictions = lrModel.transform(testData)"],"metadata":{},"outputs":[],"execution_count":78},{"cell_type":"markdown","source":["### View LR Model's predictions\n* Recall, label is the actual test value, prediction is the predicted value\n * where 0 - on-time, 1 - delayed"],"metadata":{}},{"cell_type":"code","source":["selected = predictions.select(\"label\", \"prediction\", \"probability\", \"flight_status\", \"destination\", \"destination_state\").where(\"destination = 'SEA'\")\ndisplay(selected)"],"metadata":{},"outputs":[],"execution_count":80},{"cell_type":"markdown","source":["#### Evaluate our model\nLet's use the `BinaryClassificationEvaluator` to determine the precision of our model."],"metadata":{}},{"cell_type":"code","source":["from pyspark.ml.evaluation import BinaryClassificationEvaluator\n\n# Evaluate model\nevaluator = BinaryClassificationEvaluator(rawPredictionCol=\"rawPrediction\")\nevaluator.evaluate(predictions)"],"metadata":{},"outputs":[],"execution_count":82},{"cell_type":"code","source":[""],"metadata":{},"outputs":[],"execution_count":83}],"metadata":{"name":"On-Time Flight Performance (Spark 2.0)","notebookId":969073022158310},"nbformat":4,"nbformat_minor":0}