SDS-2.2, Scalable Data Science

Archived YouTube video of this live unedited lab-lecture:

This is a scala version of the python notebook in the following talk:

Homework:

See https://www.brighttalk.com/webcast/12891/199003 (you need to subscribe freely to Bright Talk first). Then go through this scala version of the notebook from the talk.

On-Time Flight Performance with GraphFrames for Apache Spark

This notebook provides an analysis of On-Time Flight Performance and Departure Delays data using GraphFrames for Apache Spark.

Source Data:

• OpenFlights: Airport, airline and route data
• United States Department of Transportation: Bureau of Transportation Statistics (TranStats)
• Note, the data used here was extracted from the US DOT:BTS between 1/1/2014 and 3/31/2014*

References:

• GraphFrames User Guide
• GraphFrames: DataFrame-based Graphs (GitHub)
• D3 Airports Example

Preparation

Extract the Airports and Departure Delays information from S3 / DBFS

// Set File Paths
val tripdelaysFilePath = "/databricks-datasets/flights/departuredelays.csv"
val airportsnaFilePath = "/databricks-datasets/flights/airport-codes-na.txt"

tripdelaysFilePath: String = /databricks-datasets/flights/departuredelays.csv
airportsnaFilePath: String = /databricks-datasets/flights/airport-codes-na.txt

// Obtain airports dataset
// Note that "spark-csv" package is built-in datasource in Spark 2.0
val airportsna = sqlContext.read.format("com.databricks.spark.csv").
  option("header", "true").
  option("inferschema", "true").
  option("delimiter", "\t").
  load(airportsnaFilePath)

airportsna.createOrReplaceTempView("airports_na")

// Obtain departure Delays data
val departureDelays = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load(tripdelaysFilePath)
departureDelays.createOrReplaceTempView("departureDelays")
departureDelays.cache()

// Available IATA (International Air Transport Association) codes from the departuredelays sample dataset
val tripIATA = sqlContext.sql("select distinct iata from (select distinct origin as iata from departureDelays union all select distinct destination as iata from departureDelays) a")
tripIATA.createOrReplaceTempView("tripIATA")

// Only include airports with atleast one trip from the departureDelays dataset
val airports = sqlContext.sql("select f.IATA, f.City, f.State, f.Country from airports_na f join tripIATA t on t.IATA = f.IATA")
airports.createOrReplaceTempView("airports")
airports.cache()

airportsna: org.apache.spark.sql.DataFrame = [City: string, State: string ... 2 more fields]
departureDelays: org.apache.spark.sql.DataFrame = [date: string, delay: string ... 3 more fields]
tripIATA: org.apache.spark.sql.DataFrame = [iata: string]
airports: org.apache.spark.sql.DataFrame = [IATA: string, City: string ... 2 more fields]
res0: airports.type = [IATA: string, City: string ... 2 more fields]

// Build `departureDelays_geo` DataFrame
// Obtain key attributes such as Date of flight, delays, distance, and airport information (Origin, Destination)  
val departureDelays_geo = sqlContext.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") 

// RegisterTempTable
departureDelays_geo.createOrReplaceTempView("departureDelays_geo")

// Cache and Count
departureDelays_geo.cache()
departureDelays_geo.count()

departureDelays_geo: org.apache.spark.sql.DataFrame = [tripid: int, localdate: timestamp ... 8 more fields]
res3: Long = 1361141

display(departureDelays_geo)

Truncated to 30 rows

Building the Graph

Now 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.

• Rename IATA airport code to id in the Vertices Table
• Start and End airports to src and dst for the Edges Table (flights)

These are required naming conventions for vertices and edges in GraphFrames as of the time of this writing (Feb. 2016).

WARNING: If the graphframes package, required in the cell below, is not installed, follow the instructions here.

// Note, ensure you have already installed the GraphFrames spack-package
import org.apache.spark.sql.functions._
import org.graphframes._

// Create Vertices (airports) and Edges (flights)
val tripVertices = airports.withColumnRenamed("IATA", "id").distinct()
val tripEdges = departureDelays_geo.select("tripid", "delay", "src", "dst", "city_dst", "state_dst")

// Cache Vertices and Edges
tripEdges.cache()
tripVertices.cache()

import org.apache.spark.sql.functions._
import org.graphframes._
tripVertices: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: string, City: string ... 2 more fields]
tripEdges: org.apache.spark.sql.DataFrame = [tripid: int, delay: int ... 4 more fields]
res5: tripVertices.type = [id: string, City: string ... 2 more fields]

// Vertices
// The vertices of our graph are the airports
display(tripVertices)

Truncated to 30 rows

// Edges
// The edges of our graph are the flights between airports
display(tripEdges)

Truncated to 30 rows

