图解大数据 | Spark GraphFrames-基于图的数据分析挖掘

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

# This GraphFrame builds upon the vertices and edges based on our trips (flights)
tripGraph=GraphFrame(tripVertices, tripEdges)

# 查询机场个数和行程个数（查询节点和边的个数）
print("Airports:", tripGraph.vertices.count())
print("Trips:", tripGraph.edges.count())

# 查询最长延误时间（通过分组统计完成）
longestDelay = tripGraph.edges.groupby().max("delay")

# 晚点与准点航班分析（通过数据选择与过滤，进行边的分析）
print "On-time / Early Flights: %d" % tripGraph.edges.filter("delay <= 0").count()
print "Delayed Flights: %d" % tripGraph.edges.filter("delay > 0").count()

# 从旧金山出发的飞机中延迟最严重的航班（数据选择+边分析+分组统计）
tripGraph.edges.filter(“src = ‘SFO’ and delay > 0”).groupBy(“src”, “dst”).avg(“delay”).sort(desc(“avg(delay)”))

display(tripGraph.degrees.sort(desc("degree")).limit(20))

import pyspark.sql.functions as func 
topTrips = tripGraph.edges.groupBy("src", "dst").agg(func.count("delay").alias("trips"))

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

# Calculate the degreeRatio (inDeg/outDeg) 
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) 
nonTransferAirports = degreeRatio.join(airports, degreeRatio.id == airports.IATA) \ 
    .selectExpr("id", "city", "degreeRatio").filter("degreeRatio < .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) 
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))

# Example 1: Direct Seattle to San Francisco 
filteredPaths = tripGraph.bfs(fromExpr = "id = 'SEA'", toExpr = "id = 'SFO'", maxPathLength = 1) 
display(filteredPaths) 

# Example 2: Direct San Francisco and Buffalo 
filteredPaths = tripGraph.bfs(fromExpr = "id = 'SFO'", toExpr = "id = 'BUF'", maxPathLength = 1) 
display(filteredPaths) 

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

# Determining Airport ranking of importance using `pageRank` 
ranks = tripGraph.pageRank(resetProbability=0.15, maxIter=5) 
display(ranks.vertices.orderBy(ranks.vertices.pagerank.desc()).limit(20))