【Spark GraphX】航班数据网图分析

朱雀 2022-09-11 00:14 214阅读 0赞

**目录**

一、数据格式

二、需求分析

三、代码分析

--------------------

# 一、数据格式 #

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6YOd5bCR_size_20_color_FFFFFF_t_70_g_se_x_16][]

1.   从左到右依次为：日、周、航空公司、飞机注册号、航班号、起飞机场编号、起飞机场、到达机场编号、到达机场、预计起飞时间（时分）、起飞时间、起飞延迟（分钟）、到达预计时间、到达时间、到达延迟（分钟）、预计飞行时间、飞行距离；
2.  数据格式：csv格式、逗号分隔；

# 二、需求分析 #

1.  加载数据及网图构建
2.  统计机场数量
3.  统计航线数
4.  计算最长的飞行航线
5.  找出最繁忙的机场
6.  找出最重要的机场
7.  找出最便宜的飞机航线

# 三、代码分析 #

package com.dt.spark.graphx.test
    
    import org.apache.log4j.{Level, Logger}
    import org.apache.spark.SparkContext
    import org.apache.spark.graphx.{Edge, Graph, VertexId}
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.SparkSession
    
    object AirportLineAnalyse {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.ERROR)
        val spark: SparkSession = SparkSession.builder().master("local[*]").appName("airport Line").getOrCreate()
        val sc: SparkContext = spark.sparkContext
    
        /**
          * 加载数据及网图构建
          */
        //加载数据
        val flights: RDD[Array[String]] = sc.textFile("data/airport/flight.csv").map(s => s.split(","))
        //机场数据
        val airports: RDD[(Long, String)] = flights.flatMap(s => Array((s(5).toLong, s(6)), (s(7).toLong, s(8)))).distinct()
        //航线数据
        val lines: RDD[Edge[Int]] = flights.map(x => (x(5).toLong, x(7).toLong, x(16).toInt)).distinct().map(s => Edge(s._1, s._2, s._3))
        //构建网图
        val graph: Graph[String, Int] = Graph(airports, lines)
    
        /**
          * 统计机场数量
          */
        println("机场数量为:%d".format(graph.vertices.count()))
        println("机场数量为:%d".format(graph.numVertices))
    
        /**
          * 统计航线数
          */
        println("航线数为:%s".format(graph.numEdges))
    
        /**
          * 计算最长的飞行航线
          */
        println("最长的航线为:%s".format(graph.triplets.sortBy(e => e.attr, false).take(1).mkString(",")))
    
        /**
          * 找出最繁忙的机场
          */
        println("最繁忙的机场:%s".format(graph.inDegrees.sortBy(_._2, false).take(1).mkString(",")))
        println("最繁忙的机场:%s".format(graph.inDegrees.reduce((a, b) => if (a._2 > b._2) a else b)))
    
        /**
          * 找出最重要的机场
          */
        val pg: Graph[Double, Double] = graph.pageRank(0.05)
        pg.vertices.take(3).foreach(s => println(s)) //(10268,0.18770202755192045)
        val tuple: (VertexId, Double) = pg.vertices.reduce((a, b) => if (a._2 > b._2) a else b)
        println("最重要的机场为:%s".format(tuple._1))
    
        //另一种思路,找顶点入度最多的也是最重要的机场,结果也和上面吻合;
        graph.inDegrees.sortBy(_._2, false).collect().take(3).foreach(v => println(v))
    
        /**
          * 找出最便宜的飞机航线
          * SSSP问题:从初始指定的源点到达任意点的最短距离;
          * 定价模型: price = 180.0 + distance * 0.15
          */
        val count = airports.count //总机场数
        var fraction = 1.0 //系数
        var samples = airports.sample(false, fraction / count, count) //抽样
        while (samples.count < 0) {
          fraction = fraction + 1
          samples = airports.sample(false, fraction / count, count)
        }
        val source_id: VertexId = samples.first._1 //得到源点 14952
        //给初始值,转换定价模型
        val init_graph = graph.mapVertices((id, _) => if (id == source_id) 0.0 else Double.PositiveInfinity)
          .mapEdges(e => 180.toDouble + e.attr.toDouble * 0.15)
        //构建pregel模型
        val pregel_graph = init_graph.pregel(Double.PositiveInfinity)(
          (id, dist, new_dist) => math.min(dist, new_dist),
          triplet => {
            if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
              Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
            }
            else Iterator.empty
          },
          (a, b) => math.min(a, b)
        )
        //航线
        val cheap_lines = pregel_graph.edges.map { case (Edge(src_id, dst_id, price))
        => (src_id, dst_id, price)
        }.takeOrdered(3)(Ordering.by(_._3))
          .foreach(v => println("便宜航线推荐:" + v))
    
        //pregel_graph.vertices.foreach(v=>println(v))
        //机场
        val cheap_airports = pregel_graph.vertices.takeOrdered(3)(Ordering.by(_._2))
          .foreach(v => println("便宜机场:" + v))
      }
    }

机场数量为:301
    机场数量为:301
    航线数为:4088
    最长的航线为:((12173,HNL),(12478,JFK),4983)
    最繁忙的机场:(10397,152)
    最繁忙的机场:(10397,152)
    (10208,0.18681782122279475)
    (10268,0.18770202755192045)
    (14828,0.35999854863670966)
    最重要的机场为:10397
    (10397,152)
    (13930,145)
    (11298,143)
    便宜航线推荐:(15841,14256,184.65)
    便宜航线推荐:(14256,15841,184.65)
    便宜航线推荐:(10157,10930,188.4)
    便宜机场:(14952,0.0)
    便宜机场:(13930,206.1)
    便宜机场:(11298,274.5)

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6YOd5bCR_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/03479673e93a4750b2956aa5ea55fa3a.png