// Build `tripGraph` GraphFrame
// This GraphFrame builds up on the vertices and edges based on our trips (flights)
val tripGraph = GraphFrame(tripVertices, tripEdges)
println(tripGraph)

// Build `tripGraphPrime` GraphFrame
// This graphframe contains a smaller subset of data to make it easier to display motifs and subgraphs (below)
val tripEdgesPrime = departureDelays_geo.select("tripid", "delay", "src", "dst")
val tripGraphPrime = GraphFrame(tripVertices, tripEdgesPrime)

GraphFrame(v:[id: string, City: string ... 2 more fields], e:[src: string, dst: string ... 4 more fields])
tripGraph: org.graphframes.GraphFrame = GraphFrame(v:[id: string, City: string ... 2 more fields], e:[src: string, dst: string ... 4 more fields])
tripEdgesPrime: org.apache.spark.sql.DataFrame = [tripid: int, delay: int ... 2 more fields]
tripGraphPrime: org.graphframes.GraphFrame = GraphFrame(v:[id: string, City: string ... 2 more fields], e:[src: string, dst: string ... 2 more fields])

Simple Queries

Let's start with a set of simple graph queries to understand flight performance and departure delays

Determine the number of airports and trips

println(s"Airports: ${tripGraph.vertices.count()}")
println(s"Trips: ${tripGraph.edges.count()}")

Airports: 279
Trips: 1361141

Determining the longest delay in this dataset

// Finding the longest Delay
val longestDelay = tripGraph.edges.groupBy().max("delay")
display(longestDelay)

Determining the number of delayed vs. on-time / early flights

// Determining number of on-time / early flights vs. delayed flights
println(s"On-time / Early Flights: ${tripGraph.edges.filter("delay <= 0").count()}")
println(s"Delayed Flights: ${tripGraph.edges.filter("delay > 0").count()}")

On-time / Early Flights: 780469
Delayed Flights: 580672

What flights departing SFO are most likely to have significant delays

Note, delay can be <= 0 meaning the flight left on time or early

val sfoDelayedTrips = tripGraph.edges.
  filter("src = 'SFO' and delay > 0").
  groupBy("src", "dst").
  avg("delay").
  sort(desc("avg(delay)"))

sfoDelayedTrips: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [src: string, dst: string ... 1 more field]

display(sfoDelayedTrips)

Truncated to 30 rows

What destinations tend to have delays

// After displaying tripDelays, use Plot Options to set `state_dst` as a Key.
val tripDelays = tripGraph.edges.filter($"delay" > 0)
display(tripDelays)

What destinations tend to have significant delays departing from SEA

// States with the longest cumulative delays (with individual delays > 100 minutes) (origin: Seattle)
display(tripGraph.edges.filter($"src" === "SEA" && $"delay" > 100))

Vertex Degrees

• inDegrees: Incoming connections to the airport
• outDegrees: Outgoing connections from the airport
• degrees: Total connections to and from the airport

Reviewing the various properties of the property graph to understand the incoming and outgoing connections between airports.

// Degrees
// The number of degrees - the number of incoming and outgoing connections - for various airports within this sample dataset
display(tripGraph.degrees.sort($"degree".desc).limit(20))

City / Flight Relationships through Motif Finding

To 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.

What delays might we blame on SFO

/*
Using tripGraphPrime to more easily display 
- The associated edge (ab, bc) relationships 
- With the different the city / airports (a, b, c) where SFO is the connecting city (b)
- Ensuring that flight ab (i.e. the flight to SFO) occured before flight bc (i.e. flight leaving SFO)
- Note, TripID was generated based on time in the format of MMDDHHMM converted to int
- Therefore bc.tripid < ab.tripid + 10000 means the second flight (bc) occured within approx a day of the first flight (ab)
Note: In reality, we would need to be more careful to link trips ab and bc.
*/
val motifs = tripGraphPrime.
  find("(a)-[ab]->(b); (b)-[bc]->(c)").
  filter("(b.id = 'SFO') and (ab.delay > 500 or bc.delay > 500) and bc.tripid > ab.tripid and bc.tripid < ab.tripid + 10000")

display(motifs)

Determining Airport Ranking using PageRank

There 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.

// Determining Airport ranking of importance using `pageRank`
val ranks = tripGraph.pageRank.resetProbability(0.15).maxIter(5).run()

ranks: org.graphframes.GraphFrame = GraphFrame(v:[id: string, City: string ... 3 more fields], e:[src: string, dst: string ... 5 more fields])

display(ranks.vertices.orderBy($"pagerank".desc).limit(20))

BTW, A lot more delicate air-traffic arithmetic is possible for a full month of airplane co-trajectories over the radar range of Atlanta, Georgia!

See for instance:

