Spark项目模拟——航班飞行网图分析

需求描述

数据准备

数据文件一份，如下：

文件格式为CSV，字段之间分隔符为“,”

列依次为：#日、周#、航空公司、飞机注册号、航班号、起飞机场编号、起飞机场、到达机场编号、到达机场、预计起飞时间（时分）、起飞时间、起飞延迟（分钟）、到达预计时间、到达时间、到达延迟（分钟）、预计飞行时间、飞行距离

需求思路整理

创建属性图Graph[VD,ED]

1、装载CSV为RDD，每个机场作为顶点。关键字段：起飞机场编号、起飞机场、到达机场编号、到达机场、飞行距离

2、初始化顶点集airports:RDD[(VertexId,String)]，顶点属性为机场名称

3、初始化边集lines:RDD[Edge]，边属性为飞行距离

代码实现

import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.types._
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SparkSession, types}
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.StructTypeobject FlightDemo {def main(args: Array[String]): Unit = {val conf: SparkConf = new SparkConf().setAppName("etldemo").setMaster("local[*]")val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()val sc: SparkContext = spark.sparkContextimport spark.implicits._
//加载文件val flightRDD: RDD[String] = sc.textFile("in/flight.csv")
//    flightRDD.collect().foreach(println)val airPort: RDD[(VertexId, String)] = flightRDD.map(x => x.split(",")).flatMap(x => Array((x(5).toLong, x(6)), (x(7).toLong, x(8)))).distinct()
//    airPort.collect().foreach(println)
//    println(airPort.count())    //机场数量val lines: RDD[Edge[PartitionID]] = flightRDD.map(x => x.split(",")) //（起飞机场编号，降落机场编号，飞行距离）.map(x => (x(5).toLong, x(7).toLong, x(16).toInt)).distinct().map(x => Edge(x._1, x._2, x._3))
//    lines.collect().foreach(println)
//    println(lines.count())val graph = Graph(airPort,lines)
//    graph.triplets.collect().foreach(println)//机场数量val numvertices: VertexId = graph.numVerticesprintln("机场数量"+numvertices)//航线数量val numedges: VertexId = graph.numEdgesprintln("航线数量"+numedges)//计算最长的飞行航线val rdd1: RDD[EdgeTriplet[String, PartitionID]] = graph.triplets.sortBy(x=>x.attr,false)val str: Array[String] = rdd1.map(triplet => triplet.srcAttr + "   " + triplet.dstAttr + "  距离  ：" + triplet.attr).take(3)
//    str.foreach(println)//找出降落繁忙的机场val indegrees: VertexRDD[PartitionID] = graph.inDegreesval busyAirPort: (VertexId, PartitionID) = indegrees.sortBy(x=>x._2,false).take(1)(0)
//    println(busyAirPort)//找出起飞繁忙的机场val outdegrees: VertexRDD[PartitionID] = graph.outDegreesval busyinAirPort: (VertexId, PartitionID) = outdegrees.sortBy(x=>x._2,false).take(1)(0)
//    println(busyinAirPort)//找出最重要的飞行航线val vertices: VertexRDD[Double] = graph.pageRank(0.05).vertices
//    vertices.collect().foreach(println)
//    vertices.sortBy(x=>x._2).collect().foreach(println)//找出最便宜的飞行航线val value: RDD[(VertexId, String)] = airPort.sample(false,1.0/airPort.count(),1)val source_id=13930.toLongval srcAirPortName =value.first()._2val srcAirPortId= value.first()._1println(source_id,srcAirPortId,srcAirPortName)val init_graph: Graph[Double, Double] = graph.mapVertices((id, value) => {if (id == srcAirPortId) 0else Double.PositiveInfinity}).mapEdges(e=> 180 + 0.15*e.attr.toDouble)val pregel_graph: Graph[Double, Double] = init_graph.pregel(Double.PositiveInfinity,Int.MaxValue,EdgeDirection.Out)((id, dist, new_dist) => math.min(dist, new_dist),triple => {if (triple.attr + triple.srcAttr < triple.dstAttr)Iterator((triple.dstId, triple.attr + triple.srcAttr))elseIterator.empty},(dist, new_dist) => math.min(dist, new_dist))pregel_graph.triplets.collect().foreach(println)//((16218,906.15),(12892,748.05),215.55)// 初始机场13296  到 16218  最低费用为 906.15//初始机场 13296   到12892   最低费用为748.05//机场16218  到12892  最低费用为 215.55//得到 初始机场到 另一个机场的 最少费用
//    pregel_graph.vertices.collect().foreach(println)//最便宜的机票
//    val tuples: Array[(VertexId, Double)] = pregel_graph.vertices.sortBy(x=>x._2).take(3)
//    println(tuples.toList)//List((13296,0.0), (12953,209.25), (11618,211.35))//13296 为初始机场，所以费用为0}
}

总结

实现需求过程中使用了以下函数：

Spark GraphX APIvertices、edges、triplets、numEdges、numVerticesinDegrees、outDegrees、degreesmapVertices、mapEdges、mapTripletsSpark GraphX PageRankSpark GraphX Pregel

机场数量：

求顶点个数：Graph.numVertices

航线数量：

求边的个数：Graph.numEdges

最大的边属性：

对triplets按飞行距离排序（降序）并取第一个

哪个机场到达航班最多：

计算顶点的入度并排序

找出最重要的飞行航线：

PageRank
收敛误差：0.05

找出最便宜的飞行航线：

定价模型：
price = 180.0 + distance * 0.15

SSSP问题：
从初始指定的源点到达任意点的最短距离

pregel：
1、初始化源点（0）与其它顶点（Double.PositiveInfinity）
2、初始消息（Double.PositiveInfinity）
3、vprog函数计算最小值
4、sendMsg函数计算进行是否下一个迭代
5、mergeMsg函数合并接受的消息，取最小值