• Statistical regular pavings to analyze massive data of aircraft trajectories, Gloria Teng, Kenneth Kuhn and Raazesh Sainudiin, Journal of Aerospace Computing, Information, and Communication, Vol. 9, No. 1, pp. 14-25, doi: 10.2514/1.I010015, 2012. See free preprint: http://lamastex.org/preprints/AAIASubPavingATC.pdf.

Most popular flights (single city hops)

Using the tripGraph, we can quickly determine what are the most popular single city hop flights

// Determine the most popular flights (single city hops)
import org.apache.spark.sql.functions._

val topTrips = tripGraph.edges.
  groupBy("src", "dst").
  agg(count("delay").as("trips"))

import org.apache.spark.sql.functions._
topTrips: org.apache.spark.sql.DataFrame = [src: string, dst: string ... 1 more field]

// Show the top 20 most popular flights (single city hops)
display(topTrips.orderBy($"trips".desc).limit(20))

Top Transfer Cities

Many 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.

// Calculate the inDeg (flights into the airport) and outDeg (flights leaving the airport)
val inDeg = tripGraph.inDegrees
val outDeg = tripGraph.outDegrees

// Calculate the degreeRatio (inDeg/outDeg), perform inner join on "id" column
val degreeRatio = inDeg.join(outDeg, inDeg("id") === outDeg("id")).
  drop(outDeg("id")).
  selectExpr("id", "double(inDegree)/double(outDegree) as degreeRatio").
  cache()

// Join back to the `airports` DataFrame (instead of registering temp table as above)
val nonTransferAirports = degreeRatio.join(airports, degreeRatio("id") === airports("IATA")).
  selectExpr("id", "city", "degreeRatio").
  filter("degreeRatio < 0.9 or degreeRatio > 1.1")

// List out the city airports which have abnormal degree ratios
display(nonTransferAirports)

// Join back to the `airports` DataFrame (instead of registering temp table as above)
val transferAirports = degreeRatio.join(airports, degreeRatio("id") === airports("IATA")).
  selectExpr("id", "city", "degreeRatio").
  filter("degreeRatio between 0.9 and 1.1")

// List out the top 10 transfer city airports
display(transferAirports.orderBy("degreeRatio").limit(10))

Breadth First Search

Breadth-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.

// Example 1: Direct Seattle to San Francisco
// This method returns a DataFrame of valid shortest paths from vertices matching "fromExpr" to vertices matching "toExpr"
val filteredPaths = tripGraph.bfs.fromExpr((col("id") === "SEA")).toExpr(col("id") === "SFO").maxPathLength(1).run()
display(filteredPaths)

As you can see, there are a number of direct flights between Seattle and San Francisco.

// Example 2: Direct San Francisco and Buffalo
// You can also specify expression as a String, instead of Column
val filteredPaths = tripGraph.bfs.fromExpr("id = 'SFO'").toExpr("id = 'BUF'").maxPathLength(1).run()

filteredPaths: org.apache.spark.sql.DataFrame = [id: string, City: string ... 2 more fields]

filteredPaths.show()

+---+----+-----+-------+
| id|City|State|Country|
+---+----+-----+-------+
+---+----+-----+-------+

display(filteredPaths)

But there are no direct flights between San Francisco and Buffalo.

// Example 2a: Flying from San Francisco to Buffalo
val filteredPaths = tripGraph.bfs.fromExpr("id = 'SFO'").toExpr("id = 'BUF'").maxPathLength(2).run()
display(filteredPaths)

But there are flights from San Francisco to Buffalo with Minneapolis as the transfer point.

Loading the D3 Visualization

Using the airports D3 visualization to visualize airports and flight paths

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

d3a.graphs.help()

Produces a force-directed graph given a collection of edges of the following form:
case class Edge(src: String, dest: String, count: Long)

Usage:
%scala
import d3._
graphs.force(
  height = 500,
  width = 500,
  clicks: Dataset[Edge])

Visualize On-time and Early Arrivals

// On-time and Early Arrivals
import d3a._

graphs.force(
  height = 800,
  width = 1200,
  clicks = sql("select src, dst as dest, count(1) as count from departureDelays_geo where delay <= 0 group by src, dst").as[Edge])

Visualize Delayed Trips Departing from the West Coast

Notice that most of the delayed trips are with Western US cities

// Delayed Trips from CA, OR, and/or WA
import d3a._

graphs.force(
  height = 800,
  width = 1200,
  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])

Visualize All Flights (from this dataset)

// Trips (from DepartureDelays Dataset)
import d3a._

graphs.force(
  height = 800,
  width = 1200,
  clicks = sql("""select src, dst as dest, count(1) as count from departureDelays_geo group by src, dst""").as[Edge])